.netcore3.1 RabbitMq 工作队列轮询与确认消息 电脑版发表于:2021/1/21 13:20 ![](https://img.tnblog.net/arcimg/hb/585b0f1ffa7f4c2095baa20c175b32a0.png) >#.netcore3.1 RabbitMq 工作队列轮询与确认消息 [TOC] https://www.rabbitmq.com/confirms.html https://www.rabbitmq.com/tutorials/tutorial-two-dotnet.html 轮询消费 ------------ ![](https://img.tnblog.net/arcimg/hb/cd27eaba080b4ad09289584f8eeea0f4.png) tn>如上图所示,默认消息队列是通过轮询的方式将消息有序的分发到不同的消费端上去。 >### 创建100条消息 代码示例如下:+ ```csharp var factory = new ConnectionFactory() { HostName = "47.98.187.188", UserName = "bob", Password = "bob" }; // 创建一个链接 using (var connection = factory.CreateConnection()) { // 创建一个通道 using (var channel = connection.CreateModel()) { // 在这里还应该声明一个交换机 // 声明一个队列 channel.QueueDeclare( queue: "mytestqueue", durable: false, exclusive: false, autoDelete: false, arguments: null); for (int i = 0; i < 100; i++) { // 创建一个消息 string message = $"({i}) Hello World"; // 编码一个消息 var body = Encoding.UTF8.GetBytes(message); // 发布一个消息 channel.BasicPublish( exchange: string.Empty, routingKey: "mytestqueue", basicProperties: null, body: body ); } Console.ReadLine(); } } ``` tn>执行完毕后,UI上已经显示出队列的消息数量了。 ![](https://img.tnblog.net/arcimg/hb/77d5805261e4404d8413b4035996d7c7.png) tn>我们也可以点击`mytestqueue`队列里面去进行获取消息,操作如下: ![](https://img.tnblog.net/arcimg/hb/0ea49753a878432495caae2ff8b33eb0.png) >### 开启两个队列去消费100条 tn>在消费时我们可以开启一个事件处理实例,通过事件`Received`去处理`mytestqueue`队列中的消息,代码示例如下: ```csharp var factory = new ConnectionFactory() { HostName = "47.98.187.188", UserName = "bob", Password = "bob" }; // 创建一个链接 using (var connection = factory.CreateConnection()) { // 创建一个通道 using (var channel = connection.CreateModel()) { // 创建消费实例 var consumer = new EventingBasicConsumer(channel); // 事件在交付到使用者时触发。(消费处理事件) consumer.Received += (model, ea) => { var body = ea.Body.ToArray(); var message = Encoding.UTF8.GetString(body); Console.WriteLine(" Processing message: {0}", message); }; // 绑定到队列中去 channel.BasicConsume(queue: "mytestqueue", autoAck: true, consumer: consumer); } } Console.ReadLine(); ``` tn>开启两个进行处理,我们从下面的图中得出它是通过轮询的方式进行消费的。(我这里是先开启的消费端,再添加生产者的方式) ![](https://img.tnblog.net/arcimg/hb/691f0fd47c1f4e78aecd0dd10178998d.png) >### 事件处理是否阻塞 tn>我们在事件处理时添加两行代码,测试一下 ```csharp consumer.Received += (model, ea) => { var body = ea.Body.ToArray(); var message = Encoding.UTF8.GetString(body); Console.WriteLine(" Processing message: {0}", message); // 将它延迟3s Thread.Sleep(3000); Console.WriteLine(" [x] Done Message: {0}",message); }; // 绑定到队列中去 channel.BasicConsume(queue: "mytestqueue", autoAck: true, consumer: consumer); ``` ![](https://img.tnblog.net/arcimg/hb/0059ff7f92ea4b90acf5bd631a7256be.png) ![](https://img.tnblog.net/arcimg/hb/bd9e647b3a154a23886fc3ed3a28cf43.png) tn>我们发现每一个消费端在消费的时候事件都是同步的,都是在等上一个消息被事件处理完成后再处理下一个消息的。但紧接着新的疑惑就有了,如果我们在两个消费端进行消费的时候,突然有一个消费端挂掉了会怎么样? ![](https://img.tnblog.net/arcimg/hb/1257b0e59a1945a6a605323bf97a6d11.png) tn>我们发现它第`7`条`Hello World`并没有被处理完成,然后也没有被重新消费的保险措施。这就产生了新的问题:消息丢失在了消费端。这个时候我们便需要通过消息确认的方式解决这个问题。 消息确认,让消息更加安全可靠 ------------ tn>在代码中将处理实例绑定到队列时的`BasicConsume`方法里有一个`autoAck`的`bool`参数,默认值为`false`。当它为`true`的时候,默认表示**自动确认**该消息被处理了,`false`则表示手动处理的方法。`RabbitMQ`为我们提供了三种手动交付的方法: | 协议方法 | 描述 | 代码中的方法(.net core) | | ------------ | ------------ | ------------ | | basic.ack | 肯定确认 | BasicAck | | basic.nack | 否定确认 | BasicNack | | basic.reject | 否定确认,通过参数可以丢弃消息以及重新排列到队列中 | BasicReject | ACK确认交付 ------------ >### BasicAck手动确认 tn>在代码中我们先把`autoAck`设置为`false`,将自动确认改为手动确认。并在处理消息事件的的末尾添加处理完成的手动确认代码。 ```csharp consumer.Received += (model, ea) => { var body = ea.Body.ToArray(); var message = Encoding.UTF8.GetString(body); Console.WriteLine(" Processing message: {0}", message); Thread.Sleep(3000); Console.WriteLine(" [x] Done Message: {0}",message); // 手动确认 channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false); }; // 绑定到队列中去 channel.BasicConsume(queue: "mytestqueue", autoAck: false, consumer: consumer); ``` tn>在添加100条数据后,开启两个消费端,并在中途关闭掉一个客户端。 ![](https://img.tnblog.net/arcimg/hb/ec7a6baaf2154ce986f7fbc5b0a14e69.png) ![](https://img.tnblog.net/arcimg/hb/579ee389d08b4a59b58684528d8b2ce2.png) ![](https://img.tnblog.net/arcimg/hb/c40d8c5f01d64be7b2545583be692c44.png) tn>我们发现第一个消费端在处理完自己的消费任务后,开始处理断掉的消费端没有处理的消息了,最后解决消息在客户端丢失的问题。(从第13条开始)接下来我们来看看`BasicAck`中长用的参数: | 参数名 | 描述 | | ------------ | ------------ | | deliveryTag | 表示消息标签 | | multiple | 当值为`true`时,表示一次确认就能代表多次确认。为`false`一次确认只能代表一次消息的确认。 | >### BasicAck中multiple为true时的应用 tn>可以分批手动确认以减少网络流量。当它为`true`的时候,表示假如:1,2,3,4,5 这五条消息正在被几个客户端处理的时候突然有一个客户端处理`5`时调用了`multiple`为`true`这方法,此时(1,2,3,4,5)这五条消息表示都已经处理过了。如果为`false`,我只知道我自己处理的那条有没有成功嘛。 tn>例如我将`Worker`的解决方案写为`10`后处理完成。 ```csharp Thread.Sleep(10000); Console.WriteLine(" [x] Done Message: {0}",message); // 手动确认 channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: true); ``` tn>通过100条消息消费时统计`Runtime Metrics (Advanced)`的平均指标。为`true`时平均为`402`,`false`时平均为`406`。 >### BasicReject否定确认与重回队列 tn>有时,消费者无法立即处理交货,但其他情况下可能可以。在这种情况下,可能需要重新排队并让另一个消费者接收和处理它。 >否定确认,并丢弃消息 ```csharp channel.BasicReject(ea.DeliveryTag, false); ``` tn>这里我们消费`20`条数据进行测试一下,消费端代码如下 ```csharp consumer.Received += (model, ea) => { try { var body = ea.Body.ToArray(); var message = Encoding.UTF8.GetString(body); Console.WriteLine("One Processing message: {0}", message); Thread.Sleep(1000); Console.WriteLine(" [x] Done Message: {0}", message); // 否定确认,并丢弃消息 channel.BasicReject(ea.DeliveryTag, false); } catch (Exception ex) { Console.WriteLine("【Error】:", ex.Message); } }; ``` ![](https://img.tnblog.net/arcimg/hb/410591c580a84f37a885b1d95d342304.png) ![](https://img.tnblog.net/arcimg/hb/7fbe7f0b284a4c50a44dfc5f250041a5.png) >否定确认,重新排列到队列 ```csharp channel.BasicReject(ea.DeliveryTag, true); ``` tn>这里我们用随机的方式进行处理,注意需要在外围添加`Random`的实例 ```csharp consumer.Received += (model, ea) => { try { var body = ea.Body.ToArray(); var message = Encoding.UTF8.GetString(body); Console.BackgroundColor = ConsoleColor.Black; Console.ForegroundColor = ConsoleColor.White; Console.WriteLine("One Processing message: {0}", message); Thread.Sleep(1000); if (random.Next(0,2).Equals(0)) { Console.BackgroundColor = ConsoleColor.Blue; //设置背景色 Console.ForegroundColor = ConsoleColor.White; //设置前景色,即字体颜色 Console.WriteLine(" Discard Message: {0}", message); // 否定确认 channel.BasicReject(ea.DeliveryTag, false); } else { Console.BackgroundColor = ConsoleColor.Green; Console.ForegroundColor = ConsoleColor.DarkGreen; Console.WriteLine(" Return Message: {0}", message); // 重新排队 channel.BasicReject(ea.DeliveryTag, true); } } catch (Exception ex) { Console.WriteLine("【Error】:", ex.Message); } }; ``` ![](https://img.tnblog.net/arcimg/hb/1121dfbe1f6d4e14844b07e7823f7a77.png) tn>截图没截到,后面反正又重新排列到队列中去了。重新排队的消息可能立即准备好重新发送,具体取决于它们在队列中的位置以及活动的使用者使用的通道使用的预取值。这意味着,如果所有使用者由于瞬态而无法处理交货而重新排队,则他们将创建一个重新排队/重新交货循环。就网络带宽和CPU资源而言,这样的循环可能代价很高。 >### BasicReject与BasicNack区别 tn>消费者实现可以跟踪重新交付的次数并永久拒绝消息(丢弃消息),或在延迟后安排重新排队。BasicNack方法可以一次拒绝或重新排队多个消息。这就是与BasicReject不同的地方。 ```csharp consumer.Received += (model, ea) => { try { var body = ea.Body.ToArray(); var message = Encoding.UTF8.GetString(body); Console.BackgroundColor = ConsoleColor.Black; Console.ForegroundColor = ConsoleColor.White; Console.WriteLine("One Processing message: {0}", message); Thread.Sleep(1000); if (random.Next(0,2).Equals(0)) { Console.BackgroundColor = ConsoleColor.Blue; //设置背景色 Console.ForegroundColor = ConsoleColor.White; //设置前景色,即字体颜色 Console.WriteLine(" Discard Message: {0}", message); // 否定确认,减少网络请求,提高性能 channel.BasicNack(ea.DeliveryTag,multiple: true,requeue:false); } else { Console.BackgroundColor = ConsoleColor.Green; Console.ForegroundColor = ConsoleColor.DarkGreen; Console.WriteLine(" Return Message: {0}", message); // 重新排队,减少网络请求,提高性能 channel.BasicNack(ea.DeliveryTag,multiple: true,requeue:true); } } catch (Exception ex) { Console.WriteLine("【Error】:", ex.Message); } }; ``` ![](https://img.tnblog.net/arcimg/hb/f46c2540b7f84aa78b03377b226ae128.png) ![](https://img.tnblog.net/arcimg/hb/3120852090874d1aa4befd99c7d1a14c.png) tn>当你忘记了`BasicAck`时,这是一个简单的错误,但是后果很严重。当您的客户端退出时,消息将被重新发送(看起来像是随机重新发送),但是RabbitMQ将消耗越来越多的内存,因为它将无法释放任何未确认的消息。 为了调试这种错误,您可以使用 `rabbitmqctl` 打印 `messages_unacknowledged` 字段: ![](https://img.tnblog.net/arcimg/hb/d4d1f246e6e54c34bc4bcd4353b7ee17.png) 持久化 ------------ >### 问题 tn>我们已经学会了如何确保即使消费者死亡,任务也不会丢失。但是,如果`RabbitMQ服务器停止`,我们的任务仍然会丢失。<br/> RabbitMQ退出或崩溃时,除非您告诉它,否则它将忘记队列和消息。确保消息不会丢失需要做两件事:我们需要将`Queue`和`Message`都标记为持久。 >### Queue 持久化 tn>我们在声明队列的时候,将`durable`设置为`true`。这样就可以将队列持久化,但注意:已经声明的队列这样做是无效的,我们可以声明一个名字不同新的队列。 ```csharp channel.QueueDeclare( queue: "mytestqueue", durable: true, exclusive: false, autoDelete: false, arguments: null); ``` >### Message 持久化 tn>在这一点上,我们确定即使RabbitMQ重新启动,`task_queue`队列也不会丢失。现在我们需要将消息标记为持久性-通过将`IBasicProperties.SetPersistent`设置为`true`。 ```csharp var body = Encoding.UTF8.GetBytes(message); // 持久化操作,告诉Rabbitmq服务器将消息存储在磁盘上,但仍然有 // 少部分是存储在缓存中的,所以仍然不能绝对保证 var properties = channel.CreateBasicProperties(); properties.Persistent = true; // 发布一个消息 channel.BasicPublish( exchange: string.Empty, routingKey: "mytestqueuetwo", basicProperties: properties, body: body ); ``` ![](https://img.tnblog.net/arcimg/hb/7dc9cd3571ae484a99b5187f3b7f4286.png) 消息分配合理化 ------------ tn>当每条消息在`A消费端`都需要处理很长的时间,才能确认消息被处理的时候。而`B消费端`却在玩的时候,仍然平均去分发消息,就会显得很不合理。<br/> 为了更改此行为,我们可以将`BasicQos`方法与`prefetchCount = 1`设置一起使用。这告诉RabbitMQ一次不要给工人一个以上的消息。换句话说,在处理并确认上一条消息之前,不要将新消息发送给工作人员。而是将其分派给不忙的下一个工作程序。 ```csharp channel.BasicQos(0, 1, false); ``` tn>注意:如果所有工作人员都忙,则您的队列可以填满。您将需要关注这一点,并可能会增加更多的工作人员,或者有其他一些策略。 >创建Queue添加消息的代码如下 ```csharp var factory = new ConnectionFactory() { HostName = "47.98.187.188", UserName = "bob", Password = "bob" }; // 创建一个链接 using (var connection = factory.CreateConnection()) { // 创建一个通道 using (var channel = connection.CreateModel()) { // 在这里还应该声明一个交换机 // 声明一个队列 channel.QueueDeclare( queue: "mytestqueuetwo", durable: true, exclusive: false, autoDelete: false, arguments: null); for (int i = 0; i < 30; i++) { // 创建一个消息 string message = $"({i}) Hello World"; // 编码一个消息 var body = Encoding.UTF8.GetBytes(message); // 持久化操作,告诉Rabbitmq服务器将消息存储在磁盘上,但仍然有 // 少部分是存储在缓存中的,所以仍然不能绝对保证 var properties = channel.CreateBasicProperties(); properties.Persistent = true; // 发布一个消息 channel.BasicPublish( exchange: string.Empty, routingKey: "mytestqueuetwo", basicProperties: properties, body: body ); } Console.WriteLine("Finish"); Console.ReadLine(); } } ``` >消费端代码 ```csharp var factory = new ConnectionFactory() { HostName = "47.98.187.188", UserName = "bob", Password = "bob", VirtualHost = "/" }; // 获取处理时常(第一个耗时为10s,第二个耗时为2s) var sleeptime = int.Parse(args[0]); // 创建一个链接 using (var connection = factory.CreateConnection()) { // 创建一个通道 using (var channel = connection.CreateModel()) { // one by one 处理 channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false); // 创建消费实例 var consumer = new EventingBasicConsumer(channel); // 事件在交付到使用者时触发。(消费处理事件) consumer.Received += (model, ea) => { try { int sj = sleeptime; var body = ea.Body.ToArray(); var message = Encoding.UTF8.GetString(body); Console.BackgroundColor = ConsoleColor.Black; Console.ForegroundColor = ConsoleColor.White; Console.WriteLine("One Processing message: {0}", message); Thread.Sleep(sj*1000); if (random.Next(0,2).Equals(0)) { Console.BackgroundColor = ConsoleColor.Blue; //设置背景色 Console.ForegroundColor = ConsoleColor.White; //设置前景色,即字体颜色 Console.WriteLine(" Discard Message: {0}", message); // 否定确认,减少网络请求,提高性能 channel.BasicNack(ea.DeliveryTag,multiple: true,requeue:false); } else { Console.BackgroundColor = ConsoleColor.Green; Console.ForegroundColor = ConsoleColor.DarkGreen; Console.WriteLine(" Return Message: {0}", message); // 重新排队,减少网络请求,提高性能 channel.BasicNack(ea.DeliveryTag,multiple: true,requeue:true); } } catch (Exception ex) { Console.WriteLine("【Error】:", ex.Message); } }; // 绑定到队列中去 channel.BasicConsume(queue: "mytestqueuetwo", autoAck: false, consumer: consumer); Console.ReadLine(); } } ``` ![](https://img.tnblog.net/arcimg/hb/8f97e57f3495443faf34a3ca7878f53b.png) 绑定交换机 ------------ tn>在Rabbitmq中,在一个交换机上可以绑定多个队列,如下图所示 ![](https://img.tnblog.net/arcimg/hb/d1d9e47e6abb43228ca9f0580f91fa29.png) tn>在生产的时候我们就可以通过`QueueBind`将队列绑定到交换机上。 ```csharp var factory = new ConnectionFactory() { HostName = "47.98.187.188", UserName = "bob", Password = "bob" }; // 创建一个链接 using (var connection = factory.CreateConnection()) { // 创建一个通道 using (var channel = connection.CreateModel()) { // 创建交换机 普通类型 channel.ExchangeDeclare("MyExchangeName", ExchangeType.Direct); // 声明一个队列 channel.QueueDeclare( queue: "mytestqueuetwo", durable: true, exclusive: false, autoDelete: false, arguments: null); // 将队列与交换机绑定在一起 channel.QueueBind("mytestqueuetwo", "MyExchangeName", "mytestqueuetwo", null); Console.WriteLine("Finish"); Console.ReadLine(); } } ``` ![](https://img.tnblog.net/arcimg/hb/16b3b36d1244439386029ccc1b04d694.png) ![](https://img.tnblog.net/arcimg/hb/8d28a18777b5471e9add0dce457828af.png) ![](https://img.tnblog.net/arcimg/hb/9a02faf58d4244c8b3c529e8ed4e9b32.png)