Spring Boot下动态修改@JMSListener的Destination方案咨询

首先得明确：默认情况下，@JMSListener的destination属性是在应用启动时就解析绑定到消息监听容器的，之后即使修改配置文件，这个值也不会自动更新——因为容器初始化后就固定了监听目标。不过我们有两种可行方案来实现动态修改，具体如下：

方案一：手动管理JMS监听端点（推荐，灵活性最高） #

这种方式通过JmsListenerEndpointRegistry动态创建、销毁监听端点，完全掌控目标队列的切换逻辑，适合需要频繁变更或复杂场景。

步骤1：改造原有监听类 #

先去掉@JMSListener注解，把消息处理方法改成普通的Spring Bean方法：

@Component("b2b.CCRReceiver")
@Slf4j
public class CCRReceiver {
    // 保留原有变量和依赖注入

    @Transactional
    public void receiveMessage(Message message) {
        // 原有消息处理逻辑不变
    }
}

步骤2：编写动态监听管理器 #

这个类负责创建新的监听端点、替换旧端点：

@Component
public class DynamicJmsListenerManager {

    @Autowired
    private JmsListenerEndpointRegistry registry;

    @Autowired
    private CCRReceiver ccrReceiver;

    @Autowired
    @Qualifier("jmsFactory")
    private ConnectionFactory jmsFactory;

    @Autowired
    private MessageConverter messageConverter;

    @Value("${jms.concurrency}")
    private String concurrency;

    // 标记要替换的端点ID，保持唯一
    private final String endpointId = "ccr-receiver-endpoint";

    public void updateQueueDestination(String newDestination) {
        // 停止并注销旧的监听容器
        MessageListenerContainer oldContainer = registry.getListenerContainer(endpointId);
        if (oldContainer != null) {
            oldContainer.stop();
            registry.unregisterListenerContainer(endpointId);
        }

        // 创建新的监听端点
        SimpleJmsListenerEndpoint newEndpoint = new SimpleJmsListenerEndpoint();
        newEndpoint.setId(endpointId);
        newEndpoint.setDestination(newDestination);
        newEndpoint.setMessageConverter(messageConverter);
        newEndpoint.setConcurrency(concurrency);
        // 绑定消息处理逻辑
        newEndpoint.setMessageListener(message -> {
            try {
                ccrReceiver.receiveMessage(message);
            } catch (Exception e) {
                TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
                log.error("JMS消息处理失败", e);
            }
        });

        // 注册并启动新容器
        registry.registerListenerContainer(newEndpoint, jmsFactory, true);
        registry.getListenerContainer(endpointId).start();
        log.info("JMS监听目标已更新为：{}", newDestination);
    }
}

步骤3：触发配置更新 #

当配置文件（或配置中心）的队列值变更时，调用上面的updateQueueDestination方法即可。比如结合Spring Cloud Config的@RefreshScope监听配置刷新事件：

@Component
@RefreshScope
public class ConfigRefreshHandler {

    @Value("${tibco.configuration.queues.upsert}")
    private String currentDestination;

    @Autowired
    private DynamicJmsListenerManager listenerManager;

    @Autowired
    private Environment environment;

    @EventListener(RefreshEvent.class)
    public void onConfigRefresh() {
        String newDestination = environment.getProperty("tibco.configuration.queues.upsert");
        if (!currentDestination.equals(newDestination)) {
            listenerManager.updateQueueDestination(newDestination);
            currentDestination = newDestination;
        }
    }
}

方案二：自定义动态DestinationResolver（侵入性低） #

这种方式不需要修改原有@JMSListener代码，通过自定义DestinationResolver让每次获取队列时都读取最新配置值。

步骤1：实现动态DestinationResolver #

@Component
public class DynamicConfigDestinationResolver extends DynamicDestinationResolver {

    @Autowired
    private Environment environment;

    @Override
    public Destination resolveDestinationName(Session session, String destinationName, boolean pubSubDomain) throws JMSException {
        // 每次解析时都从环境中获取最新的配置值
        String actualDestination = environment.resolveRequiredPlaceholders(destinationName);
        return super.resolveDestinationName(session, actualDestination, pubSubDomain);
    }
}

步骤2：配置JMS容器工厂 #

把自定义的解析器绑定到容器工厂：

@Configuration
public class JmsConfig {

    @Autowired
    private DynamicConfigDestinationResolver dynamicDestinationResolver;

    @Bean
    public JmsListenerContainerFactory<?> jmsFactory(ConnectionFactory connectionFactory) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        // 替换默认的DestinationResolver
        factory.setDestinationResolver(dynamicDestinationResolver);
        // 其他原有配置...
        return factory;
    }
}

注意事项 #

如果你的JMS容器开启了缓存（比如setCacheLevel(CACHE_CONSUMER)），容器会缓存Consumer对象，导致Destination不会重新解析。这种情况下需要监听配置变化，重启对应的监听容器：

@Component
@RefreshScope
public class ContainerRestartHandler {

    @Value("${tibco.configuration.queues.upsert}")
    private String currentDestination;

    @Autowired
    private JmsListenerEndpointRegistry registry;

    @Autowired
    private Environment environment;

    @EventListener(RefreshEvent.class)
    public void restartContainerOnConfigChange() {
        String newDestination = environment.getProperty("tibco.configuration.queues.upsert");
        if (!currentDestination.equals(newDestination)) {
            // 容器ID默认是「Bean名称.方法名」，也可以在@JMSListener中指定id属性
            MessageListenerContainer container = registry.getListenerContainer("b2b.CCRReceiver.receiveMessage");
            if (container != null) {
                container.stop();
                container.start();
            }
            currentDestination = newDestination;
            log.info("已重启JMS监听容器，目标队列更新为：{}", newDestination);
        }
    }
}

方案对比 #

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