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

换句话说,假设队列中有10条消息.我消耗了其中的7个,然后在第5条消息上确认.这反过来删除了消费者从队列中接收的所有7条消息.有一种方法只能确认并从队列中删除消息直到第5条消息.

编辑：我已经尝试创建两个会话并从不同的会话消费,但(至少与apache qpid)这执行不一致.我的意思是不一致,有时在测试期间,无论你等待多久,一个消费者都能够接收消息,而另一个消费者根本不接收消息.这对我来说是一种解决方案,但由于不一致,不能将其作为解决方案.

解决方法:

我知道这篇文章很老,但这个答案应该会让那些后来偶然发现它的人受益.

如果您想要对您想要确认哪些消息进行细粒度控制,那么个人确认方法应该对您有所帮助.使用此确认模式,您可以在会话中确认单个消息.未被确认的消息将被重新传递.

这不是规范的一部分,但大多数队列提供者都支持它超出规范.

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可以说是个别消息可能.

Examples
  # acknowledge all received messages
  session.acknowledge

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

处理批处理时,您可以确认接收和处理的每条消息.如果遇到异常,请不要删除该消息.