Spring JMS:处理事务中的消息

1.引言

这篇文章将向您展示在使用JMS异步接收消息期间,使用者执行过程中的错误如何导致消息丢失。 然后,我将解释如何使用本地事务解决此问题。

您还将看到这种解决方案在某些情况下可能导致消息重复(例如,当它将消息保存到数据库中,然后侦听器执行失败时)。 发生这种情况的原因是因为JMS事务独立于其他事务资源(如DB)。 如果您的处理不是幂等的,或者您的应用程序不支持重复消息检测,那么您将不得不使用分布式事务。

分布式事务超出了此职位的范围。 如果您对处理分布式事务感兴趣,可以阅读这篇有趣的文章。

我已经实现了一个再现以下情况的测试应用程序:

  1. 发送和接收消息:使用者将处理收到的消息,并将其存储到数据库中。

    生产者将消息发送到队列:

    send1

    使用者从队列中检索消息并进行处理:

    发送2

  2. 消息处理之前发生错误:使用者检索消息,但是在将消息存储到DB之前执行失败。

    发送3

  3. 处理消息后发生错误:使用者检索消息,将其存储到DB,然后执行失败。

    发送4

  • 该应用程序的源代码可以在github上找到。

2.测试应用

测试应用程序执行两个测试类TestNotTransactedMessagingTestTransactedMessaging 。 这些类都将执行上述三种情况。

让我们看看在没有事务的情况下执行应用程序时的配置。

app-config.xml

应用程序配置。 基本上,它会在指定的软件包内进行检查以自动检测应用Bean:生产者和使用者。 它还配置了将在其中存储处理后的通知的内存数据库。

<context:component-scan base-package="xpadro.spring.jms.producer, xpadro.spring.jms.receiver"/><bean id="jdbcTemplate" class="org.springframework.jdbc.core.JdbcTemplate"><constructor-arg ref="dataSource"/>
</bean><jdbc:embedded-database id="dataSource"><jdbc:script location="classpath:db/schema.sql" />
</jdbc:embedded-database>

notx-jms-config.xml

配置JMS基础结构,该基础结构是:

  • 经纪人联系
  • JmsTemplate
  • 将通知发送到的队列
  • 侦听器容器,它将发送通知给侦听器以处理它们
<!-- Infrastructure -->
<bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"><property name="brokerURL" value="vm://embedded?broker.persistent=false"/>
</bean><bean id="cachingConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory"><property name="targetConnectionFactory" ref="connectionFactory"/>
</bean><bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"><property name="connectionFactory" ref="cachingConnectionFactory"/><property name="defaultDestination" ref="incomingQueue"/>
</bean><!-- Destinations -->
<bean id="incomingQueue" class="org.apache.activemq.command.ActiveMQQueue"><constructor-arg value="incoming.queue"/>
</bean><!-- Listeners -->
<jms:listener-container connection-factory="connectionFactory"><jms:listener ref="notificationProcessor" destination="incoming.queue"/>
</jms:listener-container>

生产者仅使用jmsTemplate发送通知。

@Component("producer")
public class Producer {private static Logger logger = LoggerFactory.getLogger(Producer.class);@Autowiredprivate JmsTemplate jmsTemplate;public void convertAndSendMessage(String destination, Notification notification) {jmsTemplate.convertAndSend(destination, notification);logger.info("Sending notification | Id: "+notification.getId());}
}

侦听器负责从队列中检索通知,并将其存储到数据库中。

@Component("notificationProcessor")
public class NotificationProcessor implements MessageListener {private static Logger logger = LoggerFactory.getLogger(NotificationProcessor.class);@Autowiredprivate JdbcTemplate jdbcTemplate;@Overridepublic void onMessage(Message message) {try {Notification notification = (Notification) ((ObjectMessage) message).getObject();logger.info("Received notification | Id: "+notification.getId()+" | Redelivery: "+getDeliveryNumber(message));checkPreprocessException(notification);saveToBD(notification);checkPostprocessException(message, notification);} catch (JMSException e) {throw JmsUtils.convertJmsAccessException(e);}}	...
}

当id = 1的通知到达时, checkPreprocessException方法将抛出运行时异常。 这样,在将消息存储到数据库之前,我们将导致错误。

如果到达id = 2的通知, checkPostprocessException方法将引发异常,从而在将其存储到数据库之后立即引起错误。

