SSM+ActiveMQ消息重发机制和消息确认机制Ack

it2023-05-28 75

背景：换了家新公司，接了个ssm架构的老项目，用的mq是ActiveMQ。看代码发现ActiveMQ处理消息时并没看到重试和消息确认操作，以前用的都是RabbitMQ和RocketMQ，本以为ActiveMQ没有消息重发和ACK机制，查阅资料发现还是有的，查阅资料把这个老项目的接口补上，在此记录下。

学习新东西第一原则，看官方文档。

消息重发和ACK机制官方文档地址：

http://activemq.apache.org/message-redelivery-and-dlq-handling.html

http://activemq.apache.org/redelivery-policy

应用场景：

在实际生产场景过程中，当消费者消费消息时，可能由于种种原因，导致消费者消费消息失败。例如：在一个通知系统中，生产者将通知Message 放入队列中，而消费者从队列中将消息读取出来之后，除了进行自身业务处理，还需要调用第三方服务发送短信通知用户，但在发送短信服务的时候，由于网络超时等原有导致消费失败。在这种异常情况下，希望可以建立一种机制。当未发送成功的消息，能够重新发送。处理超过一定次数后还处理不成功的，放弃处理该消息，记录下来。继续对别的消息进行处理。

代码：

mq配置文件

<?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:context="http://www.springframework.org/schema/context" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:aop="http://www.springframework.org/schema/aop" xmlns:tx="http://www.springframework.org/schema/tx" xmlns:p="http://www.springframework.org/schema/p" xmlns:util="http://www.springframework.org/schema/util" xmlns:jdbc="http://www.springframework.org/schema/jdbc" xmlns:cache="http://www.springframework.org/schema/cache" xmlns:amq="http://activemq.apache.org/schema/core" xmlns:jms="http://www.springframework.org/schema/jms" xsi:schemaLocation=" http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx.xsd http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd http://www.springframework.org/schema/cache http://www.springframework.org/schema/cache/spring-cache.xsd http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop.xsd http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms-4.0.xsd http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core-5.14.0.xsd"> <amq:connectionFactory id="amqConnectionFactory" brokerURL="tcp://127.0.0.1:61616?tcpNoDelay=true" userName="admin" password="admin" >  <property name="trustAllPackages" value="true"/> <property name="useAsyncSend" value="true"/> <property name="redeliveryPolicy" ref="activeMQRedeliveryPolicy" />  </amq:connectionFactory> <bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">  <property name="targetConnectionFactory" ref="amqConnectionFactory"/> <property name="reconnectOnException" value="true"/> <property name="cacheConsumers" value="false"/> <property name="cacheProducers" value="false"/>  <property name="sessionCacheSize" value="100" /> </bean> <bean id="activeMQRedeliveryPolicy" class="org.apache.activemq.RedeliveryPolicy">  <property name="useExponentialBackOff" value="true"></property>  <property name="maximumRedeliveries" value="5"></property>  <property name="initialRedeliveryDelay" value="1000"></property>  <property name="backOffMultiplier" value="2"></property>  <property name="maximumRedeliveryDelay" value="1000"></property> </bean>   <bean id="jmsQueueTemplate" class="org.springframework.jms.core.JmsTemplate">  <constructor-arg ref="connectionFactory" />  <property name="explicitQosEnabled" value="true"/>  <property name="deliveryPersistent" value="true"/>  <property name="priority" value="9"/>  <property name="pubSubDomain" value="false" />  <property name="receiveTimeout" value="5000" /> </bean>  <bean id="destinationOne" class="org.apache.activemq.command.ActiveMQQueue"> <constructor-arg index="0" value="newCRM.mq.test"></constructor-arg> </bean>  <bean id="jmsContainerOne" class="org.springframework.jms.listener.DefaultMessageListenerContainer"> <property name="connectionFactory" ref="connectionFactory" /> <property name="destination" ref="destinationOne" /> <property name="messageListener" ref="mqTestMessage" /> <property name="sessionAcknowledgeMode" value="4"></property> </bean>  <bean id="jmsTopicTemplate" class="org.springframework.jms.core.JmsTemplate">  <constructor-arg ref="connectionFactory" />  <property name="pubSubDomain" value="false" /> </bean> </beans>

生产者（直接写的接口，没用JUnit）

/** * @Classname TestController * @Description TODO * @Date 2020/10/19 15:10 * @Created by will */ @Controller @RequestMapping("/test") public class TestController { @Resource private QueueSender queueSender; @ResponseBody @RequestMapping(value="/test",method= RequestMethod.GET) public String test(String con, HttpSession session){ if(con==null){ return "no"; } queueSender.send("newCRM.mq.test",con); return "ok"; } }

消费者

特别注意：

session.recover();不能省略，重发信息使用，注释掉之后不会重发消息，不会从队列移除，也不会进入死信队列 textMessage.acknowledge();确认消息消费成功 /** * @Description 队列消息监听器 * @Date 2020/10/21 11:57 * @Created by will */ @Component public class MqTestMessage implements SessionAwareMessageListener { private static final Logger logger = LoggerFactory.getLogger(MqTestMessage.class.getName()); @Override 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()); //消息确认 textMessage.acknowledge(); System.out.println("--------------------end------------------------"); } catch (Exception e) { //重发信息使用，注释掉之后不会重发消息，不会从队列移除，也不会进入死信队列 session.recover(); System.out.println("我抛异常了哦"); } } } }

正常运行情况

触发重试机制时重试机制

消息最终也进入了MQ的死信队列

activemq入门建议看下这篇文章：https://blog.csdn.net/u012151597/article/details/78904171

最新回复(0)