基于multithreading的RabbitMQ消费者
我们有一个监听单个RabbitMQ队列的Windows服务并处理该消息。
我们希望扩展相同的Windows服务,以便它可以侦听RabbitMQ的多个队列并处理消息。
不确定是否可以通过使用multithreading来实现,因为每个线程都必须列出(阻塞)队列。
由于我对multithreading很陌生,需要关注以下几点的高级指南,这将有助于我开始构建原型。
是否可以通过使用线程在单个应用程序中侦听多个队列?
如何处理如果任何单个线程被播出的情况(由于exception等),如何在不重启整个Windows服务的情况下返回。
任何设计模式或开源实现可以帮助我处理这种情况。
我喜欢你写你的问题的方式 – 它开始非常广泛,专注于具体细节。 我已经成功地实现了一些非常相似的东西,目前正在开发一个开源项目,以吸取我的经验并将它们带回社区。 不幸的是,虽然 – 我还没有整齐地打包我的代码,这对你没有多大帮助! 无论如何,回答你的问题:
1. Is it possible to use threading for multiple queues.
答:是的,但它可能充满了陷阱。 也就是说,RabbitMQ .NET库并不是最好的代码片段,我发现它是AMQP协议的一个相对麻烦的实现。 其中一个最有害的警告是它如何处理“接收”或“消费”行为,如果你不小心,这会很容易导致死锁。 幸运的是,它在API文档中得到了很好的说明。 建议 – 如果可以,请使用单例连接对象。 然后,在每个线程中,使用连接创建新的IModel
和相应的使用者。
2. How to gracefully handle exceptions in threads
– 我相信这是另一个主题,我不会在这里解决它,因为有几种方法可以使用。
3. Any open-source projects?
– 我喜欢EasyNetQ背后的想法 ,尽管我最后还是自己动手了。 我希望记得在我的开源项目完成后回过头来看,因为我认为它比EasyNetQ更好。
您可能会发现此答案非常有用。 我对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 _commonQueue; // process an individual queue void ProcessQueue(int id, BlockingCollection 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>(); _commonQueue = new BlockingCollection(); // start individual queue processors var tasks = Enumerable.Range(0, 4).Select((i) => { var queue = new BlockingCollection(); 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> 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可以帮助将它们聚合为单个流。
上述就是C#学习教程:基于multithreading的RabbitMQ消费者分享的全部内容,如果对大家有所用处且需要了解更多关于C#学习教程,希望大家多多关注—计算机技术网(www.ctvol.com)!
本文来自网络收集,不代表计算机技术网立场,如涉及侵权请联系管理员删除。
ctvol管理联系方式QQ:251552304
本文章地址:https://www.ctvol.com/cdevelopment/998032.html