java - 中断JMS MessageConsumer.receive()后如何杀死 "ActiveMQ Transport"线程

我有一个使用ActiveMQConnectionFactory.createConnection()获取ActiveMQ连接的类。
然后，它创建并拥有 session ，目的地和监听该目的地的使用者(队列)。

我在使用者上调用receive()或receive(millis)以使我的消息脱离队列。在某些情况下，我必须终止(中断)在其中调用receive方法的线程。之后，我尝试关闭 session 和连接。不幸的是，在调用close且关联的“ActiveMQ Transport”线程(以及与代理的相关连接)仍然处于 Activity 状态时，我总是不断遇到异常。我得到的异常(exception)是

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

我尝试了以下连接网址
故障转移://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);
}

最佳答案

我不确定为什么收盘对您不起作用。但是您可以通过使用异步消息监听器来完成所需的操作。

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();

关于java - 中断JMS MessageConsumer.receive()后如何杀死 "ActiveMQ Transport"线程，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/6713284/