ssm--ActiveMQ如何解决数据丢失?消息重发机制和消息确认机制ACK

1. 环境搭建

ssm环境搭建 , 以SSM整合ActiveMQ为例 , 这里一切从简,只为了测试ActiveMQ的消息重发机制和ACK消息签收机制

1.1 目录结构:

在这里插入图片描述

1.2 pom.xml

 <dependencies>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.28</version>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-webmvc</artifactId>
            <version>4.1.7.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-context</artifactId>
            <version>4.1.7.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-test</artifactId>
            <version>4.1.7.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
        </dependency>
        <!--Spring整合activeMQ-->
        <dependency>
            <groupId>org.apache.activemq</groupId>
            <artifactId>activemq-all</artifactId>
            <version>5.14.0</version>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-jms</artifactId>
            <version>4.1.7.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-web</artifactId>
            <version>4.1.7.RELEASE</version>
        </dependency>
    </dependencies>

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 Application</display-name>

    <!--编码过滤器-->
    <filter>
        <filter-name>encoding</filter-name>
        <filter-class>org.springframework.web.filter.CharacterEncodingFilter</filter-class>
        <init-param>
            <param-name>encoding</param-name>
            <param-value>UTF-8</param-value>
        </init-param>
        <init-param>
            <param-name>forceEncoding</param-name>
            <param-value>true</param-value>
        </init-param>
    </filter>
    <filter-mapping>
        <filter-name>encoding</filter-name>
        <url-pattern>/*</url-pattern>
    </filter-mapping>

    <!--前端控制器-->
    <servlet>
        <servlet-name>springmvc</servlet-name>
        <servlet-class>org.springframework.web.servlet.DispatcherServlet</servlet-class>
        <init-param>
            <param-name>contextConfigLocation</param-name>
            <param-value>classpath:spring/*.xml</param-value>
        </init-param>
    </servlet>
    <servlet-mapping>
        <servlet-name>springmvc</servlet-name>
        <url-pattern>/</url-pattern>
    </servlet-mapping>
</web-app>

1.4 resources/spring/springmvc.xml

<?xml version="1.0" encoding="UTF-8"?>
<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>WriteMapNullValue</value>
                        <value>WriteDateUseDateFormat</value>
                    </array>
                </property>
            </bean>
        </mvc:message-converters>
    </mvc:annotation-driven>

    <!--开启组件扫描-->
    <context:component-scan base-package="com.sunmone.controller"/>

    <!--开启mvc注解支持:开启控制器映射器和控制器适配器-->
    <mvc:annotation-driven/>

    <!--视图解析器-->
    <bean id="internalResourceViewResolver"
          class="org.springframework.web.servlet.view.InternalResourceViewResolver">
    </bean>

    <!--释放静态资源-->
    <mvc:default-servlet-handler/>

</beans>

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

<?xml version="1.0" encoding="UTF-8"?>
<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">

    <!--RedeliveryPolicy 生产者消息重发机制-->
    <bean id="activeMQRedeliveryPolicy" class="org.apache.activemq.RedeliveryPolicy">
        <!--是否在每次尝试重新发送失败后,以指数增加超时时间-->
        <property name="useExponentialBackOff" value="true"></property>
        <!--第一次失败后重新发送之前等待500毫秒,
        第二次失败再等待500 * 2毫秒,
        这里的2就是value指数-->
        <property name="backOffMultiplier" value="1"></property>
        <!--重发次数,默认为6次 -1为无限发送-->
        <property name="maximumRedeliveries" value="5"></property>
        <!--重发时间间隔,默认为1秒-->
        <property name="initialRedeliveryDelay" value="500"></property>
        <!--最大传送延迟,只在useExponentialBackOff为true时有效(V5.5),
        假设首次重连间隔为10ms,倍数为2,那么第 二次重连时间间隔为 20ms,第三次重连时间间隔为40ms,
        当重连时间达到最大时间间隔时 也就是value值1000,以后每次重连时间间隔都为最大重连时间间隔-->
        <property name="maximumRedeliveryDelay" value="1000"></property>
    </bean>

    <!-- ActiveMQ 连接工厂 -->
    <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>

    <!--
         ActiveMQ为我们提供了一个PooledConnectionFactory,通过往里面注入一个ActiveMQConnectionFactory
         可以用来将Connection、Session和MessageProducer池化,这样可以大大的减少我们的资源消耗,要依赖于 activemq-pool包

    <bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">
        <property name="connectionFactory" ref="activeMQConnectionFactory"/>
    </bean> -->

    <!-- CachingConnectionFactory 连接工厂 继承了SingleConnectionFactory ,这俩用那个都行-->
    <bean id="CachingConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
        <!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->
        <property name="targetConnectionFactory" ref="activeMQConnectionFactory"/>
        <!--连接失效是否重连  默认为False-->
        <property name="reconnectOnException" value="true"/>
        <!--是否缓存消费者 默认Flase-->
        <property name="cacheConsumers" value="false"/>
        <!--是否缓存生产者 False-->
        <property name="cacheProducers" value="false"/>
        <!-- Session缓存数量 -->
        <property name="sessionCacheSize" value="100"/>
    </bean>


    <!-- Spring JmsTemplate 的消息生产者 start-->
    <!--  定义JmsTemplate的Queue类型 持久化True 超时时间5S-->
    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
        <!-- 指定 ConnectionFactory 工厂对象-->
        <property name="connectionFactory" ref="CachingConnectionFactory"></property>
        <!-- 使 deliveryMode 持久方式, priority 队列优先级, timeToLive过期时间 ,这些设置生效 -->
        <property name="explicitQosEnabled" value="true"/>
        <!-- 发送模式  DeliveryMode.NON_PERSISTENT=1:非持久 ; DeliveryMode.PERSISTENT=2:持久-->
        <property name="deliveryMode" value="2"/>
        <!-- 设置优先级0-9, 默认为4 -->
        <property name="priority" value="9"/>
        <!-- pubSubDomain  默认为False
            True,表示为发布订阅模式 False表示为队列模式  -->
        <property name="pubSubDomain" value="false"/>
        <!-- 接收超时时间 5s -->
        <property name="receiveTimeout" value="5000"/>
    </bean>

    <!-- 指定消息发送空间 点对点 -->
    <bean id="activeMQQueue" class="org.apache.activemq.command.ActiveMQQueue">
        <!--声明一个 Queue队列 -->
        <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>


    <!-- Queue 消息监听器-->
    <bean id="messageListener" class="com.sunmone.listener.MsgListener"></bean>

    <!-- sub 发布订阅 消息监听器-->
    <bean id="topicMsgListener" class="com.sunmone.listener.TopicMsgListener"></bean>


    <!-- Queue 消息监听容器 消息接收监听器用于异步接收消息 -->
    <bean id="jmsContainerOne" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <!--声明连接工厂-->
        <property name="connectionFactory" ref="CachingConnectionFactory"/>
        <!--指定接收消息的服务地址-->
        <property name="destination" ref="activeMQQueue"/>
        <!--指定消息监听器-->
        <property name="messageListener" ref="messageListener"/>
        <!--ACK消息签收机制
            1. AUTO_ACKNOWLEDGE 自动签收
            2. CLIENT_ACKNOWLEDGE 客户端手动确认
            3. DUPS_OK_ACKNOWLEDGE 自动批量确认
            4. INDIVIDUAL_ACKNOWLEDGE  单条消息确认-->
        <property name="sessionAcknowledgeMode" value="4"></property>
    </bean>


    <!-- Topic 消息监听容器 消息接收监听器用于异步接收消息 -->
    <bean id="jmsTopicListener" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <!--声明连接工厂-->
        <property name="connectionFactory" ref="CachingConnectionFactory"/>
        <!--指定接收消息的服务地址-->
        <property name="destination" ref="activeMQTopic"/>
        <!--指定消息监听器-->
        <property name="messageListener" ref="topicMsgListener"/>
        <!--ACK消息签收机制
            1. AUTO_ACKNOWLEDGE 自动签收
            2. CLIENT_ACKNOWLEDGE 客户端手动确认
            3. DUPS_OK_ACKNOWLEDGE 自动批量确认
            4. INDIVIDUAL_ACKNOWLEDGE  单条消息确认-->
        <property name="sessionAcknowledgeMode" value="4"></property>
    </bean>
</beans>

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";
    }

把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++);
            }
        }
    }
}

消费者正常运行情况下 :

在这里插入图片描述

控制台输出 :

在这里插入图片描述


消费者异常情况下 :

在这里插入图片描述

控制台输出

在这里插入图片描述

消息最终进入死信队列

在这里插入图片描述

2.4 总结

可以看出,保证消息的可靠性需要三方配合。

生产者需要处理好Broker的响应,出错情况下利用重试、报警等手段。

Broker需要控制响应的时机,单机情况下是消息刷盘后返回响应,集群多副本情况下,即发送至两个副本及以上的情况下再返回响应。

消费者需要在执行完真正的业务逻辑之后再返回响应给Broker。

但是要注意消息可靠性增强了,性能就下降了,等待消息刷盘、多副本同步后返回都会影响性能。因此还是看业务,例如日志的传输可能丢那么一两条关系不大,因此没必要等消息刷盘再响应


引用: https://www.cnblogs.com/Vincent-yuan/p/15501840.html
借鉴: https://blog.csdn.net/qq_38496561/article/details/109183513


版权声明:本文为weixin_44232093原创文章,遵循CC 4.0 BY-SA版权协议,转载请附上原文出处链接和本声明。