getDeliveryNumber方法返回发送消息的次数。 这仅适用于事务,因为在侦听器处理失败导致回滚之后,代理将尝试重新发送消息。

最后, saveToDB方法非常明显。 它将通知存储到数据库。

您始终可以通过单击本文开头的链接来检查此应用程序的源代码。

3.测试没有交易的消息接收

我将启动两个测试类,一个不包含事务,另一个在本地事务中。 这两个类都扩展了一个基类,该基类加载了常见的应用程序上下文并包含一些实用程序方法:

@ContextConfiguration(locations = {"/xpadro/spring/jms/config/app-config.xml"})
@DirtiesContext
public class TestBaseMessaging {protected static final String QUEUE_INCOMING = "incoming.queue";protected static final String QUEUE_DLQ = "ActiveMQ.DLQ";@Autowiredprotected JdbcTemplate jdbcTemplate;@Autowiredprotected JmsTemplate jmsTemplate;@Autowiredprotected Producer producer;@Beforepublic void prepareTest() {jdbcTemplate.update("delete from Notifications");}protected int getSavedNotifications() {return jdbcTemplate.queryForObject("select count(*) from Notifications", Integer.class);}protected int getMessagesInQueue(String queueName) {return jmsTemplate.browse(queueName, new BrowserCallback<Integer>() {@Overridepublic Integer doInJms(Session session, QueueBrowser browser) throws JMSException {Enumeration<?> messages = browser.getEnumeration();int total = 0;while (messages.hasMoreElements()) {messages.nextElement();total++;}return total;}});}
}

实用方法说明如下:

  • getSavedNotifications :返回存储到数据库的通知数。 我使用了queryForObject方法,因为自版本3.2.2开始建议使用该方法。 queryForInt方法已被弃用。
  • getMessagesInQueue :允许您检查指定队列中哪些消息仍在等待处理。 对于此测试,我们有兴趣知道仍有多少通知等待处理。

现在,让我向您展示第一个测试的代码( TestNotTransactedMessaging )。 此测试启动本文开头指示的3种情况。

@Test
public void testCorrectMessage() throws InterruptedException {Notification notification = new Notification(0, "notification to deliver correctly");producer.convertAndSendMessage(QUEUE_INCOMING, notification);Thread.sleep(6000);printResults();assertEquals(1, getSavedNotifications());assertEquals(0, getMessagesInQueue(QUEUE_INCOMING));
}@Test
public void testFailedAfterReceiveMessage() throws InterruptedException {Notification notification = new Notification(1, "notification to fail after receiving");producer.convertAndSendMessage(QUEUE_INCOMING, notification);Thread.sleep(6000);printResults();assertEquals(0, getSavedNotifications());assertEquals(0, getMessagesInQueue(QUEUE_INCOMING));
}@Test
public void testFailedAfterProcessingMessage() throws InterruptedException {Notification notification = new Notification(2, "notification to fail after processing");producer.convertAndSendMessage(QUEUE_INCOMING, notification);Thread.sleep(6000);printResults();assertEquals(1, getSavedNotifications());assertEquals(0, getMessagesInQueue(QUEUE_INCOMING));
}private void printResults() {logger.info("Total items in \"incoming\" queue: "+getMessagesInQueue(QUEUE_INCOMING));logger.info("Total items in DB: "+getSavedNotifications());
}

4,执行测试

好的,让我们执行测试,看看结果是什么:

testCorrectMessage输出:

Producer|Sending notification | Id: 0
NotificationProcessor|Received notification | Id: 0 | Redelivery: 1
TestNotTransactedMessaging|Total items in "incoming" queue: 0
TestNotTransactedMessaging|Total items in DB: 1

此处没有问题,因为消息已正确接收并存储到数据库,所以队列为空。

testFailedAfterReceiveMessage输出:

Producer|Sending notification | Id: 1
NotificationProcessor|Received notification | Id: 1 | Redelivery: 1
AbstractMessageListenerContainer|Execution of JMS message listener failed, and no ErrorHandler has been set.
java.lang.RuntimeException: error after receiving message
TestNotTransactedMessaging|Total items in "incoming" queue: 0
TestNotTransactedMessaging|Total items in DB: 0

