如何配置JmsListenerEndpointRegistry Bean并控制JMS监听器启停

核心问题在于你只声明了JmsListenerEndpointRegistry Bean但未关联实际的消息监听容器，导致无法通过它调用启停方法。以下是完整的配置与实现步骤：

1. 配置JMS监听容器工厂 #

先创建JmsListenerContainerFactory，用于生成消息监听容器，可根据需求选择DefaultJmsListenerContainerFactory（支持动态调整并发、事务）或SimpleJmsListenerContainerFactory：

@Bean
public JmsListenerContainerFactory<?> jmsListenerContainerFactory(ConnectionFactory connectionFactory) {
    DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory);
    // 配置并发数、事务等自定义属性
    factory.setConcurrency("1-3");
    factory.setSessionTransacted(true);
    return factory;
}

2. 配置JmsListenerEndpointRegistry并注册监听端点 #

手动创建MethodJmsListenerEndpoint（对应方法级的消息监听），并将其注册到JmsListenerEndpointRegistry，确保registry持有容器引用：

@Bean
public JmsListenerEndpointRegistry jmsListenerEndpointRegistry() {
    return new JmsListenerEndpointRegistry();
}

@Bean
public MethodJmsListenerEndpoint myJmsConsumerEndpoint(JmsListenerContainerFactory<?> containerFactory, MyMessageListener messageListener) {
    MethodJmsListenerEndpoint endpoint = new MethodJmsListenerEndpoint();
    // 设置唯一ID，用于后续定位容器
    endpoint.setId("myJmsConsumer");
    // 指定监听的JMS目的地（队列/主题）
    endpoint.setDestination("order-queue");
    // 绑定自定义消息监听器
    endpoint.setMessageListener(messageListener);
    endpoint.setContainerFactory(containerFactory);
    
    // 将端点注册到registry，第三个参数false表示不自动启动容器（按需启停）
    jmsListenerEndpointRegistry().registerListenerContainer(endpoint, containerFactory, false);
    return endpoint;
}

3. 自定义消息监听器 #

实现具体的消息处理逻辑：

@Component
public class MyMessageListener {
    public void onMessage(Message message) throws JMSException {
        TextMessage textMessage = (TextMessage) message;
        System.out.println("处理消息：" + textMessage.getText());
        // 业务逻辑处理
    }
}

4. 实现消费者启停控制 #

通过注入的JmsListenerEndpointRegistry获取容器，调用start()/stop()方法实现按需启停：

@Component
public class JmsConsumerController {
    private final JmsListenerEndpointRegistry registry;

    public JmsConsumerController(JmsListenerEndpointRegistry registry) {
        this.registry = registry;
    }

    // 根据标志启停消费者
    public void controlConsumer(boolean startConsumer) {
        MessageListenerContainer container = registry.getListenerContainer("myJmsConsumer");
        if (container == null) {
            throw new IllegalArgumentException("未找到ID为myJmsConsumer的消息容器");
        }
        
        if (startConsumer && !container.isRunning()) {
            container.start();
            System.out.println("JMS消费者已启动");
        } else if (!startConsumer && container.isRunning()) {
            container.stop();
            System.out.println("JMS消费者已停止");
        }
    }
}

关键注意点 #

• 必须为每个监听端点设置唯一ID，否则无法通过registry准确获取对应的容器。
• 注册端点时第三个参数设为false，避免容器自动启动，实现按需启停。
• 启停前判断容器运行状态，避免重复操作引发异常。

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