中断 JMS MessageConsumer.receive() 后如何终止“ActiveMQ 传输”线程答案

【问题标题】：How do I kill "ActiveMQ Transport" threads after interrupting JMS MessageConsumer.receive()中断 JMS MessageConsumer.receive() 后如何终止“ActiveMQ 传输”线程
【发布时间】：2011-10-06 11:55:56
【问题描述】：

我有一个使用 ActiveMQConnectionFactory.createConnection() 获取 ActiveMQ 连接的类。然后它创建并拥有会话、目的地和在该目的地上侦听的消费者（队列）。

我在消费者上调用 receive() 或 receive(millis) 以将我的消息从队列中取出。在某些情况下，我必须终止（中断）调用接收方法的线程。之后我尝试关闭会话和连接。不幸的是，我在调用 close 时经常遇到异常，并且相关的“ActiveMQ 传输”线程（以及与代理的相关连接）保持活动状态。我得到的例外是

org.myorg.awsiface.communication.MessagingException: Failed to close JMS connection
at
org.myorg.aws.communication.transport.JMSMessageTransport.cleanUp(JMSMessageTransport.java:253)
at
org.myorg.aws.communication.protocol.ContextFinishedProtocol.cleanUp(ContextFinishedProtocol.java:94)
at org.myorg.myservice.job.Job.run(Job.java:206)
Caused by: javax.jms.JMSException: java.io.InterruptedIOExceptio)
    at org.apache.activemq.util.JMSExceptionSupport.create(JMSExceptionSupport.java:62)
at org.apache.activemq.ActiveMQConnection.doAsyncSendPacket(ActiveMQConnection.java:1227)
at org.apache.activemq.ActiveMQConnection.asyncSendPacket(ActiveMQConnection.java:1219)
at org.apache.activemq.ActiveMQSession.asyncSendPacket(ActiveMQSession.java:1799)
at org.apache.activemq.ActiveMQMessageConsumer.doClose(ActiveMQMessageConsumer.java:636)
at org.apache.activemq.ActiveMQMessageConsumer.close(ActiveMQMessageConsumer.java:627)
at org.myorg.aws.communication.transport.JMSMessageTransport.cleanUp(JMSMessageTransport.java:232)
... 2 more
Caused by: java.io.InterruptedIOException
at org.apache.activemq.transport.WireFormatNegotiator.oneway(WireFormatNegotiator.java:102)
at org.apache.activemq.transport.MutexTransport.oneway(MutexTransport.java:40)
at org.apache.activemq.transport.ResponseCorrelator.oneway(ResponseCorrelator.java:60)
at org.apache.activemq.ActiveMQConnection.doAsyncSendPacket(ActiveMQConnection.java:1225)
... 7 more

我尝试了以下连接 URL 故障转移：//tcp://broker.ip.address:61616 tcp://broker.ip.address:61616 failover://tcp://broker.ip.address:61616?trace=true&closeAsync=false

我也尝试过使用 MessageConsumer::receiveNoWait() 调用并只执行我自己的 Thread.sleep，但每当我调用以下命令关闭连接时，我总是会遇到上述异常

try {
    // I added the following two lines to see if the taskRunner was handling the threads - still no luck
    TaskRunnerFactory taskRunner = ((ActiveMQConnection)connection).getSessionTaskRunner();
    taskRunner.shutdown();

    if (producer != null) {
        producer.close();
    }
    if (consumer != null) {
        consumer.close();
    }
    session.close();
    if (connection instanceof ActiveMQConnection) {
        ActiveMQConnection amqConnection = (ActiveMQConnection) connection;
        amqConnection.stop();
    }           
    connection.stop();
    connection.close();
}
catch (ConnectionClosedException e) {
    // NOOP - this is ok
}
catch (Exception e) {
    throw new MessagingException("Failed to close JMS connection", e, log);
}

【问题讨论】：

标签： java connection activemq

【解决方案1】：

一种解决方案可能是在 TCP ActiveMQ 传输线程上设置守护程序标志，例如tcp://URI:PORT?daemon=true

查看 ActiveMQ 文档：http://activemq.apache.org/tcp-transport-reference.html

【讨论】：

对我不起作用。按照其他答案使用.close()。

【解决方案2】：

我不确定为什么关闭不适合您。但是你可以通过使用异步消息监听器来完成你想要的。

Connection connection = connectionFactory.createConnection();
connection.start();

Session session = 
    connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

Destination queue = session.createQueue("queueName");

Consumer consumer = session.createConsumer( queue );

consumer.setMessageListener( new MessageListener()
{
    public void onMessage( Message message )
    { 
         // whatever was after the receive() method call
    }
} );

那么就不需要被打断了。完成后，关闭所有内容：

consumer.close();
session.close();
queue.close();
connection.close();

【讨论】：