AsyncRabbitTemplate引发channelMax超限问题及配置咨询

依赖：
spring-boot-starter-amqp:3.1.1

使用AsyncRabbitTemplate时，它持续创建新Channel，且空闲Channel无法立即关闭，触发了**"The channelMax limit is reached"**错误。相关报错日志及Spring Boot下的RabbitMQ配置如下，求解决方法？

报错日志 #

at java.base/java.lang.Thread.run(Thread.java:833)
2023-10-26 06:16:52.826 WARN  [org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#5-4] [/] o.s.a.r.l.DirectReplyToMessageListenerContainer - basicConsume failed, scheduling consumer for queue amq.rabbitmq.reply-to for restart
org.springframework.amqp.AmqpResourceNotAvailableException: The channelMax limit is reached. Try later.

我的RabbitMQ配置 #

@Configuration
@EnableRabbit
public class RabbitConfig {


    @Value("${spring.rabbitmq.listener.simple.retry.initial-interval}")
    private int retryInitialInterval;

    @Value("${spring.rabbitmq.listener.simple.retry.max-attempts}")
    private int retryMaxAttempts;

    @Value("${spring.rabbitmq.listener.simple.retry.multiplier}")
    private double retryMultiplier;

    @Value("${spring.rabbitmq.listener.simple.retry.max-interval}")
    private int retryMaxInterval;

    @Value("${spring.rabbitmq.listener.simple.retry.max-consumers}")
    private int maxConsumers;

    @Value("${spring.rabbitmq.listener.simple.retry.concurrent-consumers}")
    private int concurrentConsumers;

    @Value("${spring.rabbitmq.host}")
    private String host;

    @Value("${spring.rabbitmq.username}")
    private String username;

    @Value("${spring.rabbitmq.password}")
    private String password;

    @Bean
    public ConnectionFactory customConnectionFactory(){
        final CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
        cachingConnectionFactory.setHost(host);
        cachingConnectionFactory.setUsername(username);
        cachingConnectionFactory.setPassword(password);
        return cachingConnectionFactory;

    }


    @Bean
    public MessageConverter jsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    @Bean
    public AsyncRabbitTemplate asyncRabbitTemplate(RabbitTemplate customRabbitTemplate ) {
        return new AsyncRabbitTemplate(customRabbitTemplate);
    }
    @Bean
    public RabbitTemplate customRabbitTemplate(ConnectionFactory customConnectionFactory) {
        final RabbitTemplate rabbitTemplate = new RabbitTemplate(customConnectionFactory);
        rabbitTemplate.setMessageConverter(jsonMessageConverter());
        return rabbitTemplate;
    }

    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(final ConnectionFactory customConnectionFactory) {


        final SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(customConnectionFactory);
        factory.setMessageConverter(jsonMessageConverter());
        factory.setConcurrentConsumers(concurrentConsumers);
        factory.setMaxConcurrentConsumers(maxConsumers);
        factory.setAdviceChain(messageRetryInterceptor());
        return factory;
    }


    @Bean
    public RetryOperationsInterceptor messageRetryInterceptor(){
        return RetryInterceptorBuilder.StatelessRetryInterceptorBuilder
                .stateless()
                .maxAttempts(retryMaxAttempts)
//                .backOffOptions(
//                        retryInitialInterval,
//                        retryMultiplier,
//                        retryMaxInterval
//                )
                .recoverer(new RejectAndDontRequeueRecoverer())
                .build();
    }


}

原因分析 #

AsyncRabbitTemplate默认采用Direct Reply-To机制，会为每个异步请求创建临时消费者；而CachingConnectionFactory默认的Channel缓存配置无法及时回收空闲Channel，结合RabbitMQ的channelMax连接限制，就会触发该错误。

具体解决方案 #

1. 调整CachingConnectionFactory的Channel缓存参数
   在customConnectionFactory()中添加Channel缓存配置，设置空闲Channel超时时间和缓存大小，加速空闲资源回收：

@Bean
public ConnectionFactory customConnectionFactory(){
    final CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
    cachingConnectionFactory.setHost(host);
    cachingConnectionFactory.setUsername(username);
    cachingConnectionFactory.setPassword(password);
    // 设置每个连接的Channel缓存上限，根据并发量调整
    cachingConnectionFactory.setChannelCacheSize(50);
    // 设置Channel checkout超时时间，超时后自动关闭空闲Channel（单位：毫秒）
    cachingConnectionFactory.setChannelCheckoutTimeout(10000);
    // 明确指定缓存模式为CHANNEL，优化Channel复用逻辑
    cachingConnectionFactory.setCacheMode(CachingConnectionFactory.CacheMode.CHANNEL);
    return cachingConnectionFactory;
}

2. 配置AsyncRabbitTemplate的回复消费者复用策略
   自定义AsyncRabbitTemplate时，设置回复队列的消费者并发参数，避免频繁创建新Channel：

@Bean
public AsyncRabbitTemplate asyncRabbitTemplate(RabbitTemplate customRabbitTemplate ) {
    AsyncRabbitTemplate asyncTemplate = new AsyncRabbitTemplate(customRabbitTemplate);
    // 设置回复消费者的基础并发数
    asyncTemplate.setConcurrentConsumers(5);
    // 设置回复消费者的最大并发数
    asyncTemplate.setMaxConcurrentConsumers(20);
    // 开启自动启动，复用现有消费者实例
    asyncTemplate.setAutoStartup(true);
    return asyncTemplate;
}

3. 增大RabbitMQ服务器的channelMax参数（可选）
   如果上述配置仍无法满足业务并发需求，可以修改RabbitMQ服务器的rabbitmq.conf文件，增大channelMax值：

channel_max = 1000

注意：修改后需重启RabbitMQ生效，且过大的Channel数量会增加服务器资源消耗，需根据实际场景评估调整。

4. 使用固定回复队列替代Direct Reply-To
   自定义固定回复队列，让所有异步请求复用同一队列的消费者，从根源减少Channel创建频率：

@Bean
public Queue replyQueue() {
    // 创建持久化的固定回复队列
    return new Queue("async.reply.queue", true);
}

@Bean
public AsyncRabbitTemplate asyncRabbitTemplate(RabbitTemplate customRabbitTemplate, Queue replyQueue) {
    RabbitTemplate replyTemplate = new RabbitTemplate(customRabbitTemplate.getConnectionFactory());
    replyTemplate.setMessageConverter(jsonMessageConverter());
    // 指定固定回复队列，复用消费者
    AsyncRabbitTemplate asyncTemplate = new AsyncRabbitTemplate(customRabbitTemplate, replyTemplate, replyQueue.getName());
    asyncTemplate.setConcurrentConsumers(5);
    return asyncTemplate;
}

验证建议 #

修改配置后，通过RabbitMQ控制台监控Channels指标，观察空闲Channel是否能及时回收；同时检查应用日志，确认channelMax limit is reached错误是否消失。

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