您的描述对细节有点轻描淡写。但是，您可以使用数据库和某种管道（选择您的毒药）轻松解决此问题
在这个非常人为的示例中，我使用了 Dataflow，您可以使用您喜欢的任何结构或框架，但是问题仍然相同。在示例中，Dataflow 有一些事情可以毫不费力地完成。
- 可以使用 async 和 await 模式。
- 有序（或不有序）处理事物
- 可以使用队列处理，可以并行处理
- 配置最大并行度
- 可以创建永久管道
- 可以取消令牌等等
我不得不做出很多假设，并留下了很多想象空间。
- 您需要考虑容错性
- 实施取消制度
- 调整并行度和其他选项
- 为事件实现数据库
- 如果您的进程出现故障，则具有故障转移和重新启动机制
示例
public enum EventType
{
Event,
Final,
Finished,
Error
}
public class EventMessage
{
public int GroupId { get; set; }
public int EventId { get; set; }
public string Payload { get; set; }
public EventType EventType { get; set; }
}
public static ConcurrentDictionary<int,List<EventMessage>> _dataStore = new ConcurrentDictionary<int,List<EventMessage>>();
private static BufferBlock<EventMessage> _start;
private static ActionBlock<EventMessage> _persistBlock;
private static ActionBlock<EventMessage> _processBlock;
private static ActionBlock<EventMessage> _finalizeBlock;
private static TransformBlock<EventMessage, EventMessage> _reprocessBlock;
private static TransformBlock<EventMessage, EventMessage> _queue;
private static Random _r = new Random();
static async Task Main(string[] args)
{
// this is just a buffer that can receive asynchronous events
_start = new BufferBlock<EventMessage> (new DataflowBlockOptions(){EnsureOrdered = true});
// we need an orderly queue, the bounded capacity is 1 so we can process events in order
// ie so you don't process the final before all events are recevied
_queue = new TransformBlock<EventMessage, EventMessage>(message => message, new ExecutionDataflowBlockOptions(){BoundedCapacity = 1});
// save your events to the database
_persistBlock = new ActionBlock<EventMessage>(PersistAction, new ExecutionDataflowBlockOptions() { BoundedCapacity = 1 });
// process the final event
_processBlock = new ActionBlock<EventMessage>(ProcessAction);
// process the event from the 3rd party service
_finalizeBlock = new ActionBlock<EventMessage>(FinalizeAction);
// reprocess on failure or whatever you need to do
_reprocessBlock = new TransformBlock<EventMessage, EventMessage>(Reprocess);
// link it all together
_start.LinkTo(_queue);
_queue.LinkTo(_persistBlock, (x) => x.EventType == EventType.Event);
_queue.LinkTo(_processBlock, (x) => x.EventType == EventType.Final);
_queue.LinkTo(_finalizeBlock, (x) => x.EventType == EventType.Finished);
_queue.LinkTo(_reprocessBlock, (x) => x.EventType == EventType.Error);
_reprocessBlock.LinkTo(_start);
// create some events
var tasks= Enumerable.Range(1, 5).Select(CreateEvents);
await Task.WhenAll(tasks);
Console.ReadKey();
}
private static async Task CreateEvents(int groupId)
{
var events = Enumerable
.Range(1, _r.Next(2, 5))
.Select(x => new EventMessage()
{
GroupId = groupId,
EventId = x,
EventType = EventType.Event
});
foreach (var e in events)
{
await Task.Delay(_r.Next(10, 100));
await _start.SendAsync(e);
}
await _start.SendAsync(new EventMessage()
{
GroupId = groupId,
Payload = $"Final Event",
EventType = EventType.Final
});
}
private static EventMessage Reprocess(EventMessage e)
{
// the event come back as an error, so we push it back on the the queue
Console.WriteLine($"Reprocessing group : {e.GroupId}");
e.EventType = EventType.Final;
e.Payload = e.Payload + " Error";
return e;
}
private static async Task PersistAction(EventMessage e)
{
// this is simulating saving the event to a db
Console.WriteLine($"Saving event : {e.GroupId}:{e.EventId}");
await Task.Delay(_r.Next(10, 100));
_dataStore.AddOrUpdate(e.GroupId,
(x) => new List<EventMessage>() {e},
(x, l) =>
{
l.Add(e);
return l;
});
}
private static async Task ProcessAction(EventMessage e)
{
// this is simulating reading all the events for that group from the db
// and sending to your 3rd service
Console.WriteLine($"Sending to service : {e.GroupId}");
await Task.Delay(_r.Next(10, 100));
// this is simulating receiving a result from the 3rd party service
// just pushes the event back in to the queue, to be finialised or reprocessed
// choose randomly if it was a success or failure
// obviously this would be called by something else, possibly your message queue
if (_r.Next(0, 2) == 0)
e.EventType = EventType.Finished;
else
e.EventType = EventType.Error;
Console.WriteLine($"Service returned : {e.GroupId}, {e.EventType}");
await _start.SendAsync(e);
}
private static void FinalizeAction(EventMessage e)
{
// pruge the records, we are all done
_dataStore.TryRemove(e.GroupId, out var l);
Console.WriteLine($"*** Finalize : {e.GroupId} - {string.Join(",", l.Select(x => x.EventId))}");
}
输出
Saving event : 4:1
Saving event : 1:1
Saving event : 4:2
Saving event : 1:2
Saving event : 5:1
Saving event : 5:2
Saving event : 3:1
Saving event : 2:1
Saving event : 1:3
Saving event : 5:3
Sending to service : 1
Saving event : 5:4
Service returned : 1, Error
Sending to service : 5
Saving event : 2:2
Service returned : 5, Error
Saving event : 3:2
Saving event : 4:3
Saving event : 4:4
Sending to service : 4
Saving event : 2:3
Service returned : 4, Error
Saving event : 3:3
Sending to service : 3
Saving event : 2:4
Reprocessing group : 1
Reprocessing group : 5
Reprocessing group : 4
Service returned : 3, Error
Sending to service : 2
Reprocessing group : 3
Service returned : 2, Finished
Sending to service : 1
*** Finalize : 2 - 1,2,3,4
Service returned : 1, Finished
Sending to service : 5
*** Finalize : 1 - 1,2,3
Service returned : 5, Finished
Sending to service : 4
*** Finalize : 5 - 1,2,3,4
Service returned : 4, Finished
Sending to service : 3
*** Finalize : 4 - 1,2,3,4
Service returned : 3, Error
Reprocessing group : 3
Sending to service : 3
Service returned : 3, Finished
*** Finalize : 3 - 1,2,3
注意：这只是一个示例，它并不意味着是一个完整的解决方案或数据流的建议，甚至你应该如何解决它。这只是为了让您了解结构化管道。