Spring Boot RabbitListener：单队列绑单线程及多队列分线程处理方案

这个问题其实是Spring AMQP中消息监听容器的默认行为导致的——同一个@RabbitListener注解下的多个队列会被分配到同一个消息监听容器中，容器会用共享的线程池来处理所有队列的消息，所以才会出现两个队列共用线程的情况。下面给你几种可行的解决方案，从简单到进阶都有：

方案一：拆分监听方法（最简便直观） #

直接给每个队列单独写一个@RabbitListener监听方法，Spring会为每个监听方法创建独立的消息监听容器，默认情况下每个容器会使用专属的线程（除非你修改了并发配置）来处理对应队列的消息。

示例代码：

@RabbitListener(queues = "queue1")
public void receiverQueue1(String message) {
    logger.info("thread id:{}, message:{}", Thread.currentThread().getId(), message);
}

@RabbitListener(queues = "queue2")
public void receiverQueue2(String message) {
    logger.info("thread id:{}, message:{}", Thread.currentThread().getId(), message);
}

这样修改后，queue1和queue2的消息会在完全独立的线程中处理，你会看到日志里的线程ID完全不同。

方案二：自定义容器工厂（不拆分方法的选择） #

如果不想拆分监听方法，你可以为每个队列创建专属的消息监听容器工厂，指定每个工厂使用单线程处理消息，然后在@RabbitListener中分别绑定对应的工厂。

第一步：配置两个独立的容器工厂 #

@Bean
public SimpleRabbitListenerContainerFactory queue1ContainerFactory(ConnectionFactory connectionFactory) {
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory);
    factory.setConcurrentConsumers(1); // 强制每个容器只使用1个线程
    return factory;
}

@Bean
public SimpleRabbitListenerContainerFactory queue2ContainerFactory(ConnectionFactory connectionFactory) {
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory);
    factory.setConcurrentConsumers(1);
    return factory;
}

第二步：在监听方法中指定对应工厂 #

@RabbitListener(queues = "queue1", containerFactory = "queue1ContainerFactory")
@RabbitListener(queues = "queue2", containerFactory = "queue2ContainerFactory")
public void receiver(String message) {
    logger.info("thread id:{}, message:{}", Thread.currentThread().getId(), message);
}

这种方式可以保持单个监听方法，但配置相对繁琐，适合有特殊需求不想拆分方法的场景。

方案三：固定专属线程（进阶需求） #

如果需要确保每个队列始终使用固定的某一个线程（而不只是不同线程），可以为每个容器工厂配置专属的单线程执行器：

第一步：定义专属单线程执行器 #

@Bean
public Executor queue1Executor() {
    // 创建仅含1个线程的执行器，确保queue1的消息始终在这个线程处理
    return Executors.newSingleThreadExecutor(r -> new Thread(r, "queue1-executor-thread"));
}

@Bean
public Executor queue2Executor() {
    return Executors.newSingleThreadExecutor(r -> new Thread(r, "queue2-executor-thread"));
}

第二步：绑定到容器工厂 #

@Bean
public SimpleRabbitListenerContainerFactory queue1ContainerFactory(ConnectionFactory connectionFactory) {
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory);
    factory.setTaskExecutor(queue1Executor()); // 指定专属线程池
    return factory;
}

@Bean
public SimpleRabbitListenerContainerFactory queue2ContainerFactory(ConnectionFactory connectionFactory) {
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory);
    factory.setTaskExecutor(queue2Executor());
    return factory;
}

第三步：监听方法绑定工厂 #

和方案二的第二步一样，在@RabbitListener中指定对应的容器工厂即可。这种方式下，你甚至可以给线程命名，方便日志排查问题。

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