如何使用 .NET RabbitMQ 客户端为消费者使用预取计数

Question

它在 RabbitMQ 文档中声明如下

“根据经验，在线程之间共享 Channel 实例是 要避免的事情。应用程序应该更喜欢使用通道 每个线程而不是跨多个共享相同的通道 线程。”

目前我们正在查看预取计数，建议如果您有少量消费者并且 autoack=false，那么我们应该一次消费许多消息。但是，我们发现如果消费者使用单个执行线程发回手动确认，则预取无效。但是，如果我们将消费者处理包装在一个任务中，我们会发现预取计数确实很重要，并且大大提高了消费者的性能。

请参阅以下示例，其中我们将消费者对消息的消费包装在 Task 对象中：

class Program
{
    public static void Main()
    {
        var factory = new ConnectionFactory()
        {
            HostName = "172.20.20.13",
            UserName = "billy",
            Password = "guest",
            Port = 5671,
            VirtualHost = "/",
            Ssl = new SslOption
            {
                Enabled = true,
                ServerName = "rabbit.blah.com",
                Version = System.Security.Authentication.SslProtocols.Tls12
            }
        };
        var connection = factory.CreateConnection();
        var channel = connection.CreateModel();
        channel.BasicQos(0, 100, false);
        channel.ExchangeDeclare(exchange: "logs", type: "fanout");
        var queueName = channel.QueueDeclare().QueueName;
        Console.WriteLine(" [*] Waiting for logs.");

        var consumer = new EventingBasicConsumer(channel);
        consumer.Received += (model, ea) =>
        {
            var _result = new Task(() => {
                var body = ea.Body;
                var message = Encoding.UTF8.GetString(body);
                System.Threading.Thread.Sleep(80);

                channel.BasicAck(ea.DeliveryTag, false);
            });
            _result.Start();
        };
        channel.BasicConsume(queue: "test.queue.1", autoAck: false, consumer: consumer);

        Console.WriteLine(" Press [enter] to exit.");
        Console.ReadLine();
    }
}

我的问题是人们如何使用 .NET rabbitmq 客户端实现消费者，利用预取计数？，你必须使用某种任务手动确认吗？，它安全吗

Answer 1

来源：https://www.rabbitmq.com/api-guide.html

使用手动确认时，重要的是要考虑 什么线程进行确认。如果它不同于 接收交付的线程（例如 Consumer#handleDelivery 将交付处理委托给不同的线程），确认 将多个参数设置为 true 是不安全的，将导致 双重确认，因此是通道级协议 关闭通道的异常。确认单个消息 时间可以是安全的。

channel.basicAck(tag, false) 是线程安全的

但consumerChannel.basicAck(tag, true) 不是。

RabbitMQ and channels Java thread safety也提到了一些优点

Answer 2

您所参考的文档适用于 Java 客户端。你应该指的是this document。

您正在使用最新版本的 .NET 客户端 (5.1)，因此在 Received 事件处理程序中执行您的工作不会阻塞处理 TCP 数据的其他线程，也不会阻塞心跳- 两者都很好。

首先，调用channel.BasicQos(0, 1, false) 意味着您的消费者一次只能从RabbitMQ 接收一条就绪消息，并且在调用BasicAck 之前不会传递另一条消息。所以，真的没有理由在另一个线程中做你的工作，因为无论如何你都不会收到另一条消息。

如果您增加预取值（通过实验和运行基准测试），如果您的工作运行时间超过几毫秒，您将不得不在后台线程中完成工作。

当您在Received 事件回调中执行您的工作时，它将阻塞用于进行该回调的线程，因为回调不是在其自己的线程上执行的。因此，您可以确保您的工作非常简短，或者在另一个线程中完成工作。

我刚刚花了一些时间查看 .NET 客户端代码，我很确定 IModel 实例不是线程安全的。如果增加预取，您将有机会同时确认多条消息，因此我建议实施一个使用它的解决方案，并确保在创建连接的同一线程上调用 BasicAck。

---

_{注意：RabbitMQ 团队会监控 rabbitmq-users mailing list，并且有时只回答 StackOverflow 上的问题。}