兼职做调查哪个网站好,商城网页制作,有了域名如何建立网站,木鱼的网站spring jms 事务1.引言 这篇文章将向您展示使用JMS异步接收消息期间使用者执行过程中的错误如何导致消息丢失。 然后#xff0c;我将解释如何使用本地事务解决此问题。 您还将看到这种解决方案在某些情况下可能导致消息重复#xff08;例如#xff0c;当它将消息保存到数据… spring jms 事务 1.引言 这篇文章将向您展示使用JMS异步接收消息期间使用者执行过程中的错误如何导致消息丢失。 然后我将解释如何使用本地事务解决此问题。 您还将看到这种解决方案在某些情况下可能导致消息重复例如当它将消息保存到数据库中然后侦听器执行失败时。 发生这种情况的原因是因为JMS事务独立于其他事务资源如DB。 如果您的处理不是幂等的或者您的应用程序不支持重复消息检测那么您将不得不使用分布式事务。 分布式事务超出了本文的范围。 如果您对处理分布式事务感兴趣可以阅读这篇有趣的文章。 我已经实现了一个再现以下情况的测试应用程序 发送和接收消息消费者将处理收到的消息并将其存储到数据库中。 生产者将消息发送到队列 使用者从队列中检索消息并进行处理 消息处理之前发生错误使用者检索消息但是在将消息存储到DB之前执行失败。 处理消息后发生错误使用者检索消息将其存储到DB然后执行失败。 该应用程序的源代码可以在github上找到。 2.测试应用 测试应用程序执行两个测试类TestNotTransactedMessaging和TestTransactedMessaging 。 这些类都将执行上述三种情况。 让我们看看在没有事务的情况下执行应用程序时的配置。 app-config.xml 应用程序配置。 基本上它会在指定的包中进行检查以自动检测应用Bean生产者和使用者。 它还配置了将在其中存储处理后的通知的内存数据库。 context:component-scan base-packagexpadro.spring.jms.producer, xpadro.spring.jms.receiver/bean idjdbcTemplate classorg.springframework.jdbc.core.JdbcTemplateconstructor-arg refdataSource/
/beanjdbc:embedded-database iddataSourcejdbc:script locationclasspath:db/schema.sql /
/jdbc:embedded-database notx-jms-config.xml 配置JMS基础结构该基础结构是 经纪人联系 JmsTemplate 将通知发送到的队列 侦听器容器它将发送通知给侦听器以处理它们 !-- Infrastructure --
bean idconnectionFactory classorg.apache.activemq.ActiveMQConnectionFactoryproperty namebrokerURL valuevm://embedded?broker.persistentfalse/
/beanbean idcachingConnectionFactory classorg.springframework.jms.connection.CachingConnectionFactoryproperty nametargetConnectionFactory refconnectionFactory/
/beanbean idjmsTemplate classorg.springframework.jms.core.JmsTemplateproperty nameconnectionFactory refcachingConnectionFactory/property namedefaultDestination refincomingQueue/
/bean!-- Destinations --
bean idincomingQueue classorg.apache.activemq.command.ActiveMQQueueconstructor-arg valueincoming.queue/
/bean!-- Listeners --
jms:listener-container connection-factoryconnectionFactoryjms:listener refnotificationProcessor destinationincoming.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 BrowserCallbackInteger() {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方法并因此将其从队列中删除就认为该消息已成功传递。 因为侦听器在将消息存储到DB之前失败所以我们丢失了消息 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 idconnectionFactory classorg.apache.activemq.ActiveMQConnectionFactoryproperty namebrokerURL valuevm://embedded?broker.persistentfalse/property nameredeliveryPolicybean classorg.apache.activemq.RedeliveryPolicyproperty namemaximumRedeliveries value4//bean/property
/bean 接下来我指示侦听器将在事务内执行。 这可以通过修改侦听器容器定义来完成 jms:listener-container connection-factoryconnectionFactory acknowledgetransactedjms:listener refnotificationProcessor destinationincoming.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 在这种情况下这是发生的情况 侦听器检索到消息 它将消息存储到数据库 侦听器执行失败 代理重新发送消息。 由于情况已解决因此侦听器将消息再次存储到DB。 该消息已重复。 7结论 将本地事务添加到消息接收中可避免丢失消息。 我们必须考虑的是可能会出现重复的消息因此我们的侦听器将必须检测到它否则我们的处理必须是幂等的才能再次进行处理。 如果这不可能我们将不得不进行分布式事务因为它们支持涉及不同资源的事务。 参考 Spring JMS在XavierPadró的Blog博客上处理来自JCG合作伙伴 Xavier Padro的事务内的消息 。 翻译自: https://www.javacodegeeks.com/2014/02/spring-jms-processing-messages-within-transactions.htmlspring jms 事务
