RabbitMQ消费者中的并发消息处理

Question

我对RabbitMQ很陌生，所以如果我的问题听起来很琐碎，请原谅。我想在RabbitMQ上发布消息，它将由RabbitMQ消费者处理。

我的消费者机器是一个多核机器（最好是azure上的工作者角色）。但QueueBasicConsumer一次推送一条消息。我如何编程来利用我可以同时处理多个消息的所有核心。

1. 一种解决方案是在多个线程中打开多个通道，然后在那里处理消息。但在这种情况下，我将如何决定线程的数量。
2. 另一种方法是在主线程上读取消息，然后创建任务并将消息传递给该任务。在这种情况下，我将不得不停止使用消息，以防有许多消息（超过一个阈值）正在进行。不确定该如何实现。

Answer 1

您的第二个选项听起来更合理--在一个通道上消费，并产生多个任务来处理消息。要实现并发控制，可以使用信号量来控制飞行中的任务数。在启动任务之前，您将等待信号量变得可用，在任务完成之后，它将向信号量发出信号，以允许其他任务运行。

您还没有指定所选择的语言/技术栈，但不管您做什么--尝试使用线程池，而不是自己创建和管理线程。在.NET中，这意味着使用task.run异步处理消息。

C#代码示例：

using (var semaphore = new SemaphoreSlim(MaxMessages))
{
    while (true)
    {
        var args = (BasicDeliverEventArgs)consumer.Queue.Dequeue();
        semaphore.Wait();
        Task.Run(() => ProcessMessage(args))
            .ContinueWith(() => semaphore.Release());
    }
}