由于它在事务外部执行,因此使用确认模式(默认为自动)。 这意味着一旦调用onMessage方法并因此将其从队列中删除,就认为该消息已成功传递。 因为侦听器在将消息存储到数据库之前失败,所以我们丢失了消息!

testFailedAfterProcessingMessage输出:

2013-08-22 18:39:09,906|Producer|Sending notification | Id: 2
2013-08-22 18:39:09,906|NotificationProcessor|Received notification | Id: 2 | Redelivery: 1
2013-08-22 18:39:09,906|AbstractMessageListenerContainer|Execution of JMS message listener failed, and no ErrorHandler has been set.
java.lang.RuntimeException: error after processing message
2013-08-22 18:39:15,921|TestNotTransactedMessaging|Total items in "incoming" queue: 0
2013-08-22 18:39:15,921|TestNotTransactedMessaging|Total items in DB: 1

在这种情况下,在执行失败之前,该消息已从队列(AUTO_ACKNOWLEDGE)中删除并存储到DB。

5,添加本地交易

通常,我们不允许像测试的第二种情况那样丢失消息,因此我们要做的是在本地事务中调用侦听器。 所做的更改非常简单,并不意味着从我们的应用程序中修改一行代码。 我们只需要更改配置文件。

为了测试这3种涉及事务的情况,我将以下配置文件notx-jms-config.xml替换为:

tx-jms-config.xml

首先,我添加了在发生回滚的情况下进行的重新传递的数量(由于侦听器执行中的错误导致):

<bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"><property name="brokerURL" value="vm://embedded?broker.persistent=false"/><property name="redeliveryPolicy"><bean class="org.apache.activemq.RedeliveryPolicy"><property name="maximumRedeliveries" value="4"/></bean></property>
</bean>

接下来,我指示侦听器将在事务内执行。 这可以通过修改侦听器容器定义来完成:

<jms:listener-container connection-factory="connectionFactory" acknowledge="transacted"><jms:listener ref="notificationProcessor" destination="incoming.queue"/>
</jms:listener-container>

这将导致在本地JMS事务中执行对侦听器的每次调用。 收到消息后,事务将开始。 如果侦听器执行失败,则消息接收将回滚。

这就是我们要做的一切。 让我们使用此配置启动测试。

6,测试交易中的消息接收

来自TestTransactedMessaging类的代码实际上与先前的测试相同。 唯一的区别是,它向DLQ(死信队列)添加了查询。 在事务内执行时,如果回退消息接收,则代理会将消息发送到此队列(在所有重新传递失败之后)。

我跳过了成功接收的输出,因为它不会带来任何新的变化。

testFailedAfterReceiveMessage输出:

Producer|Sending notification | Id: 1
NotificationProcessor|Received notification | Id: 1 | Redelivery: 1
AbstractMessageListenerContainer|Execution of JMS message listener failed, and no ErrorHandler has been set.
java.lang.RuntimeException: error after receiving message
NotificationProcessor|Received notification | Id: 1 | Redelivery: 2
AbstractMessageListenerContainer|Execution of JMS message listener failed, and no ErrorHandler has been set.
...
java.lang.RuntimeException: error after receiving message
NotificationProcessor|Received notification | Id: 1 | Redelivery: 5
AbstractMessageListenerContainer|Execution of JMS message listener failed, and no ErrorHandler has been set.
java.lang.RuntimeException: error after receiving message
TestTransactedMessaging|Total items in "incoming" queue: 0
TestTransactedMessaging|Total items in "dead letter" queue: 1
TestTransactedMessaging|Total items in DB: 0

如您所见,第一次接收失败,并且代理尝试将其重新发送四次(如maximumRedeliveries属性中所示)。 由于情况持续存在,因此消息已发送到特殊DLQ队列。 这样,我们不会丢失消息。

testFailedAfterProcessingMessage输出:

Producer|Sending notification | Id: 2
NotificationProcessor|Received notification | Id: 2 | Redelivery: 1
AbstractMessageListenerContainer|Execution of JMS message listener failed, and no ErrorHandler has been set.
java.lang.RuntimeException: error after processing message
NotificationProcessor|Received notification | Id: 2 | Redelivery: 2
TestTransactedMessaging|Total items in "incoming" queue: 0
TestTransactedMessaging|Total items in "dead letter" queue: 0
TestTransactedMessaging|Total items in DB: 2

