多SimpleMessageListenerContainer引发Consumer启动失败问题排查

问题背景 #

需求如下：

• 将多个队列绑定至同一x-consistent-hash交换机
• 每个队列配置独立监听器以保证部分有序性

启动应用时出现以下错误，且应用启动停滞耗时730秒，未启动所有消费者：

ERROR 4828 --- [           main] o.s.a.r.l.SimpleMessageListenerContainer : Consumer failed to start in 60000 milliseconds; does the task executor have enough threads to support the container concurrency?

配置代码 #

@Configuration
public class RabbitMqConfig {

    @Bean
    public ConnectionFactory connectionFactory() {
        com.rabbitmq.client.ConnectionFactory cf = new com.rabbitmq.client.ConnectionFactory();
        cf.setHost("localhost");
        cf.setPort(5672);
        cf.setUsername("***");
        cf.setPassword("***");
        cf.setVirtualHost("/");
        cf.setAutomaticRecoveryEnabled(false);
        return new PooledChannelConnectionFactory(cf);
    }

    @Bean
    public RabbitTemplate feedRabbitTemplate(ConnectionFactory connectionFactory) {
        return new RabbitTemplate(connectionFactory);
    }

    @Bean
    public RabbitAdmin feedRabbitAdmin(RabbitTemplate feedRabbitTemplate) {
        return new RabbitAdmin(feedRabbitTemplate);
    }

    @Bean
    public Exchange myExchange(RabbitAdmin rabbitAdmin) {
        Map<String, Object> arguments = Map.of("hash-header", "hash-on");
        Exchange exchange = new CustomExchange("my.exchange", "x-consistent-hash", true, false, arguments);
        exchange.setAdminsThatShouldDeclare(rabbitAdmin);
        return exchange;
    }

    @Bean("myQueues")
    public List<Queue> myQueues(RabbitAdmin rabbitAdmin) {
        List<Queue> queues = new ArrayList<>();
        for (int i = 1; i <= 20; i++) {
            String queueName = String.join(".", "my.queue", Integer.toString(i));
            Queue queue = QueueBuilder.durable(queueName)
                    .singleActiveConsumer()
                    .build();

            queue.setAdminsThatShouldDeclare(rabbitAdmin);
            rabbitAdmin.declareQueue(queue);
            queues.add(queue);
        }
        return queues;
    }

    @Bean
    public List<Binding> myBindings(RabbitAdmin rabbitAdmin,
                                      @Qualifier("myQueues") List<Queue> queues,
                                      Exchange exchange) {
        List<Binding> bindings = new ArrayList<>();
        for (Queue queue : queues) {
            Binding binding = BindingBuilder.bind(queue)
                    .to(exchange)
                    .with("1")
                    .noargs();

            binding.setAdminsThatShouldDeclare(rabbitAdmin);
            rabbitAdmin.declareBinding(binding);
            bindings.add(binding);
        }
        return bindings;
    }

    @Bean
    public TaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(10);
        executor.setMaxPoolSize(10);
        executor.setQueueCapacity(50);
        executor.setThreadNamePrefix("rabbitmq-listener-");
        executor.initialize();
        return executor;
    }

    @Bean
    public List<SimpleMessageListenerContainer> feedListenerContainers(RabbitAdmin rabbitAdmin,
                                                                       @Qualifier("myQueues") List<Queue> queues,
                                                                       @Qualifier("jsonMessageConverter") Jackson2JsonMessageConverter converter,
                                                                       MyMessageHandler myMessageHandler,
                                                                       TaskExecutor taskExecutor) {
        ConnectionFactory connectionFactory = rabbitAdmin.getRabbitTemplate().getConnectionFactory();
        List<SimpleMessageListenerContainer> containers = new ArrayList<>();
        for (Queue queue : queues) {
            SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
            container.setAmqpAdmin(rabbitAdmin);
            container.setMessageListener(new MyMessageListenerAdapter(converter, myMessageHandler));
            container.setTaskExecutor(taskExecutor);
            container.addQueues(queue);
            containers.add(container);
            container.start();
        }
        return containers;
    }

    @Bean("jsonMessageConverter")
    public Jackson2JsonMessageConverter jackson2JsonMessageConverter(ObjectMapper objectMapper) {
        return new Jackson2JsonMessageConverter(objectMapper);
    }

    @Bean
    public ObjectMapper objectMapper() {
        JavaTimeModule module = new JavaTimeModule();
        ObjectMapper objectMapper = new ObjectMapper();
        objectMapper.registerModule(module);
        return objectMapper;
    }
}

问题根源 #

1. 线程池资源不足：配置的线程池核心/最大线程数仅为10，但需要启动20个独立的监听器容器（每个队列对应一个），每个容器启动至少需要1个线程，线程池无法同时满足所有容器的启动需求，导致部分容器启动超时。
2. 手动启动容器阻塞主线程：在Bean初始化阶段调用container.start()，会强制主线程等待消费者启动完成，当线程池资源不足时，后续容器启动排队等待，超过默认60秒的启动超时阈值。
3. 重复声明队列/绑定：已经通过setAdminsThatShouldDeclare配置了RabbitAdmin自动声明队列和绑定，手动调用rabbitAdmin.declareQueue/declareBinding属于冗余操作，可能引发不必要的资源竞争。
4. Single Active Consumer 线程需求：每个队列启用了singleActiveConsumer，意味着每个队列需要一个独立的线程处理消息，20个队列至少需要20个线程支撑。

解决方案 #

1. 调整线程池大小 #

将线程池的核心和最大线程数调整为至少等于队列数量（20），确保每个容器能分配到启动线程：

@Bean
public TaskExecutor taskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(20); // 匹配队列数量
    executor.setMaxPoolSize(20);
    executor.setQueueCapacity(100); // 适当增大任务队列容量
    executor.setThreadNamePrefix("rabbitmq-listener-");
    executor.initialize();
    return executor;
}

2. 移除手动启动容器的代码 #

Spring会自动管理SimpleMessageListenerContainer的生命周期，无需在Bean初始化时手动调用container.start()，避免主线程阻塞：

// 移除feedListenerContainers方法中的container.start();语句

3. 移除重复声明操作 #

删除手动调用rabbitAdmin.declareQueue和rabbitAdmin.declareBinding的代码，依赖RabbitAdmin的自动声明机制即可：

// 在myQueues方法中删除rabbitAdmin.declareQueue(queue);
// 在myBindings方法中删除rabbitAdmin.declareBinding(binding);

4. （可选）使用@RabbitListener简化配置 #

替代手动创建SimpleMessageListenerContainer，使用@RabbitListener注解让Spring自动管理容器，配置更简洁：

// 示例：为每个队列创建独立的监听方法
@Component
public class MyMessageListener {

    private final MyMessageHandler handler;

    public MyMessageListener(MyMessageHandler handler) {
        this.handler = handler;
    }

    @RabbitListener(queues = "#{myQueues.get(0)}")
    public void handleQueue1(Message message) {
        handler.process(message);
    }

    @RabbitListener(queues = "#{myQueues.get(1)}")
    public void handleQueue2(Message message) {
        handler.process(message);
    }

    // 依次为20个队列添加监听方法，或通过编程方式动态注册
}

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