Java RabbitMQ基于basicConsume消费指定数量消息的实现方案

完全可以通过basicConsume实现固定条数消息的移动，不需要循环调用basicGet，核心逻辑是在消费者中新增计数逻辑，达到目标条数后主动停止消费即可。

实现思路 #

• 在消费者内部维护原子计数器，避免异步回调场景下的计数误差
• 每成功处理1条消息（发布到目标队列+ack原消息）后计数器加1
• 计数器达到预设阈值（如10）时，主动取消当前消费者，后续消息会保留在原队列中
• 配合已设置的basicQos(10)，可以控制RabbitMQ最多预推送10条消息到消费者，避免多余消息被拉取

修改后可运行代码 #

import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.atomic.AtomicInteger;

public class FixedCountShovel {
    public static void main(String[] args) {
        // 定义需要移动的消息数量
        final int MOVE_COUNT = 10;
        // 原子计数器，适配异步回调的多线程场景
        final AtomicInteger consumedCount = new AtomicInteger(0);

        try {
            ConnectionFactory factory = new ConnectionFactory();
            // 请补充你的RabbitMQ连接配置
            // factory.setHost("xxx");
            // factory.setUsername("xxx");
            // factory.setPassword("xxx");
            
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            // 预拉取10条消息，避免多余消息被推送
            channel.basicQos(MOVE_COUNT);

            final DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag,
                                           Envelope envelope,
                                           AMQP.BasicProperties properties,
                                           byte[] body)
                    throws IOException {
                    long deliveryTag = envelope.getDeliveryTag();
                    // 将消息发布到目标交换器
                    channel.basicPublish("my-exchange", "my-routing-key", null, body);
                    // 确认原队列消息已消费
                    channel.basicAck(deliveryTag, false);
                    
                    // 计数+1，达到阈值后停止消费
                    int currentCount = consumedCount.incrementAndGet();
                    if (currentCount >= MOVE_COUNT) {
                        // 主动取消消费者，不再接收新消息
                        channel.basicCancel(consumerTag);
                        // 优雅关闭资源
                        channel.close();
                        connection.close();
                    }
                }
            };

            channel.basicConsume(
                "my-queue",
                false,
                "my-consumer-tag",
                defaultConsumer);

        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

注意事项 #

• 如果需要严格保证消息不丢失，可以开启发布确认机制，等目标队列确认消息接收成功后，再ack原队列的消息
• 如果运行过程中出现异常，未ack的消息会自动退回原队列，不会出现消息丢失
• 若原队列消息数量不足预设的移动条数，消费者会消费完所有存在的消息后保持运行，你可以根据需求新增超时逻辑，超过一定时间没收到新消息就主动关闭资源

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