在这种情况下,这是发生的情况:

  1. 侦听器检索到消息
  2. 它将消息存储到数据库
  3. 侦听器执行失败
  4. 代理重新发送消息。 由于情况已解决,因此侦听器将消息再次存储到DB。 该消息已重复。

7,结论

将本地事务添加到消息接收中可避免丢失消息。 我们必须考虑的是,可能会出现重复的消息,因此我们的侦听器将不得不检测到它,否则我们的处理必须是幂等的才能再次进行处理。 如果这不可能,我们将不得不进行分布式事务,因为它们支持涉及不同资源的事务。

参考: Spring JMS:在XavierPadró的Blog博客上处理来自JCG合作伙伴 Xavier Padro的事务内的消息 。

翻译自: https://www.javacodegeeks.com/2014/02/spring-jms-processing-messages-within-transactions.html

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/364749.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

OS X EI Captain 下解决 There was a problem confirming the ssl certificate 问题

参考&#xff1a; Problem Confirming the SSL Certificate - OSX OS X EI Captain 下解决 There was a problem confirming the ssl certificate 问题 在安装 matplotlib 时&#xff0c;出现以下错误&#xff1a; python3 -mpip install matplotlibCollecting matplotlibCould…

Linux下Python编译安装

1.安装python3 1.1下载python源码包 网址&#xff1a;https://www.python.org/downloads/release/python-362/ 下载地址&#xff1a;https://www.python.org/ftp/python/3.6.2/Python-3.6.2.tgz 1.1.1安装python前的库环境&#xff0c;非常重要 yum install gcc patch libffi-d…

call/apply以及this指向的理解

javascript是面向对象的语言&#xff0c;Function也是一种对象&#xff0c;有自己的属性和方法。call和apply就是js函数自带方法&#xff0c;挂在Fucntion.prototype上。一般调用某函数时&#xff0c;直接“函数名&#xff08;参数&#xff09;”的写法即可&#xff0c;函数内部…

Java排序算快速排序_Java排序算法 [快速排序]

