说说 RabbiMQ 的应答模式

oec2003

发布于 2021-01-12 11:24:51

4790

发布于 2021-01-12 11:24:51

文章被收录于专栏：不止dotNET

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 的值很关键，并且需要做好异常处理，防止堵塞。

希望本文对您有所帮助！

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2021-01-11，如有侵权请联系 cloudcommunity@tencent.com 删除

rabbitmq

消息队列 CMQ 版

kafka

本文分享自不止dotNET 微信公众号，前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体同步曝光计划，欢迎热爱写作的你一起参与！

rabbitmq

消息队列 CMQ 版

kafka

登录后参与评论

0 条评论

热度