更改“正在传递”的消息的状态答案

【问题标题】：Changing state of messages which are "in delivery"更改“正在传递”的消息的状态
【发布时间】：2014-02-20 19:43:41
【问题描述】：

在我的应用程序中，我在 JBoss 7 AS 上设置了一个队列 (HornetQ)。一旦收到消息（将值保存在数据库中等），我就使用 Spring 批处理做一些工作，然后消费者提交 JMS 会话。有时在处理消息时出现异常，消费者的执行会突然中止。并且消息保持在“传递中”状态。我的生产队列中大约有 30 条处于此状态的消息。
我尝试重新启动消费者，但这些消息的状态没有改变。删除这些的唯一方法来自队列的消息是重新启动队列。但在此之前，我想要一种阅读这些消息的方法它们可以被更正并再次发送到队列中进行处理。我曾尝试使用 QueueBrowser 来阅读它们，但它不起作用。我在谷歌上搜索了很多，但不能想办法阅读这些信息。

我正在使用 Transacted 会话，一旦处理完消息，我就会调用：

session.commit();

这会发送确认。我正在实施spring的

org.springframework.jms.listener.SessionAwareMessageListener

接收消息然后处理它们。在处理消息时，我正在使用 Spring Batch 在数据库中插入一些数据。对于特定情况，它会尝试插入太大而无法插入列中的数据。它抛出一个异常并且事务被中止。

现在，我已经修复了我的生产者和消费者没有这样的数据，这样这种情况就不会再发生了。

但我的问题是，我的生产队列中的 30 条“正在交付”状态消息是怎么回事？我想阅读它们，以便可以更正它们并再次发送到队列进行处理。有什么方法可以阅读这些消息吗？一旦我知道它们的内容，我可以重新启动队列并再次提交它们（在更正它们之后）。

谢谢你的期待，

苏瓦纳

【问题讨论】：

标签： jms jboss7.x spring-batch hornetq

【解决方案1】：

这完全取决于您使用的交易模式。

例如，如果您使用交易：

   // session here is a TX Session 
   MessageConsumer cons = session.createConsumer(someQueue);
   session.start();
   Message msg = consumer.receive...
   session.rollback(); // this will make the messages to be redelivered

如果您使用的是非 TX：

   // session here is auto-ack
   MessageConsumer cons = session.createConsumer(someQueue);
   session.start();
   // this means the message is ACKed as we receive, doing autoACK
   Message msg = consumer.receive...

   //however the consumer here could have a buffer from the server...
   // if you are not using the consumer any longer.. close it
   consumer.close(); // this will release messages on the client buffer

或者，您也可以在 connectionFactory 上设置 consumerWindowSize=0。

这是在 2.2.5 上，但在以下版本中从未改变：

http://docs.jboss.org/hornetq/2.2.5.Final/user-manual/en/html/flow-control.html

我正在涵盖我能想到的所有可能性，因为您没有具体说明您的消费方式。如果您提供更多详细信息，那么我将能够告诉您更多信息：

【讨论】：

您好 Clebert，非常感谢您快速详细的回复。它有帮助。我已经编辑了我的问题有更多的细节。你能帮忙吗？
我已经回答过了。你必须在你的事情失败时调用回滚。或者关闭它。如果你让它保持打开状态，它将保持这种状态。
再次感谢您的回复。但我不明白，我无法阅读 30 条“交付中”状态消息。消费者读取并处理添加到队列中的新消息，但不读取旧消息。我也尝试创建另一个消费者，但这些旧消息似乎被卡住了。 :(
消息在客户端缓存中准备好被使用。关闭消费者，他们将离开缓存。

【解决方案2】：

您确实可以使用 jmx（例如 jconsole）读取队列中的消息

在 Jboss As7 中，您可以通过以下方式进行操作： MBeans>jboss.as>messaging>default>myJmsQueue>Operations

listMessagesAsJson

[编辑]

从 2.3.0 开始，你有一个专门的方法来解决这个特定的情况：

listDeliveringMessages

见https://issues.jboss.org/browse/HORNETQ-763

【讨论】：