正确关闭ActiveMQ和Spring DefaultMessageListenerContainer [英] Properly Shutting Down ActiveMQ and Spring DefaultMessageListenerContainer

问题描述

当从Tomcat管理器发出停止命令时，我们的系统不会关闭。我已经确定它与ActiveMQ / Spring有关。我什至想出了如何关闭它，但是我的解决方案是破解（至少我希望这不是正确的方法）。我想知道关闭ActiveMQ的正确方法，以便可以删除我的黑客。

Our system will not shutdown when a "Stop" command is issued from the Tomcat Manager. I have determined that it is related to ActiveMQ/Spring. I have even figured out how to get it to shutdown, however my solution is a hack (at least I hope this isn't the "correct" way to do it). I would like to know the proper way to shutdown ActiveMQ so that I can remove my hack.

我继承了此组件，在经过大量挖掘之后，我不知道为何做出某些架构决策，我认为我理解他的想法，但我可能会遗漏一些东西。换句话说，真正的问题可能在于我们尝试使用ActiveMQ / Spring的方式。

I inherited this component and I have no information about why certain architectural decisions were made, after a lot of digging I think I understand his thoughts, but I could be missing something. In other words, the real problem could be in the way that we are trying to use ActiveMQ/Spring.

我们在ServletContainer（Tomcat 6/7）中运行，并使用ActiveMQ 5.9.1和Spring 3.0.0，可以在一个组中运行应用程序的多个实例，每个在其自己的服务器上运行的实例。 ActiveMQ用于促进多个实例之间的通信。每个实例都有自己的嵌入式代理和队列。每个实例上的每个队列都恰好有1个org.springframework.jms.listener.DefaultMessageListenerContainer监听，所以5个队列= 5个DefaultMessageListenerContainers。

We run in ServletContainer (Tomcat 6/7) and use ActiveMQ 5.9.1 and Spring 3.0.0 Multiple instances of our application can run in a "group", with each instance running on it's own server. ActiveMQ is used to facilitate communication between the multiple instances. Each instance has it's own embedded broker and it's own set of queues. Every queue on every instance has exactly 1 org.springframework.jms.listener.DefaultMessageListenerContainer listening to it, so 5 queues = 5 DefaultMessageListenerContainers for example.

我们的系统正常关闭了直到我们通过在ConnectionFactory中添加queuePrefetch = 0来修复错误为止。起初我以为这种更改在某种程度上是不正确的，但是现在我了解了这种情况，我确信我们不应该使用预取功能。

Our system shut down properly until we fixed a bug by adding queuePrefetch="0" to the ConnectionFactory. At first I assumed that this change was incorrect in some way, but now that I understand the situation, I am confident that we should not be using the prefetch functionality.

I已经创建了一个测试应用程序来复制该问题。请注意，以下信息没有提及消息产生者。那是因为我可以复制问题而无需发送/处理单个消息。只需在启动过程中创建Broker，ConnectionFactory，Queues和Listeners，就足以防止系统正常停止。

I have created a test application to replicate the issue. Note that the information below makes no mention of message producers. That is because I can replicate the issue without ever sending/processing a single message. Simply creating the Broker, ConnectionFactory, Queues and Listeners during boot, is enough to keep the system from stopping properly.

这是我的Spring XML示例配置。如果有人愿意，我将很乐意提供我的整个项目：

Here is my sample configuration from my Spring XML. I will be happy to provide my entire project if someone wants it:

<amq:broker persistent="false" id="mybroker"> 
 <amq:transportConnectors> 
  <amq:transportConnector uri="tcp://0.0.0.0:61616"/> 
 </amq:transportConnectors> 
</amq:broker> 

<amq:connectionFactory id="ConnectionFactory" brokerURL="vm://localhost?broker.persistent=false" > 
 <amq:prefetchPolicy> 
  <amq:prefetchPolicy queuePrefetch="0"/> 
 </amq:prefetchPolicy> 
</amq:connectionFactory> 

<amq:queue id="lookup.mdb.queue.cat" physicalName="DogQueue"/> 
<amq:queue id="lookup.mdb.queue.dog" physicalName="CatQueue"/> 
<amq:queue id="lookup.mdb.queue.fish" physicalName="FishQueue"/> 

<bean id="messageListener" class="org.springframework.jms.listener.DefaultMessageListenerContainer" abstract="true"> 
 <property name="connectionFactory" ref="ConnectionFactory"/> 
</bean> 

<bean parent="messageListener" id="cat"> 
 <property name="destination" ref="lookup.mdb.queue.dog"/> 
 <property name="messageListener"> 
  <bean class="com.acteksoft.common.remote.jms.WorkerMessageListener"/> 
 </property> 
 <property name="concurrentConsumers" value="200"/> 
 <property name="maxConcurrentConsumers" value="200"/> 
</bean> 

<bean parent="messageListener" id="dog"> 
 <property name="destination" ref="lookup.mdb.queue.cat"/> 
 <property name="messageListener"> 
  <bean class="com.acteksoft.common.remote.jms.WorkerMessageListener"/> 
 </property> 
 <property name="concurrentConsumers" value="200"/> 
 <property name="maxConcurrentConsumers" value="200"/> 
</bean> 

<bean parent="messageListener" id="fish"> 
 <property name="destination" ref="lookup.mdb.queue.fish"/> 
 <property name="messageListener"> 
  <bean class="com.acteksoft.common.remote.jms.WorkerMessageListener"/> 
 </property> 
 <property name="concurrentConsumers" value="200"/> 
 <property name="maxConcurrentConsumers" value="200"/> 
</bean> 

我的黑客涉及使用ServletContextListener手动停止对象。不足之处是我必须创建其他线程来停止DefaultMessageListenerContainers。也许我以错误的顺序停止了对象，但是我已经尝试了所有我能想象的。如果我尝试停止主线程中的对象，则它们将无限期地挂起。

My hack involves using a ServletContextListener to manually stop the objects. The hacky part is that I have to create additional threads to stop the DefaultMessageListenerContainers. Perhaps I'm stopping the objects in the wrong order, but I've tried everything that I can imagine. If I attempt to stop the objects in the main thread, then they hang indefinitely.

谢谢！

更新
我已经根据boday的建议尝试了以下操作，但没有成功。我还尝试将amq：transportConnector uri指定为tcp：//0.0.0.0：61616？transport.daemon = true

UPDATE I have tried the following based on boday's recommendation but it didn't work. I have also tried to specify the amq:transportConnector uri as tcp://0.0.0.0:61616?transport.daemon=true

  <amq:broker persistent="false" id="mybroker" brokerName="localhost">
   <amq:transportConnectors>
    <amq:transportConnector uri="tcp://0.0.0.0:61616?daemon=true"/>
   </amq:transportConnectors>
  </amq:broker>

  <amq:connectionFactory id="connectionFactory" brokerURL="vm://localhost" >
   <amq:prefetchPolicy>
    <amq:prefetchPolicy queuePrefetch="0"/>
   </amq:prefetchPolicy>
  </amq:connectionFactory>

有一次我试图在amq：connectionFactory元素和关闭正常，但是经过进一步测试，我了解到这些属性导致从VMTransportFactory抛出异常。这导致初始化不正确，并且基本消息功能不起作用。

At one point I tried to add similar properties to the brokerUrl parameter in the amq:connectionFactory element and the shutdown worked properly, however after further testing I learned that the properties were resulting in an exception to be thrown from VMTransportFactory. This resulted in improper initialization and the basic message functionality didn't work.

推荐答案

如果有人想知道的话，我可以看到使用ActiveMQ不可能有一个守护进程ListenerContainer。

In case anyone else is wondering, as far as I can see it's not possible to have a daemon ListenerContainer using ActiveMQ.

启动ActiveMQConnection时，它将创建一个带有非守护进程线程的ThreadPoolExecutor。看来这是为了避免从一个经纪人到另一个经纪人之间的连接故障切换时出现问题。

When the ActiveMQConnection is started, it creates a ThreadPoolExecutor with non-daemon thread. This is seemingly to avoid issues when failing over the connection from one broker to another.

https://issues.apache.org/jira/browse/AMQ-796

executor = new ThreadPoolExecutor(1, 1, 5, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(), new ThreadFactory() {
    @Override
    public Thread newThread(Runnable r) {
        Thread thread = new Thread(r, "ActiveMQ Connection Executor: " + transport);
        //Don't make these daemon threads - see https://issues.apache.org/jira/browse/AMQ-796
        //thread.setDaemon(true);
        return thread;
    }
});

这篇关于正确关闭ActiveMQ和Spring DefaultMessageListenerContainer的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！