迁移至Spring RabbitMQ后CachingConnectionFactory出现消费者取消/通道关闭错误

可能的根因 #

• 默认的CachingConnectionFactory连接池、通道池容量太小，高并发场景下资源耗尽，导致RabbitMQ强制关闭通道
• 默认心跳配置（60秒）与RabbitMQ服务器的超时设置不匹配，连接被服务器主动回收
• CamelDirectMessageListenerContainer的消费者监控机制在高负载下无法及时恢复失效的消费者

具体解决方案 #

1. 扩容连接池与通道池 #

高并发环境下，默认的连接数（1）和通道数（25）完全不够用，需要根据业务并发量调整：

@Bean
public CachingConnectionFactory connectionFactory() {
    CachingConnectionFactory factory = new CachingConnectionFactory("rabbitmq-server-host");
    factory.setUsername("your-username");
    factory.setPassword("your-password");
    // 设置连接池大小，建议根据实例CPU核数或并发请求数设置，比如10-20
    factory.setConnectionCacheSize(15);
    // 设置每个连接的通道缓存上限，高并发场景可设为50-100
    factory.setChannelCacheSize(80);
    // 添加通道获取超时，避免线程无限等待闲置通道
    factory.setChannelCheckoutTimeout(5000); // 5秒超时后抛出异常，触发业务降级或重试
    return factory;
}

2. 同步心跳与超时配置 #

确保客户端心跳与RabbitMQ服务器的heartbeat_timeout参数一致（服务器默认是60秒，建议调至30秒减少连接回收风险），同时配置合理的连接超时：

@Bean
public CachingConnectionFactory connectionFactory() {
    CachingConnectionFactory factory = new CachingConnectionFactory();
    // 设置心跳间隔，单位：秒
    factory.setRequestedHeartBeat(30);
    // 连接超时时间，单位：毫秒
    factory.setConnectionTimeout(10000);
    // 关闭连接的超时时间，单位：毫秒
    factory.setShutdownTimeout(60000);
    return factory;
}

3. 优化消费者容器配置 #

调整Camel的消费者容器参数，提升消费能力和自动恢复能力：

@Bean
public CamelDirectMessageListenerContainer testQueueListenerContainer(ConnectionFactory connectionFactory) {
    CamelDirectMessageListenerContainer container = new CamelDirectMessageListenerContainer();
    container.setConnectionFactory(connectionFactory);
    container.setQueueNames("test.queue");
    // 初始消费者数量
    container.setConcurrentConsumers(10);
    // 最大消费者数量，根据队列消息堆积情况自动扩容
    container.setMaxConcurrentConsumers(20);
    // 消费者失效后重试恢复的间隔，单位：毫秒
    container.setRecoveryInterval(5000);
    // 自动启动消费者
    container.setAutoStartup(true);
    return container;
}

如果是通过Camel路由配置，可直接在路由中指定：

from("spring-rabbitmq:test.queue")
    .listenerContainer()
    .concurrentConsumers(10)
    .maxConcurrentConsumers(20)
    .recoveryInterval(5000)
    .end()
    .process(yourMessageProcessor);

4. 开启监控排查瓶颈 #

• 开启Spring AMQP的DEBUG级日志，跟踪连接、通道的创建、销毁过程，定位资源泄漏点
• 查看RabbitMQ管理控制台的Connections和Channels页面，监控资源使用情况，确认是否存在资源耗尽的情况

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