java - 如何使用 Spring-JMS 不只确认一条消息？

Question

有一个类“MyConsumer”，它从队列接收消息并处理它们。有两个要求:

1. 如果消息包含无效内容，MyConsumer 不应确认它，但可以处理后续消息
2. 未消费的消息将在 MyConsumer 重启时再次投递

我尝试使用 spring-jms 和监听器容器支持，但找不到符合第一个要求的解决方案。

我的代码:

<amq:queue id="destination" physicalName="org.springbyexample.jms.test"/>

<amq:connectionFactory id="jmsFactory" brokerURL="tcp://localhost:11111"/>

<bean id="jmsConsumerConnectionFactory"
      class="org.springframework.jms.connection.SingleConnectionFactory"
      p:targetConnectionFactory-ref="jmsFactory"/>

<bean id="jmsConsumerTemplate" class="org.springframework.jms.core.JmsTemplate"
      p:connectionFactory-ref="jmsConsumerConnectionFactory"
      p:defaultDestination-ref="destination"/>

<bean id="jmsMessageListener" class="test.MyConsumer"/>
<bean id="errorHandler" class="test.MyErrorHandler"/>

<jms:listener-container container-type="default"
                        connection-factory="jmsConsumerConnectionFactory"
                        error-handler="errorHandler"
                        acknowledge="client">
    <jms:listener destination="org.springbyexample.jms.test" ref="jmsMessageListener"/>
</jms:listener-container>

类MyConsumer:

@Override
public void onMessage(Message message) {
    TextMessage textMessage = (TextMessage) message;
    try {
        System.out.println("!!!!!!!!! get message: " + textMessage.getText());
    } catch (JMSException e) {
        e.printStackTrace();
    }
    if (theNumberOfMessageIs(3)) {
        throw new RuntimeException("something is wrong");
    }
}

你可能会注意到listener-container中的acknowledge是client，实际上它有3个值:

1. 自动(默认)
2. 客户
3. 已成交

我尝试了所有这些，但没有一个符合我的要求。我的测试场景:

1. 生产者将 3 条消息放入队列
2. 启动一个线程来监视队列中的消息计数，当计数发生变化时打印
3. 启动consumer，它会从队列接收消息，并处理它们
4. 稍等片刻，将另外 3 条消息放入队列

对于自动:

MyConsumer收到每条消息后都会确认，无论是否抛出异常

对于客户端:

仅当 onMessage 中未引发异常时，MyConsumer 才会确认。对于第三条消息，它会抛出异常，队列中会有一条消息未消费。但是当它收到第四条消息并且没有抛出异常时，队列中的第三条消息将消失

对于交易:

如果MyConsumer中抛出异常，消息将不会被确认并被重新传递多次。之后，消息从队列中消失

但是都不符合要求1。

我想知道:我是否需要寻找 Spring-jms 之外的其他解决方案，或者我的用法不正确？

Answer 1

auto DefaultMessageListenerContainer 确实是为事务而设计的 - 正如您所发现的那样，使用 auto 时，消息始终会得到确认。您可以使用 SimpleMessagseListenerContainer 它将按照您的意愿工作，但它还有其他限制；请参阅 JavaDocs。

client 这就是 JMS 在您确认 #4 时的工作方式，#3 也会自动确认 - 请参阅 Message JavaDocs。客户端模式用于减少确认流量(例如，每 10 条消息确认一次)。

transacted 这是代理的功能，您可以配置 AMQ 在重试一定次数后将错误消息发送到死信队列。

您需要一些过程将消息从 DLQ 移回主队列以便稍后重试(可能在重新启动时的初始化期间)。