-
ssm--ActiveMQ如何解决数据丢失?消息重发机制和消息确认机制ACK
1. 环境搭建
ssm环境搭建 , 以SSM整合ActiveMQ为例 , 这里一切从简,只为了测试ActiveMQ的消息重发机制和ACK消息签收机制
1.1 目录结构:
1.2 pom.xml
<dependencies> <dependency> <groupId>com.alibabagroupId> <artifactId>fastjsonartifactId> <version>1.2.28version> dependency> <dependency> <groupId>org.springframeworkgroupId> <artifactId>spring-webmvcartifactId> <version>4.1.7.RELEASEversion> dependency> <dependency> <groupId>org.springframeworkgroupId> <artifactId>spring-contextartifactId> <version>4.1.7.RELEASEversion> dependency> <dependency> <groupId>org.springframeworkgroupId> <artifactId>spring-testartifactId> <version>4.1.7.RELEASEversion> dependency> <dependency> <groupId>junitgroupId> <artifactId>junitartifactId> <version>4.12version> dependency> <dependency> <groupId>org.apache.activemqgroupId> <artifactId>activemq-allartifactId> <version>5.14.0version> dependency> <dependency> <groupId>org.springframeworkgroupId> <artifactId>spring-jmsartifactId> <version>4.1.7.RELEASEversion> dependency> <dependency> <groupId>org.springframeworkgroupId> <artifactId>spring-webartifactId> <version>4.1.7.RELEASEversion> dependency> dependencies>- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
1.3 webapp/WEB-INF/web.xml
<web-app xmlns="http://java.sun.com/xml/ns/javaee" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_3_0.xsd" version="3.0"> <display-name>Archetype Created Web Applicationdisplay-name> <filter> <filter-name>encodingfilter-name> <filter-class>org.springframework.web.filter.CharacterEncodingFilterfilter-class> <init-param> <param-name>encodingparam-name> <param-value>UTF-8param-value> init-param> <init-param> <param-name>forceEncodingparam-name> <param-value>trueparam-value> init-param> filter> <filter-mapping> <filter-name>encodingfilter-name> <url-pattern>/*url-pattern> filter-mapping> <servlet> <servlet-name>springmvcservlet-name> <servlet-class>org.springframework.web.servlet.DispatcherServletservlet-class> <init-param> <param-name>contextConfigLocationparam-name> <param-value>classpath:spring/*.xmlparam-value> init-param> servlet> <servlet-mapping> <servlet-name>springmvcservlet-name> <url-pattern>/url-pattern> servlet-mapping> web-app>- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
1.4 resources/spring/springmvc.xml
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context" xmlns:aop="http://www.springframework.org/schema/aop" xmlns:tx="http://www.springframework.org/schema/tx" xmlns:mvc="http://www.springframework.org/schema/mvc" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx.xsd http://www.springframework.org/schema/mvc http://www.springframework.org/schema/mvc/spring-mvc.xsd"> <mvc:annotation-driven> <mvc:message-converters register-defaults="true"> <bean class="com.alibaba.fastjson.support.spring.FastJsonHttpMessageConverter"> <property name="supportedMediaTypes" value="application/json"/> <property name="features"> <array> <value>WriteMapNullValuevalue> <value>WriteDateUseDateFormatvalue> array> property> bean> mvc:message-converters> mvc:annotation-driven> <context:component-scan base-package="com.sunmone.controller"/> <mvc:annotation-driven/> <bean id="internalResourceViewResolver" class="org.springframework.web.servlet.view.InternalResourceViewResolver"> bean> <mvc:default-servlet-handler/> beans>- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
1.5 resources/spring/acclication-activemq.xml
ActiveMQ官网消息重发机制ACK机制文档:
https://activemq.apache.org/message-redelivery-and-dlq-handling.html
https://activemq.apache.org/redelivery-policy
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd"> <bean id="activeMQRedeliveryPolicy" class="org.apache.activemq.RedeliveryPolicy"> <property name="useExponentialBackOff" value="true">property> <property name="backOffMultiplier" value="1">property> <property name="maximumRedeliveries" value="5">property> <property name="initialRedeliveryDelay" value="500">property> <property name="maximumRedeliveryDelay" value="1000">property> bean> <bean id="activeMQConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"> <constructor-arg name="brokerURL" value="tcp://ip:61616">constructor-arg> <property name="redeliveryPolicy" ref="activeMQRedeliveryPolicy">property> <property name="trustAllPackages" value="true"/> <property name="useAsyncSend" value="true"/> bean> <bean id="CachingConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory"> <property name="targetConnectionFactory" ref="activeMQConnectionFactory"/> <property name="reconnectOnException" value="true"/> <property name="cacheConsumers" value="false"/> <property name="cacheProducers" value="false"/> <property name="sessionCacheSize" value="100"/> bean> <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"> <property name="connectionFactory" ref="CachingConnectionFactory">property> <property name="explicitQosEnabled" value="true"/> <property name="deliveryMode" value="2"/> <property name="priority" value="9"/> <property name="pubSubDomain" value="false"/> <property name="receiveTimeout" value="5000"/> bean> <bean id="activeMQQueue" class="org.apache.activemq.command.ActiveMQQueue"> <constructor-arg value="test.queue">constructor-arg> bean> <bean id="activeMQTopic" class="org.apache.activemq.command.ActiveMQTopic"> <constructor-arg value="test.topic">constructor-arg> bean> <bean id="messageListener" class="com.sunmone.listener.MsgListener">bean> <bean id="topicMsgListener" class="com.sunmone.listener.TopicMsgListener">bean> <bean id="jmsContainerOne" class="org.springframework.jms.listener.DefaultMessageListenerContainer"> <property name="connectionFactory" ref="CachingConnectionFactory"/> <property name="destination" ref="activeMQQueue"/> <property name="messageListener" ref="messageListener"/> <property name="sessionAcknowledgeMode" value="4">property> bean> <bean id="jmsTopicListener" class="org.springframework.jms.listener.DefaultMessageListenerContainer"> <property name="connectionFactory" ref="CachingConnectionFactory"/> <property name="destination" ref="activeMQTopic"/> <property name="messageListener" ref="topicMsgListener"/> <property name="sessionAcknowledgeMode" value="4">property> bean> beans>- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 84
- 85
- 86
- 87
- 88
- 89
- 90
- 91
- 92
- 93
- 94
- 95
- 96
- 97
- 98
- 99
- 100
- 101
- 102
- 103
- 104
- 105
- 106
- 107
- 108
- 109
- 110
- 111
- 112
- 113
- 114
- 115
- 116
- 117
- 118
- 119
- 120
- 121
- 122
- 123
- 124
- 125
- 126
- 127
2. 如何解决数据丢失?
先来看看这张图
可以看到一共有三个阶段，分别是生产消息、存储消息和消费消息。我们从这三个阶段分别入手来看看如何确保消息不会丢失
2.1 生产者
生产者发送消息至Broker，需要处理Broker的响应，不论是同步还是异步发送消息，同步和异步回调都需要做好try-catch，妥善的处理响应，如果Broker返回写入失败等错误消息，需要重试发送。当多次发送失败需要作报警，日志记录等。
这样就能保证在生产消息阶段消息不会丢失
2.1.1 生产者代码
创建TestController作为消息生产者
@RestController @RequestMapping("/test") public class TestController { // 注入消息模板对象 @Autowired private JmsTemplate jmsTemplate; // 注入点对点消息空间对象 @Autowired private ActiveMQQueue activeMQQueue; // 注入发布订阅消息空间对象 @Autowired private ActiveMQTopic activeMQTopic; /* * Queue 生产者生产消息 * */ @RequestMapping(value = "/sendMsg", method = RequestMethod.GET) public String sendMsg(final String msg) { if (msg == null) { return "error..."; } try { jmsTemplate.send(activeMQQueue, new MessageCreator() { public Message createMessage(Session session) throws JMSException { return session.createTextMessage(msg); } }); } catch (JmsException e) { try { // 消息发送失败 , 重新发送消息 jmsTemplate.convertAndSend(activeMQTopic, "服务器宕机,消息重新发送......"); } catch (JmsException ex) { System.out.println("再次发送失败,服务器确认宕机,记录并通知"); ex.printStackTrace(); return "error..."; } e.printStackTrace(); return "error..."; } return "ok"; }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
把ActiveMQ服务器关掉,访问消息生产者,模拟服务器宕机
控制台输出
可以看到,程序正常响应到服务器宕机的情况,保存并记录
因为我们配置了服务器连接失败自动重连
所以此时Spring正在积极帮我们重连
由此可见,此策略可以保证消息生产者在生产消息阶段,消息不会丢失
2.2 存储消息
存储消息阶段需要在消息刷盘之后再给生产者响应，假设消息写入缓存中就返回响应，那么机器突然断电这消息就没了，而生产者以为已经发送成功了。
如果Broker是集群部署，有多副本机制，即消息不仅仅要写入当前Broker,还需要写入副本机中。那配置成至少写入两台机子后再给生产者响应。这样基本上就能保证存储的可靠了。一台挂了还有一台还在呢（假如怕两台都挂了…那就再多些）
2.3 消费者
当消费者拿到消息之后直接存入内存队列中就直接返回给Broker消费成功，这是不对的。
你需要考虑拿到消息放在内存之后消费者就宕机了怎么办。所以我们应该在消费者真正执行完业务逻辑之后，利用ACK的单条消息签收机制,发送给Broker消费成功的响应，这才是真正的消费了
所以只要我们在消息业务逻辑处理完成之后再给Broker响应，那么消费阶段消息就不会丢失。
2.3.1 消费者代码
session.recover() : 不能省略，重发信息使用，注释掉之后不会重发消息，不会从队列移除，也不会进入死信队列
textMessage.acknowledge() : 确认消息消费成功
创建 MsgListener 监听消息
public class MsgListener implements SessionAwareMessageListener { public static int num = 1; public void onMessage(Message message, Session session) throws JMSException { if (message instanceof TextMessage) { System.out.println("--------------------start------------------------"); TextMessage textMessage = (TextMessage) message; try { if ("重发机制".equals(textMessage.getText())) { System.out.println("----------------"); throw new RuntimeException("抛出异常,模拟消息消费失败场景..."); } System.out.println(textMessage.getText()); // 手动签收消息,在业务执行完毕后执行 调用acknowledge(), 消息才会被Broker端确认消费 textMessage.acknowledge(); System.out.println("--------------------end------------------------"); } catch (Exception e) { //重发信息使用，注释掉之后不会重发消息，不会从队列移除，也不会进入死信队列 session.recover(); System.out.println("消费异常...消息重发机制触发 : " + num++); } } } }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
消费者正常运行情况下 :
控制台输出 :
消费者异常情况下 :
控制台输出
消息最终进入死信队列
2.4 总结
可以看出，保证消息的可靠性需要三方配合。
生产者需要处理好Broker的响应，出错情况下利用重试、报警等手段。
Broker需要控制响应的时机，单机情况下是消息刷盘后返回响应，集群多副本情况下，即发送至两个副本及以上的情况下再返回响应。
消费者需要在执行完真正的业务逻辑之后再返回响应给Broker。
但是要注意消息可靠性增强了，性能就下降了，等待消息刷盘、多副本同步后返回都会影响性能。因此还是看业务，例如日志的传输可能丢那么一两条关系不大，因此没必要等消息刷盘再响应
引用: https://www.cnblogs.com/Vincent-yuan/p/15501840.html
借鉴: https://blog.csdn.net/qq_38496561/article/details/109183513 -
相关阅读:
29. 【Android教程】折叠列表 ExpandableListView
经典场的量子化
开发实战经验分享：互联网医院系统源码与在线问诊APP搭建
Redis详解
正则表达式大全，30个正则表达式详细案例
网页的用户注册功能
Android中的Drawable（三）
CentOS7 内核升级
多媒体应用设计师
libnice 源码分析
- 原文地址：https://blog.csdn.net/weixin_44232093/article/details/126462331