Spring Boot RabbitMQ：如何确保多消费者串行消费消息

嗨，我来帮你搞定这个需求！你之前用exclusive=true的方式虽然能保证同一时间只有一个消费者，但确实限制了扩展性，完全不符合你想要多个消费者轮流处理消息的预期。下面给你一个更合理的解决方案：

核心思路：利用消费者预取（Prefetch Count）+ 手动消息确认 #

RabbitMQ的预取计数可以控制每个消费者从队列中获取的未确认消息数量，配合手动确认机制，就能让多个消费者串行地轮流获取消息——每个消费者每次只处理一条消息，处理完并确认后才会拿到下一条，这样队列里的消息就会依次分配给各个在线消费者，正好符合你要的consumer1: message1 → message3、consumer2: message2 → message4的效果。

具体实现步骤 #

1. 配置RabbitMQ容器工厂 #

首先需要自定义一个容器工厂，设置预取计数为1，并开启手动确认模式：

@Configuration
public class RabbitMQConfig {

    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        // 每个消费者每次只预取1条消息，处理完确认后才会获取下一条
        factory.setPrefetchCount(1);
        // 开启手动确认模式，由代码控制消息的确认逻辑
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        return factory;
    }
}

2. 实现消费者逻辑 #

在消费者方法中，使用自定义的容器工厂，并在处理完消息后手动确认（如果处理失败可以选择拒绝或重新入队）：

@Component
public class SerialMessageConsumer {

    private static final Logger logger = LoggerFactory.getLogger(SerialMessageConsumer.class);

    // 指定使用我们自定义的容器工厂
    @RabbitListener(queues = Constants.QUEUE_TOPIC, containerFactory = "rabbitListenerContainerFactory")
    public void processMessage(Message message, Channel channel) throws IOException {
        try {
            String messageContent = new String(message.getBody(), StandardCharsets.UTF_8);
            logger.info("{} 正在处理消息: {}", Thread.currentThread().getName(), messageContent);
            
            // 这里替换成你的业务处理逻辑
            simulateBusinessProcessing();
            
            // 手动确认消息：第二个参数false表示只确认当前这条消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            logger.info("{} 完成消息处理: {}", Thread.currentThread().getName(), messageContent);
        } catch (Exception e) {
            logger.error("处理消息失败", e);
            // 如果处理失败，拒绝消息并重新入队（第三个参数true表示重新入队）
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
        }
    }

    private void simulateBusinessProcessing() throws InterruptedException {
        // 模拟业务处理耗时
        Thread.sleep(1000);
    }
}

为什么这个方案可行？ #

• 当prefetchCount=1时，RabbitMQ会确保每个消费者在确认完上一条消息前，不会收到新的消息。
• 多个消费者同时在线时，队列会将消息依次分配给空闲的消费者（处理完上一条并确认的消费者），自然形成你要的轮流串行处理的效果。
• 相比exclusive=true，这个方案支持横向扩展：你可以根据消息量增加消费者实例，提升整体处理能力，同时保持每个消费者串行处理自己的消息队列。

额外注意点 #

• 如果你的业务允许消息被并行处理，但只是希望每个消费者自身串行处理消息，这个方案同样适用。
• 手动确认机制能保证消息不丢失：如果某个消费者在处理过程中崩溃，未确认的消息会被RabbitMQ重新分配给其他在线消费者。

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