使用MessageListener的JMS MessageConsumer在ActiveMQ关闭时终止 [英] JMS MessageConsumer Using MessageListener Terminates on ActiveMQ Shutdown

问题描述

尝试使JMS MessageConsumer幸免于ActiveMQ重新启动，因此它可以使用故障转移传输协议重新连接.

Trying to have a JMS MessageConsumer survive ActiveMQ reboots, so it can reconnect using the Failover Transport protocol.

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

However, it terminates upon shutdown of ActiveMQ.

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

This looks like a bug that was reported and "resolved", but I'm still seeing this in the latest version of ActiveMQ 5.10.0

我使用了以下Maven依赖项

I used the following maven dependency

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

以下是一些使用的示例代码

Here is some sample code using

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一起使用，因为它是非阻塞且异步的.

I would like it to work with MessageListener is it is non-blocking and asynchronous.

在此方面的任何帮助都将不胜感激.

Any help with this is greatly appreciated.

我已经按照上面JIRA的建议尝试过的某些事情是在非守护进程线程中运行此命令，但这没用.

Somethings I have already tried as suggested by the JIRA reported above is to run this in non-daemon thread, but that didn't work.

我尝试过

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

    }

}

推荐答案

感谢Tim，

是的.我刚刚添加了至少一个用户线程，以使该程序不会终止.

Yes that worked. I just added, to keep at least one user thread alive so the program doesnt terminate.

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

    }


}

这篇关于使用MessageListener的JMS MessageConsumer在ActiveMQ关闭时终止的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！