使用 MessageListener 的 JMS MessageConsumer 在 ActiveMQ 关闭时终止答案

【问题标题】：JMS MessageConsumer Using MessageListener Terminates on ActiveMQ Shutdown使用 MessageListener 的 JMS MessageConsumer 在 ActiveMQ 关闭时终止
【发布时间】：2014-10-23 00:43:23
【问题描述】：

尝试让 JMS MessageConsumer 在 ActiveMQ 重新启动后仍然存在，以便它可以使用故障转移传输协议重新连接。

但是，它会在 ActiveMQ 关闭时终止。

这看起来像是一个已报告并“解决”的错误，但我仍然在最新版本的 ActiveMQ 5.10.0 中看到此问题

我使用了以下maven依赖

    <dependency>
        <groupId>org.apache.activemq</groupId>
        <artifactId>activemq-all</artifactId>
        <version>5.10.0</version>
    </dependency>

这里是一些使用示例代码

public class SimpleConsumer {

public static void main(String[] args) throws Exception {
    String url = "failover:(tcp://ACTIVE_MQ_HOST:61616)";
    String destination = "test-topic";

    TopicConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
            url);

    ActiveMQConnection connection = (ActiveMQConnection) connectionFactory
            .createConnection();

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

    Topic topic = session.createTopic(destination);

    MessageConsumer consumer = session.createConsumer(topic);
    connection.start();

// Uncomment these lines and comment out the lines below and it will work
//      while (true) {
//          Message msg = consumer.receive();
//          if (msg instanceof TextMessage) {
//              System.out.println("msg received = " + msg);
//          }
//      }

    consumer.setMessageListener(new MessageListener() {

        public void onMessage(Message msg) {
            System.out.println("msg received = " + msg);
        }

    });

}

}

如果它是非阻塞和异步的，我希望它与 MessageListener 一起使用。

非常感谢任何帮助。

按照上面报道的 JIRA 的建议，我已经尝试过的一些方法是在非守护线程中运行它，但这不起作用。

我试过了

public class SimpleConsumerThread {

    public static void main(String[] args) throws Exception {


        Thread t = new Thread() {
            public void run() {
                try {               
                    String url = "failover:(tcp://ACTIVEMQ_HOST:61616)";
                    String destination = "test-topic";

                    TopicConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);

                    ActiveMQConnection connection = (ActiveMQConnection) connectionFactory.createConnection();

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

                    Topic topic = session.createTopic(destination);

                    MessageConsumer consumer = session.createConsumer(topic);
                    connection.start();
                    consumer.setMessageListener(new MessageListener() {

                        public void onMessage(Message msg) {
                            System.out.println("msg received = " + msg);
                        }

                    });
                } catch (JMSException e) {
                    e.printStackTrace();
                }               
            }
        };
        t.setDaemon(false);

        t.start();

    }

}

【问题讨论】：

标签： multithreading jms activemq failover message-listener

【解决方案1】：

您的线程解决方案不起作用的原因是线程在 run() 方法完成后终止，然后您没有像以前那样运行非守护线程。依靠第三方库的内部线程模型来保持应用程序运行并不是一个好主意。

无论 ActiveMQ 客户端中的错误或其他配置复杂性如何，最好的解决方案是使用 while(true) sleep() 范例来保持主线程处于活动状态。

【讨论】：

【解决方案2】：

谢谢蒂姆，

是的，有效。我刚刚添加了，以保持至少一个用户线程处于活动状态，这样程序就不会终止。

    while(true) {
        Thread.sleep(1000);
    }

干杯，

public class SimpleConsumer {

    static Logger logger = Logger.getLogger(SimpleConsumer.class);

    public static void main(String[] args) throws Exception {
        String url = "failover:(tcp://sydapp057lx.fxdms.net:61615)";
        String destination = "test-topic";

        TopicConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
                url);

        ActiveMQConnection connection = (ActiveMQConnection) connectionFactory
                .createConnection();

        connection.setExceptionListener(new ExceptionListener() { 
            public void onException(JMSException e) {
                logger.debug("got exception = " + e);
            } 
        });

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

        Topic topic = session.createTopic(destination);

        MessageConsumer consumer = session.createConsumer(topic);
        connection.start();

        consumer.setMessageListener(new MessageListener() {

            public void onMessage(Message msg) {
                logger.debug("msg received = " + msg);
            }

        });

        while(true) {
            Thread.sleep(1000);
        }

    }


}

【讨论】：