读取 ActiveMQ 消息而不删除

Posted 2023-03-10

【中文标题】读取 ActiveMQ 消息而不删除

【英文标题】：Reading ActiveMQ messages without deletion

【发布时间】：2011-12-25 10:32:58

【问题描述】：

我正在尝试通读 ActiveMQ 消息并根据一些过滤器来处理其中一些消息或将其他消息留在队列中。 我使用带有以下代码的 NMS API：

        Uri connecturi = new Uri("activemq:tcp://model.net:61616");
        IConnectionFactory factory = new NMSConnectionFactory(connecturi);
        List<ModelBuilderBase> result = new List<ModelBuilderBase>();
        using (IConnection connection = factory.CreateConnection())
        using (ISession session = connection.CreateSession())
        
            IDestination destination = SessionUtil.GetDestination(session, "queue://cidModelbuilderQ");
            using (IMessageConsumer consumer = session.CreateConsumer(destination))
            
                connection.Start();
                ITextMessage message;
                while ((message = consumer.ReceiveNoWait() as ITextMessage) != null)
                
                    if (message.Properties[MANDATOR] == null || message.Properties[REFCODE] == null)
                        continue;
                    var mandator = message.Properties[MANDATOR].ToString();
                    var refCode = message.Properties[REFCODE].ToString();
                    result.Add(ModelBuilderFactory.Instance.GetInstance(refCode, mandator));
                
            

问题是收到消息后，消息被删除。我可以以某种方式更改此行为并仅在成功处理后手动删除消息吗？

【参考方案1】：

虽然为此编写工作代码并不容易，但感谢 ARS 的回答，我现在有了这个工作解决方案：

        Uri connecturi = new Uri("activemq:tcp://model.net:61616");
        IConnectionFactory factory = new NMSConnectionFactory(connecturi);
        List<ModelBuilderBase> result = new List<ModelBuilderBase>();
        using (IConnection connection = factory.CreateConnection())
        using (ISession session = connection.CreateSession())
        

            IDestination destination = SessionUtil.GetDestination(session, "queue://cidModelbuilderQ");
            using (IMessageConsumer consumer = session.CreateConsumer(destination))
            
                connection.Start();
                var q = session.GetQueue("cidModelbuilderQ");
                var b = session.CreateBrowser(q);
                var msgs = b.GetEnumerator();
                while (msgs.MoveNext())
                
                    ITextMessage message = msgs.Current as ITextMessage;
                    if (message.Properties[MANDATOR] == null || message.Properties[REFCODE] == null)
                        continue;
                    var mandator = message.Properties[MANDATOR].ToString();
                    var refCode = message.Properties[REFCODE].ToString();
                    result.Add(ModelBuilderFactory.Instance.GetInstance(refCode, mandator));
                
            
        

【参考方案2】：

创建一个 QueueBrowser 对象以查看指定队列上的消息。对消息执行逻辑，然后创建一个 QueueReceiver 对象来接收来自指定队列的消息。