并发处理rabbitmq消息答案

【问题标题】：Handle rabbitmq messages concurrenrtly并发处理rabbitmq消息
【发布时间】：2017-02-16 22:14:15
【问题描述】：

我问了一个问题here，关于为什么使用 Thread.Run 启动进程没有执行我预期的那么多并发请求。

这个问题背后的原因是我试图创建一个类，它可以从 rabbitmq 队列中提取消息并同时处理它们，最多可以同时处理最大数量的并发消息。

为此，我在 EventingBasicConsumer 类的 Received 处理程序中完成了以下操作。

async void Handle(EventArgs e) 
{
    await _semaphore.WaitAsync();

    var thread = new Thread(() =>
    {
        Process(e);
        _semaphore.Release(); 
        _channel.BasicAck(....);
    });
    thread.Start();
}

然而，上一篇文章中的 cmets 除非做 CPU 密集型工作，否则不会启动线程。

上述处理程序不知道工作是否会受到 CPU 限制、网络、磁盘或其他方面的限制。（Process 是一个抽象方法）。

即便如此我觉得还是要在这里启动一个线程或任务，否则Process方法会阻塞rabbitmq线程，直到完成后才会再次调用事件处理程序。所以我一次只能处理一种方法。

在这里开始一个新的Thread 好吗？最初我使用了Task.Run，但这并没有产生所需的工人数量。见其他帖子。

仅供参考。通过在信号量上设置InitialCount 来限制并发线程的数量。

【问题讨论】：

标签： c# multithreading rabbitmq

【解决方案1】：

正如在链接问题中已经说过的那样，大量线程并不能保证性能，好像它们的数量超过了逻辑核心的数量，你会遇到thread starvation 没有实际工作的情况。

但是，如果您仍然需要处理并发操作的数量，您可以尝试使用 TPL Dataflow 库，设置 MaxDegreeOfParallelism，就像在 this tutorial 中一样。

var workerBlock = new ActionBlock<EventArgs>(
    // Process event
    e => Process(e),
    // Specify a maximum degree of parallelism.
    new ExecutionDataflowBlockOptions
    {
        MaxDegreeOfParallelism = InitialCount
    });
var bufferBlock = new BufferBlock();
// link the blocks for automatically propagading the messages
bufferBlock.LinkTo(workerBlock);

// asynchronously send the message
await bufferBlock.SendAsync(...);
// synchronously send the message
bufferBlock.Post(...);

BufferBlock 是一个队列，所以消息的顺序会被保留。此外，您可以添加不同的处理程序（具有不同程度的并行性），并使用过滤器 lambda 链接块：

bufferBlock.LinkTo(cpuWorkerBlock, e => e is CpuEventArgs);
bufferBlock.LinkTo(networkWorkerBlock, e => e is NetworkEventArgs);
bufferBlock.LinkTo(diskWorkerBlock, e => e is DiskEventArgs);

但在这种情况下，您应该在链的末尾设置一个默认处理程序，这样消息就不会消失（您可以为此使用 NullTarget 块）：

bufferBlock.LinkTo(DataflowBlock.NullTarget<EventArgs>);

此外，该块可以是观察者，因此它们可以在 UI 端与 Reactive Extensions 完美配合。

【讨论】：