在并行执行中对任务进行分组，以防止具有相同 groupingID 的消息同时执行

Question

我有一个消费者服务使用 EasyNetQ 订阅者从 RabbitMQ 队列中检索消息。每条消息需要数十秒的时间来处理，我需要并行运行它们以确保我能够跟上生产者的速度。但是，每条消息都有一个属性，称为 groupingId。重要的是不要同时执行具有相同 groupingId 的任务，因为这会导致资源冲突。

可能有数百个 groupingIds，并且在通常的实践中，任何时候都不会有太多消息具有相同的 Id。然而，数据可能是突发性的，导致同时出现数百个相同 Id 的集群。

我认为 TPL Dataflow 可能是一个不错的选择，但我对它不是很熟悉，也不知道如何用它来实现我所需要的。任何指导将不胜感激。

Answer 1

创建分组 ID 字典并锁定它们。

首先，在某处创建字典，可能作为成员变量。

ConcurrentDictionary<int,object> _locks = new ConcurrentDictionary<int, object>();

当你需要处理消息时，使用这个逻辑。

if (!_locks.ContainsKey(message.GroupingID))
{
    _locks.TryAdd(message.GroupingID, new object());
}
lock (_locks[message.GroupingID])
{
    ProcessMessage(message);
}