c# - RabbitMQ 中的并发

Question

经过一周的编码和搜索论坛，似乎是时候问...

我有一个 C# 应用程序，它使用 EventingBasicConsumer 处理 RabbitMQ 发送的消息。我想同时处理多条消息，所以我在同一个连接上实例化了几个 channel (在本例中为 8 个)，每个 channel 都有一个消费者。然后我将一个事件处理程序附加到每个消费者的 Received 事件。根据我目前的所有阅读，此设置应该允许事件处理程序由消费者并发触发，每个消费者都在自己的线程中运行。但在我的例子中，消费者仅在前一个消费者确认其消息后才按顺序接收消息。

有没有其他人经历过这种行为？我的理解是否正确，在这种情况下处理在技术上应该是并发的？

下面是一个基本代码，可以更好地说明问题:

Initialise() {
    ConsumerChannels_ = new IModel[ConsumerCount_];
    Consumers_ = new EventingBasicConsumer[ConsumerCount_];
    for (int i = 0; i < ConsumerCount_; ++i)
    {
         ConsumerChannels_[i] = Connection_.CreateModel();
         Consumers_[i] = new EventingBasicConsumer(ConsumerChannels_[i]);
         Consumers_[i].Received += MessageReceived;
    }
}

MessageReceived(IBasicConsumer sender, BasicDeliverEventArgs e)
{
    int id = GetConsumerIndex(sender);
    Log_.Debug("Consumer " + id + ": processing started...");         
    // do some time consuming processing here
    sender.Model.BasicAck(e.DeliveryTag, false);
    Log_.Debug("Consumer " + id + ": processing ended.");
}

我希望看到的是://并发处理

Consumer 1: processing started...

Consumer 2: processing started...

Consumer 3: processing started...

...

Consumer 6: processing ended.

Consumer 7: processing ended.

Consumer 8: processing ended.

但我得到的是://顺序处理

Consumer 1: processing started...

Consumer 1: processing ended.

Consumer 2: processing started...

Consumer 2: processing ended.

...

Consumer 8: processing started...

Consumer 8: processing ended.

如有任何关于如何进行的想法，我们将不胜感激。

Answer 1

您实际上可以在创建 ConnectionFactory 时设置并行处理任务的数量!

ConnectionFactory factory = new ConnectionFactory
{
    ConsumerDispatchConcurrency = 2,
};

默认值为1，即serial/sequential processing。

我通过剖析 .NET client's source code 发现了这一点.这是有趣的部分(concurrency 是从 ConsumerDispatchConcurrency 设置的):

Func<Task> loopStart = ProcessChannelAsync;
if (concurrency == 1)
{
    _worker = Task.Run(loopStart);
}
else
{
    var tasks = new Task[concurrency];
    for (int i = 0; i < concurrency; i++)
    {
        tasks[i] = Task.Run(loopStart);
    }
    _worker = Task.WhenAll(tasks);
}

但请注意，这可能会导致竞争条件!该属性有此备注:

For concurrency greater than one this removes the guarantee that consumers handle messages in the order they receive them. In addition to that consumers need to be thread/concurrency safe.