package cn.com.dom4j.sort;public class QuickSort {/**快速排序在 Java中, 快速排序被用作基本数据类型的排序 (当然, 不只快速排序一种)快速排序是实践中的一种快速的排序算法, 在 C或对 Java基本类型的排序证特别有用.它的平均运行时间是 O(N logN), 该算法之所以特别快, 主…

Java Enterprise软件与应有的内容

许多开发人员在其职业生涯的某个阶段最终进入了Java“企业”世界。 我知道&#xff0c;仅此一个词就能引起各种反应&#xff0c;这是正确的。 通常&#xff0c;存在许多有趣的技术挑战的环境最终会成为那些没人愿意从事的环境&#xff0c;因为它们脆弱&#xff0c;难以处理且没…

OO第二单元作业总结

一&#xff1a;设计策略 第一次作业&#xff1a;第一次是单电梯傻瓜调度策略&#xff0c;因此我把调度器当作共享资源对象&#xff0c;有一个put和一个get方法&#xff0c;因为只有一个电梯&#xff0c;并且单次取出和投放一个请求&#xff0c;因此只需要同步控制一下这两个方法…

LESS+to+MCSS

此文已由作者郑海波授权网易云社区发布。欢迎访问网易云社区&#xff0c;了解更多网易技术产品运营经验一、前言虽然首页没有开始做&#xff0c;昨天仍决定将[MCSS](https://github.com/NetEaseWD/mcss)从身边的基友们开始向杭研推广了&#xff0c;从开始做这个直到现在推广遇到…

jmeter找不到java.dll_Windows下Jmeter安装出现Not able to find Java executable or version问题解决方案...

最近在做一个开放接口平台性能测试 , 指标是最少达到1000/s的并发 , 接口鉴权 百万级的表 在1s内完成..在众多压测工具中 ,,选择了Apache的jmeter ,于官网下载了最新版本http://jmeter.apache.org/download_jmeter.cgi (jmeter下载地址)由于jmeter运行是基于java的,所以需要…

localStorage/cookie 用法分析与简单封装

本地存储是HTML5中提出来的概念&#xff0c;分localStorage和sessionStorage。通过本地存储&#xff0c;web应用程序能够在用户浏览器中对数据进行本地的存储。与 cookie 不同&#xff0c;存储限制要大得多&#xff08;至少5MB&#xff09;&#xff0c;并且信息不会被传输到服务…

使用Lucene的搜索服务器搜索Jira问题

您可能还记得我的第一篇博客文章&#xff0c;其中描述了Lucene开发人员如何使用Lucene搜索应用程序查找我们的Jira问题来食用我们自己的狗食。 该应用程序已成为许多Lucene现代功能的强大展示&#xff0c;例如侧向钻取和动态范围刻面&#xff0c; 基于中缀匹配的新建议 &#…

迭代加深搜索 C++解题报告 :[SCOI2005]骑士精神

题目 此题根据题目可知是迭代加深搜索。 首先应该枚举空格的位置&#xff0c;让空格像一个马一样移动。 但迭代加深搜索之后时间复杂度还是非常的高&#xff0c;根本过不了题。 感觉也想不出什么减枝&#xff0c;于是便要用到了乐观估计函数&#xff08;Optimistic Estimation …

一个web项目web.xml的配置中context-param配置作用

<context-param>的作用:web.xml的配置中<context-param>配置作用1. 启动一个WEB项目的时候,容器(如:Tomcat)会去读它的配置文件web.xml.读两个节点: <listener></listener> 和 <context-param></context-param>2.紧接着,容器创建一个Servl…

JavA持有类_关于继承:Java持有对象的超类的类型

本问题已经有最佳答案&#xff0c;请猛点这里访问。我有两个类动物和狗。因为你可以从动物身上延伸出来。我可以毫无问题地编写这些类&#xff0c;但我注意到我可以像这样创建一个新的dog对象&#xff1a;Dog firstDog new Dog("rocky");没关系&#xff0c;但是当我…

CSS3 选择器用法小结

表单选择器&#xff1a; /*:enabled 可用的 :disabled 被禁用的 :focus 获取了焦点的 多用在表单元素上*/ input:enabled {...} input:disabled {...} input:focus {...} 属性选择器&#xff1a; /*[attribute] [attributevalue] 选择具有对应属性的元素 即使此属性没有实际作…

Postgres和Oracle与Hibernate的兼容性

在某些情况下&#xff0c;您的JEE应用程序需要支持Postgres和Oracle作为数据库。 Hibernate应该在这里完成工作&#xff0c;但是有些细节值得一提。 在为已经运行Oracle的应用程序启用Postgres时&#xff0c;我遇到了以下棘手的部分&#xff1a; BLOBs支持&#xff0c; CLO…

Ubuntu|ython3 :ImportError: cannot import name 'main'

1、问题 rootstephen-K55VD:/usr/bin# pip3 Traceback (most recent call last):File "/usr/bin/pip3", line 9, in <module>from pip import main ImportError: cannot import name main2、解决办法 编辑/usr/bin/pip3&#xff0c;修改代码 from pip._interna…

51Nod 1362 搬箱子 —— 组合数(非质数取模) (差分TLE)

题目&#xff1a;http://www.51nod.com/Challenge/Problem.html#!#problemId1362 首先&#xff0c;\( f[i][j] \) 是一个 \( i \) 次多项式&#xff1b; 如果考虑差分&#xff0c;用一个列向量维护 0 次差分到 \( n \) 次差分即可&#xff0c;在第 \( n \) 次上差分数组已经是一…

java星座查询系统_星座查询示例代码

package api.jisuapi.astro;import api.util.HttpUtil;import net.sf.json.JSONArray;import net.sf.json.JSONObject;public class All {public static final String APPKEY "your_appkey_here";// 你的appkeypublic static final String URL "https://api.j…

Java 8 Friday Goodies:轻松派本地缓存

在Data Geekery &#xff0c;我们喜欢Java。 而且&#xff0c;由于我们真的很喜欢jOOQ的流畅的API和查询DSL &#xff0c;我们对Java 8将为我们的生态系统带来什么感到非常兴奋。 我们已经写了一些关于Java 8好东西的博客 &#xff0c;现在我们觉得是时候开始一个新的博客系列了…

【JS】使用变量作为object的key-方法汇总

1、方法一 1 var a id; 2 var str {a : 12} 3 var obj eval("("str")"); 结果&#xff1a; 1 obj{id:12} 2、方法二 1 var aname; 2 var obj{}; 3 obj[a]jack; 结果&#xff1a; 1 obj{name:jack} 3、方法三 1 var aage; 2 var obj{[a]:23}; 结果&…