如何在tomcat上优雅地关闭ActiveMQ会话线程

Question

我们如何优雅地关闭守护进程线程[ActiveMQ会话:ID:PC-63704-1472105244157-1:1:1]？

<properties>
    <activemq.version>5.13.0</activemq.version>
</properties>

<dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-client</artifactId>
</dependency>
<dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-pool</artifactId>
</dependency>
<dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-jms-pool</artifactId>
</dependency>

spring-boot activeMQ设置如下所示

spring.activemq.mqtt.broker-url=tcp://127.0.0.1:1883
spring.activemq.broker-url=tcp://127.0.0.1:61616
spring.activemq.in-memory=false
spring.activemq.pooled=true

对于发送消息，我们只是简单地autowired JmsTemplate并发送消息出去：

@Autowired
private JmsTemplate jmsTemplate;

MessageCreator messageCreator = new MessageCreator() {
    @Override
    public Message createMessage(Session session) throws JMSException {
        // return Message
    }
};

jmsTemplate.send(new ActiveMQTopic(destinationName), messageCreator);

对于接收（监听）消息，我们使用Spring DefaultMessageListenerContainer(DMLC)

@Bean(name = { "afterCreateQueue" })
public Queue afterCreateQueue() {
    return new ActiveMQQueue(properties.getId() + "_After_Create_Queue");
}

@Bean
public DefaultMessageListenerContainer afterCreateJmsListenerContainer(ConnectionFactory connectionFactory)
        throws KeyManagementException, NoSuchAlgorithmException, KeyStoreException {
    final DefaultMessageListenerContainer defaultMessageListenerContainer = new DefaultMessageListenerContainer();
    defaultMessageListenerContainer.setConnectionFactory(connectionFactory);
    defaultMessageListenerContainer.setDestination(afterCreateQueue());

    final MessageListenerAdapter listener = new MessageListenerAdapter(afterCreateMessageDelegate());
    listener.setDefaultListenerMethod("handleMessage");

    defaultMessageListenerContainer.setMessageListener(listener);
    return defaultMessageListenerContainer;
}

22-Aug-2016 17:23:32.870 WARNING [localhost-startStop-2] org.apache.catalina.loader.WebappClassLoaderBase.clearReferencesThreads The web application [demo] appears to have started a thread named [ActiveMQ Session: ID:PC-63704-1472105244157-1:1:1] but has failed to stop it. This is very likely to create a memory leak. Stack trace of thread:
java.lang.Object.wait(Native Method)
java.lang.Object.wait(Object.java:502)
org.apache.activemq.thread.DedicatedTaskRunner.runTask(DedicatedTaskRunner.java:119)
org.apache.activemq.thread.DedicatedTaskRunner$1.run(DedicatedTaskRunner.java:42)
22-Aug-2016 17:23:32.872 WARNING [localhost-startStop-2] org.apache.catalina.loader.WebappClassLoaderBase.clearReferencesThreads The web application [demo] appears to have started a thread named [ActiveMQ InactivityMonitor Worker] but has failed to stop it. This is very likely to create a memory leak. Stack trace of thread:
sun.misc.Unsafe.park(Native Method)
java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
java.util.concurrent.SynchronousQueue$TransferStack.awaitFulfill(SynchronousQueue.java:460)
java.util.concurrent.SynchronousQueue$TransferStack.transfer(SynchronousQueue.java:362)
java.util.concurrent.SynchronousQueue.poll(SynchronousQueue.java:941)
java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1066)
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
java.lang.Thread.run(Thread.java:745)

org.apache.catalina.startup.Bootstrap at localhost:63675    
Thread [main] (Running) 
Daemon Thread [NioBlockingSelector.BlockPoller-1] (Running) 
Daemon Thread [NioBlockingSelector.BlockPoller-2] (Running) 
Daemon Thread [NioBlockingSelector.BlockPoller-3] (Running) 
Daemon Thread [AsyncLogger-1] (Running) 
Thread [ActiveMQ Session: ID:PC-63704-1472105244157-1:1:1] (Running)    
Daemon Thread [ContainerBackgroundProcessor[StandardEngine[Catalina]]] (Running)    
Daemon Thread [http-nio-1010-ClientPoller-0] (Running)  
Daemon Thread [http-nio-1010-ClientPoller-1] (Running)  
Daemon Thread [http-nio-1010-Acceptor-0] (Running)  
Daemon Thread [http-nio-1443-ClientPoller-0] (Running)  
Daemon Thread [http-nio-1443-ClientPoller-1] (Running)  
Daemon Thread [http-nio-1443-Acceptor-0] (Running)  
Daemon Thread [ajp-nio-8019-ClientPoller-0] (Running)   
Daemon Thread [ajp-nio-8019-ClientPoller-1] (Running)   
Daemon Thread [ajp-nio-8019-Acceptor-0] (Running)   
Daemon Thread [http-nio-1010-exec-1] (Running)  

Thread [ActiveMQ Session: ID:MoboLink-63704-1472105244157-1:1:1] (Suspended)    
waiting for: Object  (id=96)    
Object.wait(long) line: not available [native method]   
Object.wait() line: 502 
DedicatedTaskRunner.runTask() line: 119 
DedicatedTaskRunner$1.run() line: 42

@Autowired
private DefaultMessageListenerContainer afterCreateJmsListenerContainer;
@Autowired
private PooledConnectionFactory pooledJmsConnectionFactory;

@EventListener
private void onContextClosed(ContextClosedEvent event)
{
    try
    {
        if (afterCreateJmsListenerContainer != null) {
            afterCreateJmsListenerContainer.shutdown();
        }

        if (pooledJmsConnectionFactory != null) {
            pooledJmsConnectionFactory.clear();
        }
    } catch (Exception e)
    {
        logger.error("Exception", e);
    }
}

我们还有其他的设置吗？

Answer 1

请参见DefaultMessageListenerContainer中的receiveTimeout设置，如果是-1，侦听器线程将被阻塞，直到它得到消息，否则每个超时周期它将检查消费者是否被停止。默认值为1000ms。源代码

如果在指定的重试之后没有收到消息，则该设置还将停止该特定侦听器，其他设置可用于控制活动侦听器的数量。看这里和这里。