Spring Boot RabbitMQ 单队列多消费者实现及消息处理问题咨询

一、多消费者配置修改 #

你现有代码只需调整SimpleMessageListenerContainer的配置即可实现多消费者，无需改动其他核心逻辑：

修改配置类的容器Bean #

在原有container方法中新增并发消费者、预取计数相关配置即可，修改后代码如下：

@Bean
SimpleMessageListenerContainer container(ConnectionFactory connectionFactory
        , MessageListenerAdapter messageListenerAdapter){
    SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
    container.setConnectionFactory(connectionFactory);
    container.setQueueNames(QUEUE_NAME);
    container.setMessageListener(messageListenerAdapter);
    // 设置初始并发消费者数量，可根据业务需求调整
    container.setConcurrentConsumers(3);
    // 设置最大并发消费者数量，可根据服务器性能上限调整
    container.setMaxConcurrentConsumers(10);
    // 公平分发配置：每个消费者一次仅预取1条消息，处理完成后才会接收下一条
    container.setPrefetchCount(1);
    // 可选配置：开启手动消息确认，避免业务处理失败时消息丢失
    // container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
    return container;
}

可选简化配置：使用@RabbitListener注解 #

如果你不想手动声明SimpleMessageListenerContainer，也可以直接在消息处理方法上加注解实现：

@Service
public class Receive {
    // concurrency参数直接指定并发消费者范围，效果和上面手动配置容器一致
    @RabbitListener(queues = ConfigureRabbitMq.QUEUE_NAME, concurrency = "3-10")
    public void handleMessage(String messageBody){
        // 打印线程名即可验证多消费者生效
        System.out.println("消费者线程【" + Thread.currentThread().getName() + "】处理消息：" + messageBody);
    }
}

二、队列分发逻辑说明 #

RabbitMQ单队列对多消费者的分发策略默认有两种：

• 轮询分发（默认策略）：默认不配置prefetchCount时，RabbitMQ会按顺序把消息平均分配给所有消费者，不管消费者当前的负载情况，比如2个消费者的场景下，奇数序号消息给第一个消费者，偶数序号消息给第二个消费者，容易出现负载不均的情况。
• 公平分发：配置prefetchCount = 1后生效，RabbitMQ只会给已经处理完上一条消息、返回ACK的消费者分发新消息，避免出现部分消费者消息堆积、部分消费者空闲的问题，适合消息处理耗时差异较大的场景。

注意事项 #

如果开启手动消息确认，需要在业务处理成功后手动调用ACK方法，处理失败时调用NACK方法决定是否重新入队，避免消息丢失或者无限重复消费。

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