Java RabbitMQ如何实现处理完当前消息后再接收下一条队列消息

要实现「每次仅接收一条消息，处理完成后再接收下一条」的需求，需要同时配置两个核心参数：

• 关闭自动消息确认（autoAck）：你提供的代码里已经将basicConsume的第二个参数设为false，这一步是正确的，后续需要手动调用ack方法通知RabbitMQ消息已处理完成。
• 设置通道预取计数（prefetch count）为1：这是你当前代码缺失的核心配置，该配置会告诉RabbitMQ不要给当前消费者同时推送超过1条未确认的消息，只有等你返回确认通知之后才会推送下一条。

完整实现代码如下：

// 1. 创建通道、声明队列等前置操作保留原有逻辑即可
// 核心配置：设置预取数为1，每次仅拉取1条未确认消息
channel.basicQos(1);

DeliverCallback deliverCallback = new DeliverCallback() {
    @Override
    public void handle(String s, Delivery delivery) throws IOException {
        long deliveryTag = delivery.getEnvelope().getDeliveryTag();
        try {
            System.out.println("Tag: "+ deliveryTag);
            String message = new String(delivery.getBody(), "UTF-8");
            // 此处编写你的业务处理逻辑
            // ...
            // 业务处理成功后手动ack，通知RabbitMQ可以删除该消息并推送下一条
            // 第二个参数multiple设为false，代表仅确认当前这一条消息
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            // 业务处理失败时，根据需求选择对应处理逻辑
            // 方案1：消息重新入队等待重试，第三个参数requeue设为true
            channel.basicNack(deliveryTag, false, true);
            // 方案2：直接丢弃当前消息
            // channel.basicReject(deliveryTag, false);
        }
    }
};

// autoAck设为false，关闭自动确认
channel.basicConsume(QUEUE_NAME, false, deliverCallback, new CancelCallback() {
    @Override
    public void handle(String consumerTag) throws IOException {}
});

注意事项 #

如果缺失channel.basicQos(1)配置，即使关闭了自动ack，RabbitMQ还是会把队列中的消息尽可能多的推送到客户端本地缓存，还是会出现同时收到多条消息的情况。

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