如何限制RabbitMQ消息接收量？实现单条消息消费处理

要实现每次仅消费并处理一条消息，核心是通过RabbitMQ的预取机制和手动消息确认来控制消息推送节奏，具体调整如下：

关键修改点 #

• 关闭自动消息确认（autoAck: false）：避免RabbitMQ一推送消息就标记为已消费，确保只有在你处理完消息后才确认。
• 设置预取计数（BasicQos）：限制RabbitMQ推送给当前消费者的未确认消息数量为1，这样只有处理完当前消息并确认后，才会收到下一条。

修改后的代码 #

public void Consume()
{
    if (_channel != null)
    {
        // 设置预取计数：每次只获取1条未确认消息
        _channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);

        var consumer = new EventingBasicConsumer(_channel);

        consumer.Received += (model, ea) =>
        {
            try
            {
                var body = ea.Body.ToArray();
                var message = Encoding.UTF8.GetString(body);
                var evt = new MessageEventArgs(body, message);

                // 处理消息
                OnMessageReceived(evt);

                // 手动确认消息：告知RabbitMQ消息已处理完成，可以推送下一条
                _channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
            }
            catch (Exception ex)
            {
                // 处理失败时的可选操作：拒绝消息并重新入队（或直接丢弃）
                _channel.BasicNack(deliveryTag: ea.DeliveryTag, multiple: false, requeue: true);
            }
        };

        // 关闭自动确认，改为手动确认
        _channel.BasicConsume(queue: _queue, autoAck: false, consumer: consumer);
    }
}

代码说明 #

• BasicQos的prefetchCount:1：告诉RabbitMQ，当前消费者最多保留1条未确认的消息，处理完并确认后才会再发新的。
• BasicAck：在消息处理完成后调用，确认消息已消费，RabbitMQ会从队列中移除该消息，并推送下一条。
• BasicNack（可选）：如果消息处理失败，可以调用这个方法拒绝消息，requeue:true会让消息重新回到队列，避免丢失。

内容的提问来源于stack exchange，提问作者Ian Jowett