基于多线程的 RabbitMQ 消费者

Question

我们有一个 Windows 服务，它监听单个 RabbitMQ 队列并处理消息。

我们希望扩展相同的 Windows 服务，以便它可以监听多个 RabbitMQ 队列并处理消息。

不确定是否可以通过使用多线程来实现，因为每个线程都必须监听（阻塞）队列。

由于我对多线程非常陌生，因此需要以下几点的高级指南，这将有助于我开始构建原型。

1. 是否可以使用线程在单个应用程序中侦听多个队列？
2. 单线程关闭时如何处理 down（由于异常等），如何在不重新启动的情况下恢复 整个 Windows 服务。
3. 任何可以帮助我处理这种情况的设计模式或开源实现。

Answer 1

我喜欢你写问题的方式——它一开始就非常广泛，而且侧重于细节。我已经成功地实现了一些非常相似的东西，并且目前正在开发一个开源项目，以吸取我的经验教训并将其回馈给社区。不幸的是，我还没有把我的代码整齐地打包，这对你没有多大帮助！无论如何，回答你的问题：

1. Is it possible to use threading for multiple queues.

A：是的，但它可能充满陷阱。也就是说，RabbitMQ .NET 库并不是最好的代码，我发现它是 AMQP 协议的一个相对繁琐的实现。最有害的警告之一是它如何处理“接收”或“消费”行为，如果你不小心，很容易导致死锁。幸运的是，它在 API 文档中得到了很好的说明。 建议 - 如果可以，请使用单例连接对象。然后，在每个线程中，使用连接创建一个新的IModel和对应的消费者。

2. How to gracefully handle exceptions in threads - 我相信这是另一个话题，我不会在这里讨论，因为您可以使用多种方法。

3. Any open-source projects? - 我喜欢EasyNetQ 背后的想法，尽管我最终还是自己动手了。希望我的开源项目完成后我会记得跟进，因为我相信它比 EasyNetQ 有更好的改进。

Answer 2

您可能会发现this answer 很有帮助。我对 RabbitMQ 的工作原理有一个非常基本的了解，但我可能会按照那里的建议继续每个线程每个通道一个订阅者。

为此组织线程模型肯定不止一个选项。实际的实现将取决于您需要如何处理来自多个队列的消息：并行，或通过聚合它们并序列化处理。以下代码是一个控制台应用程序，它实现了后一种情况的模拟。它使用Task Parallel Library 和BlockingCollection 类（对于此类任务非常方便）。

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

namespace Console_21842880
{
    class Program
    {
        BlockingCollection<object> _commonQueue;

        // process an individual queue
        void ProcessQueue(int id, BlockingCollection<object> queue, CancellationToken token)
        {
            while (true)
            {
                // observe cancellation
                token.ThrowIfCancellationRequested();
                // get a message, this blocks and waits
                var message = queue.Take(token);

                // process this message
                // just place it to the common queue
                var wrapperMessage = "queue " + id + ", message: " + message;
                _commonQueue.Add(wrapperMessage);
            }
        }

        // process the common aggregated queue
        void ProcessCommonQeueue(CancellationToken token)
        {
            while (true)
            {
                // observe cancellation
                token.ThrowIfCancellationRequested();
                // this blocks and waits

                // get a message, this blocks and waits
                var message = _commonQueue.Take(token);

                // process this message
                Console.WriteLine(message.ToString());
            }
        }

        // run the whole process
        async Task RunAsync(CancellationToken token)
        {
            var queues = new List<BlockingCollection<object>>();
            _commonQueue = new BlockingCollection<object>();

            // start individual queue processors
            var tasks = Enumerable.Range(0, 4).Select((i) =>
            {
                var queue = new BlockingCollection<object>();
                queues.Add(queue);

                return Task.Factory.StartNew(
                    () => ProcessQeueue(i, queue, token), 
                    TaskCreationOptions.LongRunning);
            }).ToList();

            // start the common queue processor
            tasks.Add(Task.Factory.StartNew(
                () => ProcessCommonQeueue(token),
                TaskCreationOptions.LongRunning));

            // start the simulators
            tasks.AddRange(Enumerable.Range(0, 4).Select((i) => 
                SimulateMessagesAsync(queues, token)));

            // wait for all started tasks to complete
            await Task.WhenAll(tasks);
        }

        // simulate a message source
        async Task SimulateMessagesAsync(List<BlockingCollection<object>> queues, CancellationToken token)
        {
            var random = new Random(Environment.TickCount);
            while (true)
            {
                token.ThrowIfCancellationRequested();
                await Task.Delay(random.Next(100, 1000));
                var queue = queues[random.Next(0, queues.Count)];
                var message = Guid.NewGuid().ToString() + " " +  DateTime.Now.ToString();
                queue.Add(message);
            }
        }

        // entry point
        static void Main(string[] args)
        {
            Console.WriteLine("Ctrl+C to stop...");

            var cts = new CancellationTokenSource();
            Console.CancelKeyPress += (s, e) =>
            {
                // cancel upon Ctrl+C
                e.Cancel = true;
                cts.Cancel();
            };

            try
            {
                new Program().RunAsync(cts.Token).Wait();
            }
            catch (Exception ex)
            {
                if (ex is AggregateException)
                    ex = ex.InnerException;
                Console.WriteLine(ex.Message);
            }

            Console.WriteLine("Press Enter to exit");
            Console.ReadLine();
        }
    }
}

另一个想法可能是使用Reactive Extensions (Rx)。如果您可以将到达的消息视为事件，那么 Rx 可以帮助将它们聚合成单个流。