经过 maxInactivityDuration 设置后 ActiveMQ InactivityMonitor 未关闭

Question

我正在我的应用程序中实现 ActiveMQ。我有一个用例，当远程侦听队列中没有消息时，我需要关闭 activeMQ 连接，并且在有消息时必须重新打开连接。我知道仅设置 maxInactivityDuration=30000 不会帮助我实现此用例，但至少在经过 30 秒超时设置后它不会关闭连接。请让我知道这方面出了什么问题，并让我知道如何实现我的用例。谢谢你的解决方案。

@Configuration
@EnableJms
@ImportResource("classpath*:beans.xml")
public class MessagingConfiguration {   


    @Value("${activemq.broker.url}")
    private String BROKER_URL = "tcp://localhost:61616?wireFormat.maxInactivityDuration=30000";

    @Value("${activemq.request.queue}")
    private String REQUEST_QUEUE = "test.request";

    @Value("${activemq.response.queue}")
    private String RESPONSE_QUEUE = "test.response";

    @Value("${activemq.borker.username}")
    private String BROKER_USERNAME = "admin";

    @Value("${activemq.borker.password}")
    private String BROKER_PASSWORD = "admin"; 

    @Autowired
    ListenerClass messageListener;

    @Autowired
    JmsExceptionListener jmsExceptionListener;

    @Bean
    public ActiveMQConnectionFactory connectionFactory() {
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
        connectionFactory.setBrokerURL(BROKER_URL);
        connectionFactory.setUserName(BROKER_USERNAME);
        connectionFactory.setPassword(BROKER_PASSWORD);
        connectionFactory.setTrustAllPackages(true);
        connectionFactory.setRedeliveryPolicy(redeliveryPolicy());
        return connectionFactory;
    }

    @Bean
    public RedeliveryPolicy redeliveryPolicy() {
        RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy();
        redeliveryPolicy.setBackOffMultiplier(3); // Wait 5 seconds first re-delivery, then 15, 45 seconds
        redeliveryPolicy.setInitialRedeliveryDelay(5000);
        redeliveryPolicy.setMaximumRedeliveries(3);
        redeliveryPolicy.setUseExponentialBackOff(true);
        return redeliveryPolicy;
    }

    @Bean
    public ConnectionFactory cachingConnectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setTargetConnectionFactory(connectionFactory());
        connectionFactory.setExceptionListener(jmsExceptionListener);
        connectionFactory.setSessionCacheSize(100);
        connectionFactory.setCacheConsumers(false);
        connectionFactory.setCacheProducers(false);
        return connectionFactory;
    }

    @Bean
    public JmsTemplate jmsTemplate() {
        JmsTemplate template = new JmsTemplate();
        template.setConnectionFactory(connectionFactory());
        template.setDefaultDestinationName(REQUEST_QUEUE);
        template.setSessionAcknowledgeModeName("CLIENT_ACKNOWLEDGE");
        template.setMessageConverter(converter());
        return template;
    }

    @Bean
    public DefaultMessageListenerContainer jmsListenerContainer() {
        DefaultMessageListenerContainer factory = new DefaultMessageListenerContainer();
        factory.setConnectionFactory(connectionFactory());
        factory.setConcurrency("1-1");
        factory.setDestinationName(RESPONSE_QUEUE);
        factory.setMessageListener(messageListener);
        factory.setExceptionListener(jmsExceptionListener);
        return factory;
    }

    @Bean
    MessageConverter converter() {
        MappingJackson2MessageConverter converter = new MappingJackson2MessageConverter();
        converter.setTargetType(MessageType.TEXT);
        converter.setTypeIdPropertyName("_type");
        return converter;
    }

}

控制台日志：

