你的描述有点轻描淡写。但是，您可以通过数据库和某种管道（选择您的毒药）轻松解决此问题
在这个非常人为的例子中，我使用了dataflow，您可以使用任何您喜欢的结构或框架，但是问题仍然是一样的。在这个示例中，dataflow可以毫不费力地完成一些事情。
可以使用异步和等待模式。
以有序的方式处理事物（或不）
可以使用队列进行处理，可以并行处理事情
配置最大并行度
可以创建永久管道
可以取消代币和更多吗
我不得不做了很多假设，留下了很多想象。
您需要考虑容错性
实行取消制度
调整平行度和其他选项
为事件实现一个数据库
如果你的进程失败了，有一个故障恢复和重启机制
例子

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

注意：这只是一个例子，并不意味着它是一个完整的解决方案或数据流的建议，甚至你应该如何解决它。它只是给你一个结构化管道的概念。