c#实现基于事件驱动的自动流程控制,包含任务超时,任务执行metric
实现事件驱动模型
这个类实现了事件分发,和队列任务执行,以及成功回调,失败回调和超时处理
using System;
using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Tasks;
public class EventDrivenTaskQueue : IDisposable
{
private readonly ConcurrentDictionary<string, Func<object, CancellationToken, Task<object>>> _tasks = new ConcurrentDictionary<string, Func<object, CancellationToken, Task<object>>>();
private readonly ConcurrentDictionary<string, ManualResetEventSlim> _events = new ConcurrentDictionary<string, ManualResetEventSlim>();
private readonly CancellationTokenSource _cancellationTokenSource = new CancellationTokenSource();
private readonly Thread _workerThread;
private readonly ConcurrentDictionary<string, object> _taskResults = new ConcurrentDictionary<string, object>();
private readonly ConcurrentDictionary<string, TimeSpan> _taskTimeouts = new ConcurrentDictionary<string, TimeSpan>();
private readonly ConcurrentQueue<string> _taskQueue = new ConcurrentQueue<string>();
public event Action<string> TaskStarted;
public event Action<string> TaskCompleted;
public event Action<string, string> TaskFailed;
public EventDrivenTaskQueue()
{
_workerThread = new Thread(ProcessQueue)
{
IsBackground = true
};
_workerThread.Start();
}
public void RegisterTask(string name, Func<object, CancellationToken, Task<object>> task, TimeSpan timeout)
{
_tasks[name] = task;
_events[name] = new ManualResetEventSlim(false);
_taskTimeouts[name] = timeout;
}
public void EmitEvent(string eventName, object parameter = null)
{
if (_events.TryGetValue(eventName, out var eventSlim))
{
_taskResults[eventName] = parameter;
_taskQueue.Enqueue(eventName);
eventSlim.Set(); // Signal the event
}
}
private void ProcessQueue()
{
while (!_cancellationTokenSource.Token.IsCancellationRequested)
{
if (_taskQueue.TryDequeue(out var taskName))
{
var taskFunc = _tasks[taskName];
var eventSlim = _events[taskName];
var timeout = _taskTimeouts[taskName];
eventSlim.Reset();
TaskStarted?.Invoke(taskName);
var cts = new CancellationTokenSource();
var timeoutTask = Task.Delay(timeout, cts.Token);
var completedTask = Task.Run(async () =>
{
try
{
return await taskFunc(_taskResults[taskName], cts.Token);
}
catch (OperationCanceledException)
{
return null;
}
}, cts.Token);
var finishedTask = Task.WhenAny(completedTask, timeoutTask).Result;
if (finishedTask == timeoutTask)
{
TaskFailed?.Invoke(taskName, "Timeout");
cts.Cancel(); // Cancel the task
}
else
{
try
{
var result = completedTask.Result;
TaskCompleted?.Invoke(taskName);
if (result != null)
{
_taskResults[taskName] = result;
}
}
catch (Exception ex)
{
TaskFailed?.Invoke(taskName, ex.Message);
}
}
}
else
{
Thread.Sleep(100); // Prevent tight loop if queue is empty
}
}
}
public void Dispose()
{
_cancellationTokenSource.Cancel();
_workerThread.Join(); // Wait for the worker thread to exit
foreach (var eventSlim in _events.Values)
{
eventSlim.Dispose();
}
_cancellationTokenSource.Dispose();
}
}
使用实例
public class Program
{
public static async Task Main(string[] args)
{
using var taskQueue = new EventDrivenTaskQueue();
taskQueue.TaskStarted += taskId => Console.WriteLine($"Task {taskId} started");
taskQueue.TaskCompleted += taskId => Console.WriteLine($"Task {taskId} completed");
taskQueue.TaskFailed += (taskId, reason) => Console.WriteLine($"Task {taskId} failed: {reason}");
taskQueue.RegisterTask("task1", async (param, cancellationToken) =>
{
Console.WriteLine("Task 1 started with param: " + param);
await Task.Delay(1000, cancellationToken); // Simulate work
Console.WriteLine("Task 1 completed");
return "result_from_task1";
}, TimeSpan.FromSeconds(5));
taskQueue.RegisterTask("task2", async (param, cancellationToken) =>
{
Console.WriteLine("Task 2 started with param: " + param);
await Task.Delay(1000, cancellationToken); // Simulate work
Console.WriteLine("Task 2 completed");
return "result_from_task2";
}, TimeSpan.FromSeconds(5));
taskQueue.RegisterTask("done", async (param, cancellationToken) =>
{
Console.WriteLine("All tasks done with param: " + param);
await Task.Delay(500, cancellationToken); // Simulate work
return null;
}, TimeSpan.FromSeconds(5));
// Start the task chain by emitting the first event
taskQueue.EmitEvent("task1", "initial_param");
// Emit task2 event when task1 completes
taskQueue.TaskCompleted += taskId =>
{
if (taskId == "task1")
{
taskQueue.EmitEvent("task2", taskQueue.TaskResults[taskId]);
}
else if (taskId == "task2")
{
taskQueue.EmitEvent("done", taskQueue.TaskResults[taskId]);
}
};
// Give time for tasks to complete
await Task.Delay(10000);
}
}
原文地址:https://blog.csdn.net/isyoungboy/article/details/140615248
免责声明:本站文章内容转载自网络资源,如本站内容侵犯了原著者的合法权益,可联系本站删除。更多内容请关注自学内容网(zxcms.com)!