RabbitMQ 消息确认机制

Posted 2020-08-25 liangshiwei

消息确认机制

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

 

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)

事务控制方式 (tx)

但是要知道这些都是耗费性能的，其中事务的性能消耗最大，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毫秒