Azure Service Bus会话异常关闭未恢复，配置并发时问题凸显求解决

在Spring Boot集成Microsoft Azure Service Bus时，配置JMS监听的concurrency参数后，出现监听会话突然关闭、连接无法自动恢复的情况，且无明确错误码，异常日志如下：

javax.jms.IllegalStateException: The Session is closed
    at org.apache.qpid.jms.JmsSession.checkClosed(JmsSession.java:1101) ~[qpid-jms-client-0.53.0.jar!/:na]
    at org.apache.qpid.jms.JmsSession.rollback(JmsSession.java:263) ~[qpid-jms-client-0.53.0.jar!/:na]
    at org.messaginghub.pooled.jms.JmsPoolSession.close(JmsPoolSession.java:112) ~[pooled-jms-1.2.4.jar!/:na]
    at org.springframework.jms.support.JmsUtils.closeSession(JmsUtils.java:109) ~[spring-jms-5.3.27.jar!/:5.3.27]
    at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.clearResources(DefaultMessageListenerContainer.java:1289) ~[spring-jms-5.3.27.jar!/:5.3.27]
    at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.run(DefaultMessageListenerContainer.java:1135) ~[spring-jms-5.3.27.jar!/:5.3.27]
    at java.base/java.lang.Thread.run(Unknown Source) ~[na:na]

2023-10-03 21:41:40.521  WARN 1 --- [ontainer#0-4019] o.s.j.l.DefaultMessageListenerContainer  : Setup of JMS message listener invoker failed for destination 'carton-status-update' - trying to recover. Cause: The MessageConsumer is closed
2023-10-03 21:41:40.525  WARN 1 --- [ontainer#0-4019] o.m.pooled.jms.JmsPoolSession            : Caught exception trying rollback() when putting session back into the pool, will invalidate. javax.jms.IllegalStateException: The Session is closed

javax.jms.IllegalStateException: The Session is closed
    at org.apache.qpid.jms.JmsSession.checkClosed(JmsSession.java:1101) ~[qpid-jms-client-0.53.0.jar!/:na]
    at org.apache.qpid.jms.JmsSession.rollback(JmsSession.java:263) ~[qpid-jms-client-0.53.0.jar!/:na]
    at org.messaginghub.pooled.jms.JmsPoolSession.close(JmsPoolSession.java:112) ~[pooled-jms-1.2.4.jar!/:na]
    at org.messaginghub.pooled.jms.JmsPoolConnection.cleanupAllLoanedSessions(JmsPoolConnection.java:269) ~[pooled-jms-1.2.4.jar!/:na]
    at org.messaginghub.pooled.jms.JmsPoolConnection.close(JmsPoolConnection.java:80) ~[pooled-jms-1.2.4.jar!/:na]
    at org.springframework.jms.connection.ConnectionFactoryUtils.releaseConnection(ConnectionFactoryUtils.java:81) ~[spring-jms-5.3.27.jar!/:5.3.27]
    at org.springframework.jms.listener.AbstractJmsListeningContainer.refreshSharedConnection(AbstractJmsListeningContainer.java:394) ~[spring-jms-5.3.27.jar!/:5.3.27]
    at org.springframework.jms.listener.DefaultMessageListenerContainer.refreshConnectionUntilSuccessful(DefaultMessageListenerContainer.java:985) ~[spring-jms-5.3.27.jar!/:5.3.27]
    at org.springframework.jms.listener.DefaultMessageListenerContainer.recoverAfterListenerSetupFailure(DefaultMessageListenerContainer.java:959) ~[spring-jms-5.3.27.jar!/:5.3.27]
    at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.run(DefaultMessageListenerContainer.java:1146) ~[spring-jms-5.3.27.jar!/:5.3.27]
    at java.base/java.lang.Thread.run(Unknown Source) ~[na:na] 

当前使用的JMS监听注解配置：

@JmsListener(destination = TOPIC_NAME, containerFactory = "topicJmsListenerContainerFactory", subscription = SUBSCRIPTION_NAME, concurrency="15-20")

解决方案 #

1. 调整连接池配置，避免会话回收时的rollback异常 #

配置JmsPoolConnectionFactory时，关闭会话归还时的自动rollback操作，防止因已关闭会话触发异常导致连接池失效：

@Bean
public JmsPoolConnectionFactory pooledConnectionFactory(AzureServiceBusJmsConnectionFactory azureConnectionFactory) {
    JmsPoolConnectionFactory pool = new JmsPoolConnectionFactory();
    pool.setConnectionFactory(azureConnectionFactory);
    pool.setRollbackOnReturn(false); // 核心配置：禁用归还会话时的rollback
    pool.setMaxConnections(20); // 匹配并发上限
    pool.setMaxSessionsPerConnection(1); // 每个连接绑定一个会话，避免会话池交叉影响
    return pool;
}

2. 配置监听器容器的主动恢复策略 #

针对DefaultJmsListenerContainerFactory，显式设置恢复间隔和缓存策略，确保异常后能快速重建会话：

@Bean
public DefaultJmsListenerContainerFactory topicJmsListenerContainerFactory(JmsPoolConnectionFactory pooledConnectionFactory) {
    DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
    factory.setConnectionFactory(pooledConnectionFactory);
    factory.setConcurrency("15-20");
    factory.setRecoveryInterval(5000); // 异常后每5秒尝试恢复连接
    factory.setCacheLevel(DefaultMessageListenerContainer.CACHE_CONSUMER); // 缓存消费者，减少重建开销
    factory.setSessionTransacted(false); // 无需事务时关闭，避免不必要的rollback操作
    return factory;
}

3. 增强Azure Service Bus客户端的连接稳定性 #

在配置文件中添加客户端心跳和超时配置，提前检测并回收无效连接：

spring:
  jms:
    servicebus:
      connection-string: ${AZURE_SERVICEBUS_CONNECTION_STRING}
      idle-timeout: 300000 # 5分钟空闲超时，自动回收闲置连接
      client-id: ${spring.application.name} # 标识客户端，便于Azure端排查

4. 自定义异常处理，主动触发连接刷新 #

实现JmsListenerErrorHandler，当检测到会话关闭异常时，主动触发监听器容器的连接刷新：

@Component
public class CustomJmsErrorHandler implements JmsListenerErrorHandler {
    @Autowired
    private ApplicationContext context;

    @Override
    public Object handleError(Message message, ListenerExecutionFailedException exception) throws Exception {
        // 识别会话关闭异常，主动刷新连接
        if (exception.getCause() instanceof IllegalStateException && 
            exception.getCause().getMessage().contains("The Session is closed")) {
            DefaultMessageListenerContainer container = context.getBean(DefaultMessageListenerContainer.class);
            container.refreshConnectionUntilSuccessful();
        }
        throw exception;
    }
}

在监听注解中绑定自定义错误处理器：

@JmsListener(destination = TOPIC_NAME, containerFactory = "topicJmsListenerContainerFactory", 
             subscription = SUBSCRIPTION_NAME, concurrency="15-20", errorHandler = "customJmsErrorHandler")

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