如何在 ActiveMQ 中获取所有排队的消息？答案

【问题标题】：How to get all enqueued messages in ActiveMQ?如何在 ActiveMQ 中获取所有排队的消息？
【发布时间】：2016-09-13 20:44:59
【问题描述】：

我想构建一个简单的消费者程序（在 java 中）来获取存储在 ActiveMQ 主题中的所有消息。我有一个生产者在队列中发送 TextMessage。

但我不知道如何开始编写我的消费者来检索旧消息并等待新消息。

如果你有例子，谢谢！

这是我的制作人：http://pastebin.com/uRy9D8mY

这是我的消费者：http://pastebin.com/bZh4r66e

当我在我的消费者之前运行我的生产者，然后运行消费者时，我什么也没得到。当我运行我的消费者和生产者时，我在队列中添加了 72 条消息，但我的消费者只收到了 24 条消息...

【问题讨论】：

标签： java activemq

【解决方案1】：

我建议阅读本教程（Apache ActiveMQ 也是如此）SUN Jms tutorial

JMS/ActiveMQ 程序的编写方式有很多种，可以使用 Spring 等各种框架，也可以使用纯 java。

基本上，像这样编写一个监听器类：

public class MyListener implements MessageListener{
   public void onMessage(Message message){
      // Read and handle message here.
   }
}

由于您已经在生成消息，我假设您已建立并运行连接。

session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
consumer = session.createConsumer("MyQueue");
listener = new MyListener ();
consumer.setMessageListener(listener);
connection.start(); 
// At this point, messages should arrive from the queue to your listener.

还有一些错误处理代码没有包含在这个例子中，但是你应该可以借助教程和JMS文档来弄清楚。

【讨论】：

谢谢！我已经按照您的 Sun jms 教程进行操作，但遇到了 2 个问题：我无法接收已排队的消息，当我等待新消息时，我仅收到 72 条消息中的 24 条。
好的。我想您必须更详细才能获得更多帮助。您的设置如何，您的消费方式等。
我添加了消费者和生产者代码。我使用默认配置的 ActiveMQ。
从 72 条消息中只得到 24 条消息的错误是什么？你是怎么解决的？@kdelemm

【解决方案2】：

使用下面给出的代码，您可以读取队列中排队的所有消息。

在这段代码中，while 循环是一个无休止的循环，它将迭代所有队列中的消息。
一旦队列中没有消息，它将等待 5 秒，然后自动停止连接并中断循环。

如果您需要一个无休止的消费者，它会在新添加到队列时读取所有消息，然后删除 else 部分，这样程序就不会终止。

    ConnectionFactory factory =  new ActiveMQConnectionFactory("tcp://localhost:61616");
    Connection con = factory.createConnection();
    Session session =   con.createSession(false, Session.AUTO_ACKNOWLEDGE);
    Queue queue = session.createQueue("tmp_queue2");
    MessageConsumer consumer = session.createConsumer(queue);
    con.start();     
    while (true) {     
        Message msg = consumer.receive(5000); 
        if (msg instanceof TextMessage) {
            TextMessage tm = (TextMessage) msg;
            System.out.println(tm.getText());       
        }
        else{
            System.out.println("Queue Empty"); 
            con.stop();
            break;
        }
    }

希望这个消费者计划能帮助刚接触 ActiveMQ 的人。

【讨论】：