RabbitMQ 异常与任务分发

Posted 超越梦想

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了RabbitMQ 异常与任务分发相关的知识,希望对你有一定的参考价值。

RabbitMQ 异常与任务分发

 

异常情况处理

上篇最后提到了这个问题, consumer异常退出、queue出错、甚至rabbitMQ崩溃。因为它们都是软件 ,软件都会有bug,这是无法避免的。所以RabbitMQ在设计的时候也想到了这一点

在之前,消息分发给consumer后立即就会被标记为已消费,这时候如果consumber接到了一个消息但是还没有来的及处理就异常退出,那么这个消息的状态是已被消费的,于是就会造成消息丢失的问题。

可以看到在进行消费的方法里,第二个参数noAck(不进行确认)我们是设置为true。在这里我们应该把它改变成false,也就是 queue需要我们的consumer进行确认这个消息已被正常处理

image

处理的代码也很简单,一共有两个步骤。第一个把noAck改成false

//消费结果需要进行确认
 channel.BasicConsume("firstTest", false, consumer);

第二部分就是在我们消费完成后进行交付

//进行交付,确定此消息已经处理完成
channel.BasicAck(deliveryTag: e.DeliveryTag, multiple: false);

那么 ,如果我们没有进行交付会出现什么情况呢?  queue会把这个消息交给其它的consumer去处理,如果都没有交付的代码呢。那么这个消息会一直存在,所以,千万不要忘了进行交付!

 

消息的处理部分已经结束,下面可以说队列与消息的待久化。事实上关于队列的我们已经有了,可以看声明 queue的第二个参数durable已经是设置为true的。剩下的就是针对我们内容的,也就是消息的持久化

在发布消息的时候,可以看到第三个参数basicProperties传的是null的。这时候我们就应该创建一个IBasicProperties了

//内容的基本属性
 var properties = channel.CreateBasicProperties();
//设置内容的持久化
 properties.Persistent = true;

在发布消息的时候把perperties做成参数传进去,这时我们的更改已经完成了

channel.BasicPublish("firstExchange", routingKey: "firstExchange_Demo_firstTest", basicProperties: properties, body: msg);

 

可以说到最后一个了,在之前queue的消息分发是第n个消息给第n个consumer的,这就会造成最开始处理消息的consumer在处理完成后会有闲置的可能性,而后面的一直在忙,消息分发的不均匀造成了很大的资源浪费,所以我们需要的是把消息分发给闲置的consumber。

这个操作是在consumer中完成的,在声明频道之后就指定这个consumer在同一时间内只处理一个消息,在上个消息未交付之前不要给我再分发消息。这个操作是容易完成的

 

//公平分发、同一时间只处理一个消息。
 channel.BasicQos(0, 1, false);

以上是关于RabbitMQ 异常与任务分发的主要内容,如果未能解决你的问题,请参考以下文章

RABBITMQ初探——消息分发

RabbitMQ消息队列:任务分发机制

RabbitMq之不公平分发预取值

rabbitmq 公平分发和消息接收确认(转载)

RabbitMQ入门:工作队列(Work Queue)

04-RabbitMQ-Work Queues&&消息应答&&持久化&&不公平分发