java - 迭代 ActiveMQ.Advisory.Expired.Queue 中的非持久 activemq 过期消息

Question

我正在 activemq 上构建一个应用程序，其中 iam 从生产者发送一条消息，其中我的交付模式为 NON_PERSISTENT(iam 不处理 PERSISTENT 交付模式，我知道它将存储在 DLQ 中 - 这不是我的设计)并使用 Producer.setTimeToLive(2000) 设置消息的生存时间。该功能表示消息将在 2 秒后过期。

我看到过期的消息在activeMQ管理控制台主题部分的ActiveMQ.Advisory.Expired.Queue中排队，即 http://localhost:8161/admin/topics.jsp 。

我的问题是如何迭代ActiveMQ.Advisory.Expired.Queue，以便我可以访问过期消息的MessageID。任何代码示例都是太棒了。

Answer 1

对目标ActiveMQ.Advisory.Expired.Queue的订阅就像任何主题一样，它返回一个ActiveMQMessage。可以通过 ActiveMQMessage 的 getDataStructure 方法检索 DataStructure 对象(ConsumerInfo、ProducerInfo、ConnectionInfo...)。

文档http://activemq.apache.org/advisory-message.html

示例:

Destination advisoryDestination = AdvisorySupport.getExpiredQueueMessageAdvisoryTopic(destination)
MessageConsumer consumer = session.createConsumer(advisoryDestination);
consumer.setMessageListener(this);

public void onMessage(Message msg){
    String messageId =   msg.getJMSMessageID();
    String orignalMessageId =   msg.getStringProperty(org.apache.activemq.advisory.AdvisorySupport.MSG_PROPERTY_MESSAGE_ID);
    if (msg instanceof ActiveMQMessage){
        try {
             ActiveMQMessage aMsg =  (ActiveMQMessage)msg;
             ProducerInfo prod = (ProducerInfo) aMsg.getDataStructure();
        } catch (JMSException e) {
            log.error("Failed to process message: " + msg);
        }
    }
}