在前面一章介绍了在.Net Core中如何使用RabbitMQ,至此入门的的部分就完成了,我们内心中一定还有很多疑问:如果多个消费者消费同一个队列怎么办?如果这几个消费者分任务的权重不同怎么办?怎么把同一个队列不同级别的任务分发给不同的消费者?如果消费者异常离线怎么办?不要着急,后面将慢慢解开面纱。我们将结合实际的应用场景来讲解更多的高级用法。
任务分发机制
设想如果把每个消息当做一个任务,生产者把任务发布到RabbitMQ,然后Consumer接收消息处理任务,如果我们发现一个Consumer不能完成任务处理怎么办呢,我们会增加Consumer的数量。由一个Consumer增加到两个Consumer,如图由C变为C1和C2共同来分单工作。如果C1和C2是完全一样的,那RabbitMQ会将任务平均分发到两个消费者。
如下我们
新建ProductAckDemo开发布订阅内容
新建ConsumerAckDemo1和ConsumerAckDemo2项目来订阅同一个队列在接收到消息后sleep1秒模拟任务处理的时间。
ProductAckDemo代码,生产100条带编号的消息:
using System;using System.Text;using RabbitMQ.Client;using RabbitMQ.Client.Events;namespace ProductAckDemo{ class Program { static void Main(string[] args) { String exchangeName = "wytExchange"; String routeKeyName = "wytRouteKey"; String message = "Task --"; ConnectionFactory factory = new ConnectionFactory(); factory.HostName = "192.168.63.129"; factory.Port = 5672; factory.VirtualHost = "/wyt"; factory.UserName = "wyt"; factory.Password = "wyt"; using (IConnection connection=factory.CreateConnection()) { using (IModel channel=connection.CreateModel()) { channel.ExchangeDeclare(exchange: exchangeName, type: "direct", durable: true, autoDelete: false, arguments:null); channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false); IBasicProperties properties = channel.CreateBasicProperties(); properties.Persistent = true; for (int i = 0; i < 100; i++) { Byte[] body = Encoding.UTF8.GetBytes(message+i); channel.BasicPublish(exchange: exchangeName, routingKey: routeKeyName, basicProperties: properties, body: body); } } } } }}
ConsumerAckDemo1与ConsumerAckDemo2代码(一样的)
using System;using System.Text;using System.Threading;using RabbitMQ.Client;using RabbitMQ.Client.Events;namespace ConsumerAckDemo1{ class Program { static void Main(string[] args) { String exchangeName = "wytExchange"; String queueName = "wytQueue"; String routeKeyName = "wytRouteKey"; ConnectionFactory factory = new ConnectionFactory(); factory.HostName = "192.168.63.129"; factory.Port = 5672; factory.VirtualHost = "/wyt"; factory.UserName = "wyt"; factory.Password = "wyt"; using (IConnection connection=factory.CreateConnection()) { using (IModel channel=connection.CreateModel()) { channel.ExchangeDeclare(exchange: exchangeName, type: "direct", durable: true, autoDelete: false, arguments: null); channel.QueueDeclare(queue: queueName, durable: true, exclusive: false, autoDelete: false, arguments: null); channel.QueueBind(queue: queueName, exchange: exchangeName, routingKey: routeKeyName, arguments: null); channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false); EventingBasicConsumer consumer = new EventingBasicConsumer(channel); consumer.Received += (model, ea) => { Byte[] body = ea.Body; String message = Encoding.UTF8.GetString(body); Console.WriteLine(" [x] {0}", message); Thread.Sleep(1000); channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false); }; channel.BasicConsume(queue: queueName, autoAck: false, consumer: consumer); Console.WriteLine(" Press [enter] to exit."); Console.ReadLine(); } } } }}
打开两个中断窗口分别执行ConsumerAckDemo1和ConsumerAckDemo2。确定两个程序处于订阅状态,然后执行ProductAckDemo程序。
看到上面的图结果就一目了然了。因为两个程序的时间相同所以任务是完全平均分发到两个消费者的。我们修改下ConsumerAckDemo2脚本的sleep时间为2秒,看下结果会怎么样。
可以看到ConsumerAckDemo1共收到66条消息,ConsumerAckDemo2脚本收到34条消息,基本是按照2:1来分配。那RabbitMQ是如何来保证这样的分发机制呢,下面看RabbitMQ是如何通过ACK确认机制来实现任务分发的。
ACK消息确认机制
首先RabbitMQ支持消息确认机制来本证消息被consumer正常处理,当然也可以通过no-ack不使用确认机制。RabbitMQ默认是使用ACK确认机制的。当Consumer接收到RabbitMQ发布的消息时需要在适当的时机发送一个ACK确认的包来告知RabbitMQ,自己接收到了消息并成功处理。所以前面讲到适当的时机建议是在处理完消息任务后发送。正如我们之前的代码
EventingBasicConsumer consumer = new EventingBasicConsumer(channel);consumer.Received += (model, ea) =>{ ... channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);//手动发送ACK应答 ...};channel.BasicConsume(queue: queueName, autoAck: false, consumer: consumer);//不自动应答
那如果不发送会怎样呢?
在RabbitMQ中有一个prefetch_count的概念,这个参数的意思是允许Consumer最多同时处理几个任务。我的版本的RabbitMQ默认这个参数是1,也就是说如果某一个Consumer在收到消息后没有发送ACK确认包,RabbitMQ就会任务Consumer还在处理任务,当有1个消息都没有发送ACK确认包时,RabbitMQ就不会再发送消息给该Consumer。
我们把ConsumerAckDemo2的sleep时间改回1秒,并且注释掉ACK确认。发现ConsumerAckDemo2只收到1条消息。通过WEB管理工具也可以看到有1条消息是没有被ACK确认的
当然任务并不会一直卡在这里,在这是RabbitMQ任务ConsumerAckDemo2在处理这1个任务。如果ConsumerAckDemo2忽然终止RabbitMQ会重新分发任务。如果我终止ConsumerAckDemo2,1条任务被重新分发到了ConsumerAckDemo1。再查看下WEB管理工具,unackd已经为0
如果Consumer数量很多或者希望每个Consumer同时只处理一个任务可以通过在Consumer中设置PrefetchCount来实现更加均匀的任务分发。
channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);
如下我修改PrefetchCount为1。在WEB管理插件中可以看到已经有一个Consumer的PrefetchCount为1了。