博客
关于我
强烈建议你试试无所不能的chatGPT,快点击我
说说 RabbiMQ 的应答模式
阅读量:4033 次
发布时间:2019-05-24

本文共 4566 字,大约阅读时间需要 15 分钟。

RabbiMQ 我们都很熟悉了,是很常用的一个开源消息队列。搞懂 RabbiMQ 的应答模式对我们排查错误很有帮助,也能避免一些坑。本文说说 RabbiMQ 的应答模式。

生产者发出一条消息给 RabbiMQ ,RabbiMQ 将消息推送给消费者,消费者处理完消息后告诉 RabbiMQ,我已经接收到消息并处理了,RabbiMQ 收到通知后会将消息从队列中删除。消费者通知 MQ 的这个过程就是消息的应答。在 RabbiMQ 中有两种应答模式:自动应答和手动应答。

版本

  • dotNET Core :3.1

  • RabbitMQ:3.8.2

  • RabbitMQ.Client:6.2.1

自动应答

当 RabbiMQ 开启了消息的自动应答,一旦 RabbiMQ 将消息分发给了消费者,就会将消息从内存中删除。这种情况下,如果正在执行的消费者挂掉,就会丢失正在处理的消息。

生产者代码

static void Main(string[] args){    ConnectionFactory factory = new ConnectionFactory    {        UserName = "oec2003",        Password = "000000",        HostName = "10.211.55.6"    };    using (var connection = factory.CreateConnection())    using (var channel = connection.CreateModel())    {        Console.WriteLine("RabbitMQ连接成功,请输入消息,输入exit退出");        channel.QueueDeclare("oec2003", false, false, false, null);        string input;        do        {            input = Console.ReadLine();            var body = Encoding.UTF8.GetBytes(input);            channel.BasicPublish("", "oec2003", null, body);        }        while (input.Trim().ToLower() != "exit");    }}

消费者代码

