首页 文章

正确关闭ActiveMQ和Spring DefaultMessageListenerContainer

提问于
浏览 次
5

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

我继承了这个组件，我没有关于为什么做出某些架构决策的信息，经过大量的挖掘后我觉得我理解他的想法，但我可能会遗漏一些东西 . 换句话说，真正的问题可能在于我们尝试使用ActiveMQ / Spring的方式 .

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

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

我创建了一个测试应用程序来复制问题 . 请注意，以下信息未提及消息生成者 . 这是因为我可以在不发送/处理单个消息的情况下复制问题 . 只需在引导期间创建Broker，ConnectionFactory，Queues和Listener就足以防止系统正常停止 .

这是我的Spring XML示例配置 . 如果有人想要，我会很乐意提供我的整个项目：

<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>

我的hack涉及使用ServletContextListener手动停止对象 . hacky部分是我必须创建额外的线程来停止DefaultMessageListenerContainers . 也许我正在以错误的顺序停止对象，但我已经尝试了我能想象到的一切 . 如果我试图阻止主线程中的对象，那么它们会无限期地挂起 .

先感谢您！

UPDATE 我已根据boday 's recommendation but it didn' t work尝试了以下内容 . 我还尝试将amq：transportConnector uri指定为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元素中添加类似的属性到brokerUrl参数，并且关闭工作正常，但是在进一步测试后我了解到这些属性导致从VMTransportFactory抛出异常 . 这导致初始化不正确，基本消息功能无法正常工作 .

activemq spring-jms

2 回答

  • 0

    尝试在TCP传输上设置 daemon=true ，这允许进程作为deamon线程运行，不会阻止容器的关闭

    见http://activemq.apache.org/tcp-transport-reference.html

    回复于
  • 1

    如果其他人想知道，据我所知，使用ActiveMQ不可能有一个守护进程ListenerContainer .

    启动ActiveMQConnection时，它会创建一个带有非守护程序线程的ThreadPoolExecutor . 这似乎是为了避免在从一个代理到另一个代理的连接失败时出现问题 .

    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;
    }
});
    回复于

相关问题