让一个简单的 Spring JMS 客户端确认工作

Posted

技术标签:

【中文标题】让一个简单的 Spring JMS 客户端确认工作【英文标题】:Getting a simple Spring JMS client acknowledge to work 【发布时间】:2012-02-24 02:41:50 【问题描述】:

刚刚开始考虑让 JMS ActiveMQ Acknowledgements 在 Spring 中工作。到目前为止,我有一个消费者工作正常,除了当我不确认消息时,它仍然从队列中取出(我希望它留在那里或以死信队列结束)。

<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:jms="http://www.springframework.org/schema/jms"
       xmlns:p="http://www.springframework.org/schema/p"
       xsi:schemaLocation="
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms-3.0.xsd">

    <!-- A JMS connection factory for ActiveMQ -->
    <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"
    p:brokerURL="failover://(tcp://jms1:61616,tcp://jms2:61616)?randomize=false&amp;jms.redeliveryPolicy.maximumRedeliveries=5" />

    <!-- A POJO that implements the JMS message listener -->
    <bean id="simpleMessageListener" class="com.company.ConsumerClass" />

    <!-- A JMS namespace aware Spring configuration for the message listener container -->
    <jms:listener-container
            container-type="default"
            connection-factory="connectionFactory"
            acknowledge="client"
            concurrency="10-50"
            cache="consumer">
        <jms:listener destination="someQueue" ref="simpleMessageListener" method="onMessage" />
    </jms:listener-container>
</beans>

在 ConsumerClass 中,我的简单消费者如下所示:

@Override public final void onMessage(Message message) 
    Object postedMessage = null;
    try 
        postedMessage = ((ObjectMessage) message).getObject();

        if (postedMessage.getClass() == SomeMessageType.class) 
            try 
                //Some logic here

                message.acknowledge();
                return; //Success Here
             catch (MyException e) 
                logger.error("Could not process message, but as I didn't call acknowledge I expect it to end up in the dead message queue");
            
        
     catch (JMSException e) 
        logger.error("Error occurred pulling Message from Queue", e);
    

    //Also worth noting, if I throw new RuntimeException("Aww Noos"); here then it won't take it from the queue, but it won't get consumed (or end up as dead letter)...

【问题讨论】:

【参考方案1】:

阅读此文档:Spring JMS 容器不使用message.acknowledge()

侦听器容器提供以下消息确认选项:

“sessionAcknowledgeMode”设置为“AUTO_ACKNOWLEDGE”(默认):在监听器执行前自动确认消息;在抛出异常的情况下不重新投递。 “sessionAcknowledgeMode”设置为“CLIENT_ACKNOWLEDGE”:监听器执行成功后自动消息确认;在抛出异常的情况下不重新投递。 “sessionAcknowledgeMode”设置为“DUPS_OK_ACKNOWLEDGE”:监听器执行期间或之后的延迟消息确认;在抛出异常的情况下可能会重新交付。 “sessionTransacted”设置为“true”:成功监听执行后的事务确认;在抛出异常的情况下保证重新交付。

【讨论】:

源文档:Spring Framework's AbstractMessageListenerContainer javax.jms.Session.AUTO_ACKNOWLEDGE 适合任何困惑的人。【参考方案2】:

我在http://ourcraft.wordpress.com/2008/07/21/simple-jms-transaction-rollbacks-work/找到了答案

如果你改变了 acknowledge="transacted" 并确保你抛出 new RuntimeException("Message could not be used. Roll back transaction"); 看起来效果很好。在 OnMessage() 例程的末尾。

仍然不知道 acknowledge="client" 实现了什么

【讨论】:

客户端确认处理侦听器方法返回时的确认。无论是优雅地还是通过异常,如果方法调用没有被中断,它将在完成时得到确认。由于它不关心是否抛出异常,它会消耗队列中的消息。【参考方案3】:

如今 Spring 提供了对普通 JMS 消息侦听器的良好包装器。

请参阅AbstractMessageListenerContainer 的 JavaDocs。

"sessionAcknowledgeMode"设置为"CLIENT_ACKNOWLEDGE":监听执行成功后自动消息确认;在引发用户异常以及其他侦听器执行中断(例如 JVM 死亡)的情况下,尽最大努力重新传递。

因此,当您定义@JmsListener 方法时,确认成功完成后会自动发送,但您可以抛出异常再次接收消息。

【讨论】:

【参考方案4】:

在你的消费者的消息上调用 acknowledge() 方法。

考虑以下场景:应用程序接收但未确认消息。应用程序接收到后续消息并确认它。以前的消息会发生什么?前一条消息也被视为已确认。通常,确认特定消息会确认会话接收到的所有先前消息。在上面的输出中,只有消息 5 被明确确认。消息 5 之前的所有消息都被隐式确认。消息 5 之后的消息未被确认。

更多详情请咨询this article

也可以查看这篇文章Sun Java System Message Queue 4.3 Developer's Guide for Java Clients

【讨论】:

【参考方案5】:

在我的实践中,客户确认很少使用。典型设置是自动确认,因此如果您的代码从 onMessage() 正常返回(无异常),则消息会自动确认。如果您的代码从 onMessage() 引发异常,则没有确认消息,并且消息通常会重新传递到预先配置的次数,之后通常会被丢弃或放入死消息队列。

在您的情况下,从 JMS 服务器的角度来看,看起来客户端请求消息但从未确认,因此客户端仍在“处理”它。在这种情况下,同一队列上的其他消费者将看不到消息,因此您可能会觉得消息“离开队列”,而事实上它仍然存在。显然你也不会在死消息队列中看到这样的消息。

我建议你阅读 JMS 规范以清楚地了解不同的确认模式。

【讨论】:

对不起,在这种情况下,如果我在 message.acknowledge() 之后返回则成功(我在代码示例中添加了这个) 另外,如果我取消注释 throw RuntimeException(它必须是运行时才能满足 MessageListener 接口)我得到一个错误:08:39:59,066 WARN org.springframework.jms.listener.DefaultMessageListenerContainer# 0-2 listener.DefaultMessageListenerContainer:694 - JMS 消息侦听器的执行失败,并且没有设置 ErrorHandler。 java.lang.RuntimeException: Aww Noos 我可以看到,在这种情况下,消息由消费者持有,直到我终止它,此时他们将尝试在另一个消费者身上消费,这让我认为 acknowledge="client" 确实什么都没有 实际上,如果我正确阅读了javadoc,“auto”会在调用监听器之前确认:“sessionAcknowledgeMode”设置为“AUTO_ACKNOWLEDGE”(默认):在监听器执行之前自动确认消息;在抛出异常的情况下不重新投递。”【参考方案6】:

使用下面的代码就可以了。

<bean id="containerName"  class="org.springframework.jms.listener.DefaultMessageListenerContainer">
    <property name="connectionFactory" ref=connectionFactoryBean />
    <property name="destinationName" ref="queue" />
    <property name="messageListener" ref="listner" />
    <property name="sessionAcknowledgeModeName" value="CLIENT_ACKNOWLEDGE"/>
</bean>

【讨论】:

以上是关于让一个简单的 Spring JMS 客户端确认工作的主要内容,如果未能解决你的问题,请参考以下文章

JMS消息使用RDD来确保仅用于确认目的的安全性

JMS的可靠性

在使用 JMS 消息之前保持 spring 上下文处于活动状态

ActiveMQ(03):JMS的可靠性机制

AWS SQS JMS确认

Spring整合JMS(消息中间件)