RabbitMQ 消息确认机制

消息确认机制

在之前异常处理部分就已经写了，对于consumer的异常退出导致消息丢失，可以时候consumer的消息确认机制。重复的就不说了，这里说一些不一样的。

consumer的消息确认机制

当一个消费者收到一个快递，但是这个包裹是破损的，这时候一般会有以下选择

下面是具体的方法，BasicReject同时承担了扔掉消息与退回。区别在第二个参数。BasicNack则是批量进行上面两个操作，DeliveryTag小于或等于当前消息的都会进行该操作，当然是否批量是由第二个参数来决定的

//扔掉消息
channel.BasicReject(result.DeliveryTag, false);

//退回消息
channel.BasicReject(result.DeliveryTag, true);

//批量退回或删除,中间的参数 是否批量 true是/false否 (也就是只一条)
channel.BasicNack(result.DeliveryTag, true, true);

BasicRecover方法则是进行补发操作，其中的参数如果为true是把消息退回到queue但是有可能被其它的consumer接收到，设置为false是只补发给当前的consumer

//补发消息 true退回到queue中/false只补发给当前的consumer
channel.BasicRecover(true);

消息发布者的确认机制

消息不只是在consumer处理的时候出问题，在发布的时候也可能会出问题。不是说把消息向方法里一丢就一定会成功的。所以发布的确认机制也是为高可靠性保驾护航。

rabbitmq为我们提供了两种方式


但是要知道这些都是耗费性能的，其中事务的性能消耗最大，confirm其次

第一种confirm方式，发布后等待rabbitmq返回消息发布状态

//创建返回一个新的频道
using (var channel = RabbitMqHelper.GetConnection().CreateModel())
{
for (var i = 0; i < 6; i++)
{
channel.BasicPublish(string.Empty, "testqueue", null, Encoding.UTF8.GetBytes($"这是{i}个消息"));
}
//等待发布成功并返回发布状态
bool isok = channel.WaitForConfirms();

Console.ReadKey();

}

第二种事务控制方式

//创建返回一个新的频道
using (var channel = RabbitMqHelper.GetConnection().CreateModel())
{
try
{
//锁往
channel.TxSelect();
for (var i = 0; i < 6; i++)
{
channel.BasicPublish(string.Empty, "testqueue", null, Encoding.UTF8.GetBytes($"这是{i}个消息"));
}
//提交
channel.TxCommit();

Console.ReadKey();
}
catch (Exception e)
{
//回退
channel.TxRollback();
}

}

下面是针对速度测试图,发布一万条消息。 第一次我使用了事务控制方式。耗时915毫秒




第二次使用了confirm方式，耗时374毫秒




第三次没有使用消息确认机制，耗时317毫秒