EasyNetQ是否支持手动Ack？批量消费场景需求问询

问题描述 #

我需要用EasyNetQ实现消费者任务，从一个或多个RabbitMQ队列读取批量消息（比如100条）并在上游做批量处理。要求上游处理完成前不执行Ack，避免进程崩溃时消息丢失。

当前测试伪代码如下：

var bus = RabbitHutch.CreateBus("host=localhost");
var consumer = bus.Advanced.Consume(queue, (body, properties, info) => 
    {
        MessageHandler(body, properties, info);
    });

// ...

static void MessageHandler(ReadOnlyMemory<byte> content, MessageProperties props, MessageReceivedInfo info)
{
      messageBatch.Add((content, info));

      if (messageBatch.Count >= 100)
          // 触发其他线程执行批量处理
}

EasyNetQ默认会在bus.Advanced.Consume的处理函数返回后立即自动Ack，请问能否关闭自动Ack，若可以，后续该调用哪个函数手动发送Ack？

注：已有RabbitMQ.Client的相关方案，想了解EasyNetQ客户端的支持情况。

解决方案 #

1. 关闭自动Ack #

EasyNetQ支持关闭自动Ack，只需使用带AutoAck参数的Consume重载方法，将参数设为false即可。修改后的消费初始化代码：

var consumer = bus.Advanced.Consume(queue, autoAck: false, (body, properties, info) => 
{
    MessageHandler(body, properties, info);
});

2. 手动发送Ack #

批量处理完成后，可调用bus.Advanced.Ack方法手动确认消息，该方法需传入两个核心参数：

• deliveryTag：从MessageReceivedInfo.DeliveryTag获取，是当前消息的唯一标识
• multiple：设为true时会确认所有小于等于当前deliveryTag的未确认消息（适合批量Ack场景），设为false则仅确认当前单条消息

批量处理完成后的Ack示例：

// 假设messageBatch是当前待处理的消息集合
if (messageBatch.Count >= 100)
{
    // 执行上游批量处理逻辑
    ProcessBatch(messageBatch);
    
    // 批量确认所有已处理的消息
    var lastDeliveryTag = messageBatch.Last().info.DeliveryTag;
    bus.Advanced.Ack(lastDeliveryTag, multiple: true);
    
    // 清空批量集合，准备接收下一批消息
    messageBatch.Clear();
}

补充说明 #

• 若处理过程中需要拒绝消息，可使用bus.Advanced.Nack方法，参数逻辑与Ack类似，还可通过requeue参数决定是否将消息重新放回队列
• 需保证批量集合的线程安全性，避免多线程场景下的消息重复或丢失问题

内容的提问来源于stack exchange，提问作者M.M