static void Main(string[] args){    ConnectionFactory factory = new ConnectionFactory    {        UserName = "oec2003",        Password = "000000",        HostName = "10.211.55.6"    };    using (var connection = factory.CreateConnection())    using (var channel = connection.CreateModel())    {        Console.WriteLine("消费者开始监听......");        channel.QueueDeclare("oec2003", false, false, false, null);        EventingBasicConsumer consumer = new EventingBasicConsumer(channel);        consumer.Received += (ch, ea) =>        {            string message = Encoding.Default.GetString(ea.Body.ToArray());            Console.WriteLine($"{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")}:收到消息: {message}");            System.Threading.Thread.Sleep(10000);        };        channel.BasicConsume("oec2003", true, consumer);        Console.ReadKey();    }}
  • channel.BasicConsume 方法的第二个参数设置为 true 表示自动应答;

  • 开启自动应答后,消息是生产者发布后,当有消费者连接上后,所有的消息都会被自动确认,并且从内存中删除,这时如果消费者进程挂掉,没有处理的消息会丢失,正在处理中的消息也不会被重新投递;

  • 自动应答的好处是消息队列不会处于堵塞状态,但代价有点大,生产环境中还是不建议使用。

手动应答

手动应答,当消费者接收到消息处理完后,需要发送一个回执,告诉 RabbiMQ 服务端,这时 RabbiMQ 才会将该消息删除。

生产者的代码和上面的一样,消费者代码需要做相关调整,如下:

static void Main(string[] args){    ConnectionFactory factory = new ConnectionFactory    {        UserName = "oec2003",        Password = "000000",        HostName = "10.211.55.6"    };    using (var connection = factory.CreateConnection())    using (var channel = connection.CreateModel())    {        Console.WriteLine("消费者开始监听......");        EventingBasicConsumer consumer = new EventingBasicConsumer(channel);        consumer.Received += (ch, ea) =>        {            string message = Encoding.Default.GetString(ea.Body.ToArray());            Console.WriteLine($"{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")}:收到消息: {message}");                       channel.BasicAck(ea.DeliveryTag, false);        };        channel.BasicConsume("oec2003", false, consumer);        Console.ReadKey();    }}
  • channel.BasicConsume 方法的第二个参数设置为 false ,表示手动应答模式;

  • 在处理完消息后调用 channel.BasicAck(ea.DeliveryTag, false); 来进行应答,告诉 RabbiMQ 消息已经收到,RabbiMQ 收到这个回执后,才会删除消息。

可能遇到的问题

流量控制问题

在手动模式下,生产者发送消息后消息会从 Ready 进入到 Unacked 中,当消费者进行应答之后消息从 Unacked 中删除。

如果消息的产生速度远远大于消费者的处理速度,这时消息就会都在消费者处进行积压了。我们会看到 Unacked 中的数量会越来越大,这样消费者的压力就会越来越大,这时就需要使用 Qos 来进行限流。

Qos

在消费者中使用 channel.BasicQos(0, 2, false); 来进行 Qos 的设置,如下图:

BasicQos 方法有三个参数:

  • prefetchSize:批量获取消息的总大小,0为不限制;

  • prefetchCount:每次处理消息的个数,比如 prefetchCount 设置为 2 ,那么处于 Unacked 状态的消息最多就 2 条,当其中一条进行了得到了应答后,才会从 Ready 中转入一条到 Unacked

  • global:设置为 true 表示对 channel 进行控制,否则对每个消费者进行限制,一个 channel 可以有多个消费者

为什么使用 Qos :

  • 提高服务稳定性,因为有 prefetchCount 参数的控制,不会有海量的数据涌进来导致消费者服务挂掉;

  • 提高吞吐量,当队列有多个消费者时,每个消费者的能力不一样,我们可以通过 prefetchCount 参数来合理安排每个消费者的处理能力,不会出现有的空闲,有的积压。

prefetchCount 是一个非常关键的参数,当消费者处理消息时,出现一些异常情况,导致无法进行 Ack 应答,没有应答的数量大于等于 prefetchCount 时,队列就会发生堵塞。所以我们一定要确保消息的处理能够被异常捕获,并在 finally 中进行 Ack 应答,代码如下:

try{    string message = Encoding.Default.GetString(ea.Body.ToArray());    Console.WriteLine($"{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")}:收到消息: {message}");    if (message == "error")    {        throw new Exception("mq error");    }    else if (message == "sleep")    {        System.Threading.Thread.Sleep(60000);    }}catch (Exception){    //处理异常}finally{    channel.BasicAck(ea.DeliveryTag, false);}

一旦队列堵塞了,一种处理方式就是断掉客户端,这样,处在 Unacked 中的消息会重新回到 Ready 中,会重新进行投递进行消费。

总结

1、自动应答模式需要慎用,特别是生产环境;

2、不开启 Qos ,消费者可能会面临很大压力,但消息不会堵塞(测试过 500 个未进行 Ack 没有造成堵塞),现在不确定在没有 Qos 的情况下,有没有默认的最大 prefetchCount ;

3、开启 Qos ,prefetchCount 的值很关键,并且需要做好异常处理,防止堵塞。

希望本文对您有所帮助!

转载地址:http://hbudi.baihongyu.com/

你可能感兴趣的文章
[LeetCode By Python]9. Palindrome Number
查看>>
[leetCode By Python] 14. Longest Common Prefix
查看>>
[leetCode By Python]111. Minimum Depth of Binary Tree
查看>>
[LeetCode By Python]118. Pascal's Triangle
查看>>
[LeetCode By Python]121. Best Time to Buy and Sell Stock
查看>>
[LeetCode By Python]122. Best Time to Buy and Sell Stock II
查看>>
[LeetCode By Python]125. Valid Palindrome
查看>>
[LeetCode By Python]136. Single Number
查看>>
[LeetCode By Python]172. Factorial Trailing Zeroes
查看>>
[LeetCode By MYSQL] Combine Two Tables
查看>>
python jieba分词模块的基本用法
查看>>
[CCF BY C++]2017.12 最小差值
查看>>
[CCF BY C++]2017-12 游戏
查看>>
如何打开ipynb文件
查看>>
[Leetcode BY python ]190. Reverse Bits
查看>>
Android下调用收发短信邮件等(转载)
查看>>
Android中电池信息(Battery information)的取得
查看>>
SVN客户端命令详解
查看>>
Android/Linux 内存监视
查看>>
Linux系统信息查看
查看>>