java - Jms 消息仅在消息确认之前确认

Question

如何确保消息确认仅删除在 jms 代理中调用确认的消息之前的消息。目前我有一个系统，它从 jms 队列中消费并部分处理它。稍后，这些消息中的一批会被不同的线程持久化。我现在需要在消息上确认。但问题是我必须停止使用消息，否则确认先前收到的消息也会确认收到的所有其他后续消息。

换句话说，假设我在队列中有 10 条消息。我消耗了其中的 7 个，然后在第 5 条消息上确认。这反过来会从队列中删除消费者收到的所有 7 条消息。有没有办法只确认并从队列中删除消息，直到第 5 条消息。

编辑:我已经尝试创建两个 session 并从不同的 session 中使用，但是(至少使用 apache qpid)这执行不一致。我的意思是不一致，有时在测试过程中，一个消费者能够接收消息，而另一个消费者根本没有收到消息，无论您等待多长时间。这对我来说是一种解决方案，但由于不一致，无法将其用作解决方案。

Answer 1

我知道这篇文章很旧，但这个答案应该会让后来偶然发现它的人受益。

如果您想细粒度地控制要确认的消息，individual 确认方法应该对您有所帮助。使用这种确认模式，您可以确认 session 中的单个消息。尚未确认的消息将被重新传送。

这不是规范的一部分，但大多数队列提供程序在规范之外支持它。

Oracle

For more flexibility, Message Queue lets you customize the JMS client-acknowledge mode. In client-acknowledge mode, the client explicitly acknowledges message consumption by invoking the acknowledge() method of a message object.

The standard behavior of this method is to cause the session to acknowledge all messages that have been consumed by any consumer in the session since the last time the method was invoked. (That is, the session acknowledges the current message and all previously unacknowledged messages, regardless of who consumed them.)

In addition to the standard behavior specified by JMS, Message Queue lets you use client-acknowledge mode to acknowledge one message at a time.

public interface com.sun.messaging.jms.Message {
          void acknowledgeThisMessage() throws JMSException;
          void acknowledgeUpThroughThisMessage() throws JMSException;
}

ActiveMQ

One can imagine other acknowledge modes that would be useful too, for example: CONSUMER_ACKNOWLEDGE where Message.acknowledge() would acknowledge only messages received up on a particular MessageConsumer, or CONSUMER_CHECKPOINT_ACKNOWLEDGE where Message.acknowledge() would acknowledge only messages received up to and including the Message instance on which the method was called.

But without embarking on all these various different possibilities, would it be possible to consider just adding INDIVIDUAL_ACKNOWLEDGE mode? This alone would make it possible for multithreaded applications to achieve whatever behaviors they need.

connection.createQueueSession(false, ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE);

我没有亲自使用过 QPID，但是 documentation hints事实上，单个消息 acks 是可能的。

Examples
  # acknowledge all received messages
  session.acknowledge

  # acknowledge a single message
  session.acknowledge :message => message

在处理批处理时，您可以确认收到和处理的每条消息。如果遇到异常，请不要确认消息。