C#从Rabbitmq每隔x间隔以y批量大小消费/读取消息答案

【问题标题】：C# Consuming / Reading message every x interval with y batch size From Rabbit MQC#从Rabbitmq每隔x间隔以y批量大小消费/读取消息
【发布时间】：2019-11-21 04:27:30
【问题描述】：

我使用 Rabbit MQ 按顺序处理消息。现在我有一个场景，不需要立即消费消息，但我想每 1(x) 分钟从 MQ 消费/读取消息，批量大小为 20(y)

所以我可以一次处理这 20 条消息并在一次调用中保存到数据库，而不是为每条消息调用 20 次。

那么如何在每 x 间隔批量接收/消费消息。

我看到了以下信息，Consume messages in batches - RabbitMQ 和 Rabbitmq retrieve multiple messages using single synchronous call using .NET

我试图实现第二个问题的实现（questions/32309155）但没有工作，不明白“consumer.Received”会收到 **_fetchSize ==> 20 ** 的意思，它会在单次读取中收到 20 条消息？或者它将如何工作，因为我将 fetchsize 更改为 10，但 consumer.received 正在接收单个消息。

using (var connection = factory.CreateConnection())
        {
            using (var channel = connection.CreateModel())
            {
                channel.BasicQos(0, 1, false);
                channel.ExchangeDeclare("helloExchange", type:"direct");
                channel.QueueDeclare(queue: "hello", durable: true, exclusive: false, autoDelete: false,
                    arguments: null);

                channel.QueueBind("hello", "helloExchange", routingKey:"hello");

                var consumer = new EventingBasicConsumer(channel);

                consumer.Received += (model, ea) =>
                {
                    bool canAck = false;
                    var retryCount = 0;



                    try
                    {
                        var body = ea.Body;
                        var message = Encoding.UTF8.GetString(body);
                       // DO PROCESS MESSAGE HERE
                        Console.WriteLine($"{typeof(MyConsumer).Name} Message consumed {message}");
                        canAck = true;
                    }
                    catch (Exception ex)
                    {canAck = false;
                      // LOG ERROR
                    }

                    try
                    {
                        if (canAck)
                        {
                            channel.BasicAck(ea.DeliveryTag, false);
                        }
                        else
                        {
                            channel.BasicNack(ea.DeliveryTag, false, false);
                        }
                    }
                    catch (AlreadyClosedException ex)
                    {
                        Console.WriteLine(ex.Message + " >> RabbitMQ is closed!");
                    }
                };
                channel.BasicConsume(queue: "hello", autoAck: false, consumer: consumer);

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

【问题讨论】：

标签： c# rabbitmq

【解决方案1】：

消费者一次只会收到一条消息。获取大小将决定有多少未确认的消息可以在通道上传递给消费者。

因此，以 3 的 fetch 大小为例，将传递 3 条消息，但如果您不确认任何消息，则意味着即使您收到新消息进入队列（第 4 条），它也不会发送到通道（因此不会被消耗），直到您至少确认 3 条初始消息中的一条。

在您的情况下，为了能够处理您的场景并批量使用消息，您可以执行以下操作：

将 fetchSize 设置为 20
每当您收到新消息时，您都会将其添加到列表中
一旦达到限制 (20)，您就开始处理它并确认批处理中的所有消息（从而清除列表）
批处理完成并确认所有消息后，您将开始接收新消息（然后再次启动流程）

即使这在技术上是可行的，我也不会采用这种实现，因为它会使处理重试/错误处理等变得更加复杂。使用消息队列，您可以选择使您的代码具有原子性、幂等性和弹性，这将导致更容易的错误/重试处理。

【讨论】：