Aug 20, 2017 5:24:47 PM org.apache.catalina.startup.Catalina start
INFO: Server startup in 61255 ms
17:24:52.216 [ActiveMQ InactivityMonitor WriteCheckTimer] DEBUG o.a.a.t.AbstractInactivityMonitor - WriteChecker: 10001ms elapsed since last write check.
17:24:52.216 [ActiveMQ InactivityMonitor Worker] DEBUG o.a.a.t.AbstractInactivityMonitor - Running WriteCheck[tcp://127.0.0.1:61616]
17:24:53.145 [ActiveMQ InactivityMonitor WriteCheckTimer] DEBUG o.a.a.t.AbstractInactivityMonitor - WriteChecker: 10000ms elapsed since last write check.
17:24:53.145 [ActiveMQ InactivityMonitor Worker] DEBUG o.a.a.t.AbstractInactivityMonitor - Running WriteCheck[tcp://127.0.0.1:61616]
17:24:56.134 [ActiveMQ InactivityMonitor WriteCheckTimer] DEBUG o.a.a.t.AbstractInactivityMonitor - WriteChecker: 10001ms elapsed since last write check.
17:24:56.134 [ActiveMQ InactivityMonitor Worker] DEBUG o.a.a.t.AbstractInactivityMonitor - Running WriteCheck[tcp://127.0.0.1:61616]
17:24:56.336 [ActiveMQ InactivityMonitor WriteCheckTimer] DEBUG o.a.a.t.AbstractInactivityMonitor - WriteChecker: 10000ms elapsed since last write check.
17:24:56.336 [ActiveMQ InactivityMonitor Worker] DEBUG o.a.a.t.AbstractInactivityMonitor - Running WriteCheck[tcp://127.0.0.1:61616]
17:25:00.252 [ActiveMQ InactivityMonitor WriteCheckTimer] DEBUG o.a.a.t.AbstractInactivityMonitor - WriteChecker: 10000ms elapsed since last write check.
17:25:00.257 [ActiveMQ InactivityMonitor Worker] DEBUG o.a.a.t.AbstractInactivityMonitor - Running WriteCheck[tcp://127.0.0.1:61616]
17:25:02.217 [ActiveMQ InactivityMonitor WriteCheckTimer] DEBUG o.a.a.t.AbstractInactivityMonitor - WriteChecker: 10001ms elapsed since last write check.
17:25:02.217 [ActiveMQ InactivityMonitor Worker] DEBUG o.a.a.t.AbstractInactivityMonitor - Running WriteCheck[tcp://127.0.0.1:61616]
17:25:03.145 [ActiveMQ InactivityMonitor ReadCheckTimer] DEBUG o.a.a.t.AbstractInactivityMonitor - 30001ms elapsed since last read check.
17:25:03.146 [ActiveMQ InactivityMonitor WriteCheckTimer] DEBUG o.a.a.t.AbstractInactivityMonitor - WriteChecker: 10001ms elapsed since last write check.
17:25:03.146 [ActiveMQ InactivityMonitor Worker] DEBUG o.a.a.t.AbstractInactivityMonitor - Running WriteCheck[tcp://127.0.0.1:61616]
17:25:03.262 [ActiveMQ InactivityMonitor WriteCheckTimer] DEBUG o.a.a.t.AbstractInactivityMonitor - WriteChecker: 10000ms elapsed since last write check.
17:25:03.262 [ActiveMQ InactivityMonitor Worker] DEBUG o.a.a.t.AbstractInactivityMonitor - Running WriteCheck[tcp://127.0.0.1:61616]
17:25:06.133 [ActiveMQ InactivityMonitor ReadCheckTimer] DEBUG o.a.a.t.AbstractInactivityMonitor - 30001ms elapsed since last read check.
17:25:06.135 [ActiveMQ InactivityMonitor WriteCheckTimer] DEBUG o.a.a.t.AbstractInactivityMonitor - WriteChecker: 10001ms elapsed since last write check.
17:25:06.135 [ActiveMQ InactivityMonitor Worker] DEBUG o.a.a.t.AbstractInactivityMonitor - Running WriteCheck[tcp://127.0.0.1:61616]
17:25:06.337 [ActiveMQ InactivityMonitor WriteCheckTimer] DEBUG o.a.a.t.AbstractInactivityMonitor - WriteChecker: 10001ms elapsed since last write check.
17:25:06.337 [ActiveMQ InactivityMonitor Worker] DEBUG o.a.a.t.AbstractInactivityMonitor - Running WriteCheck[tcp://127.0.0.1:61616]
17:25:10.253 [ActiveMQ InactivityMonitor WriteCheckTimer] DEBUG o.a.a.t.AbstractInactivityMonitor - WriteChecker: 10001ms elapsed since last write check.
17:25:10.253 [ActiveMQ InactivityMonitor Worker] DEBUG o.a.a.t.AbstractInactivityMonitor - Running WriteCheck[tcp://127.0.0.1:61616]

Answer 1

看起来根据您的代码，您已经为连接注册了一个侦听器，因此它不会处于空闲状态，因此不会被关闭。

但是如果你想强制关闭连接，那么你可能需要为你的连接设置 maxTimeout。