Spring Boot 整合 ActiveMQ 实现手动确认和重发消息

    消息队列中间件是分布式系统中重要的组件，已经逐渐成为企业系统内部通信的核心手段。主要功能包括松耦合、异步消息、流量削锋、可靠投递、广播、流量控制、最终一致性等。实现高性能，高可用，可伸缩和最终一致性架构。消息形式支持点对点和订阅-发布。

   消息队列中间件常见的应用场景包括应用解耦、异步处理、流量错峰与流控、日志处理等等。目前常见的消息队列中间件有ActiveMQ、RabbitMQ、ZeroMQ、Kafka、MetaMQ和RocketMQ等。本文在Spring Boot整合ActiveMQ的基础上，以点对点模式为例介绍两种消息处理机制——手动确认和重发消息。

   之所以介绍消息重发机制，是因为ActiveMQ被消费失败的消息将会驻留在队列中，因为没有进行消息确认，所以，下次还会监听到这条消息。

软件环境

• Apache ActiveMQ 5.15.9
• Java version 11
• IntelliJ IDEA 2020.2 (Ultimate Edition)
• Spring Boot 2.3.1.RELEASE

 

1.引入ActiveMQ pom 坐标和基本配置

   引入pom坐标： 

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-activemq</artifactId>
            <version>2.3.3.RELEASE</version>
        </dependency>

2. 修改配置文件

spring.activemq.broker-url=tcp://localhost:61616
spring.activemq.in-memory=true spring.activemq.pool.enabled=false #默认值false，表示点到点模式，true时代表发布订阅模式 spring.jms.pub-sub-domain=false spring.activemq.user=wiener spring.activemq.password=wiener123

3. SpringBoot 的启动类配置

   在 SpringBoot 的启动类上加上一个 @EnableJms 注解。我的项目去掉此注解也可以正常启动，故没有添加，你的项目如果启动失败，请添加。　

4. ActiveMQ 连接配置

   当我们发送消息的时候，会出现发送失败的情况，此时我们需要用到ActiveMQ提供的消息重发机制，重新发送消息。那么问题来了，我们怎么知道消息是否发送成功呢？ActiveMQ还有消息确认机制，消费者在接收到消息的时候可以进行确认。本节将学习ActiveMQ消息的确认机制和重发机制。

   消息确认机制有五种，在session对象中定义了如下四种：

1. SESSION_TRANSACTED= 0：事务提交并确认
2. AUTO_ACKNOWLEDGE= 1 ：自动确认
3. CLIENT_ACKNOWLEDGE= 2：客户端手动确认
4. UPS_OK_ACKNOWLEDGE= 3： 自动批量确认

   在ActiveMQSession类中补充了一个自定义的ACK机制:

   INDIVIDUAL_ACKNOWLEDGE= 4：单条消息确认。

 

   消息确认机制的使用场景分两种：

   1、带事务的session

   如果session带有事务，并且事务成功提交，则消息被自动签收。如果事务回滚，则消息会被再次传送。

   2、不带事务的session

   不带事务的session的签收方式，取决于session的配置。

   新建一个配置类ActiveMQConfig用来配置 ActiveMQ 连接属性，在工厂中设置开启单条消息确认模式INDIVIDUAL_ACKNOWLEDGE。注意：ActiveMQ 默认是开启事务的，且消息确认机制与事务机制是冲突的，只能二选一，所以演示消息确认前，请先关闭事务。

   配置代码如下：

import org.apache.activemq.ActiveMQConnectionFactory; import org.apache.activemq.ActiveMQSession; import org.apache.activemq.RedeliveryPolicy; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.jms.config.DefaultJmsListenerContainerFactory; import org.springframework.jms.config.JmsListenerContainerFactory; import org.springframework.jms.config.SimpleJmsListenerContainerFactory; /** * 配置 Active MQ * @author Wiener * @date 2020/9/15 */ @Configuration public class ActiveMQConfig { @Value("${spring.activemq.user}") private String userName; @Value("${spring.activemq.password}") private String password; @Value("${spring.activemq.broker-url}") private String brokerUrl; /** * 配置名字为givenConnectionFactory的连接工厂 * @return
     */ @Bean("givenConnectionFactory") public ActiveMQConnectionFactory connectionFactory() { ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(userName, password, brokerUrl); // 设置重发机制
         RedeliveryPolicy policy = new RedeliveryPolicy(); policy.setUseExponentialBackOff(Boolean.TRUE); // 消息处理失败重新处理次数,默认为6次
         policy.setMaximumRedeliveries(2); // 重发时间间隔,默认为1秒
         policy.setInitialRedeliveryDelay(1000L); policy.setBackOffMultiplier(2); policy.setMaximumRedeliveryDelay(1000L); activeMQConnectionFactory.setRedeliveryPolicy(policy); return activeMQConnectionFactory; } /** * 在Queue模式中，对消息的监听需要对containerFactory进行配置 * * @param givenConnectionFactory * @return
     */ @Bean("queueListener") public JmsListenerContainerFactory<?> queueJmsListenerContainerFactory(ActiveMQConnectionFactory givenConnectionFactory) { SimpleJmsListenerContainerFactory factory = new SimpleJmsListenerContainerFactory(); // 关闭事务
         factory.setSessionTransacted(false); // 手动确认消息
 factory.setSessionAcknowledgeMode(ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE); factory.setPubSubDomain(false); factory.setConnectionFactory(givenConnectionFactory); return factory; } @Bean("topicListener") public JmsListenerContainerFactory<?> jmsListenerContainerTopic(ActiveMQConnectionFactory givenConnectionFactory){ //设置为发布订阅模式, 默认情况下使用生产消费者方式
         DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory(); bean.setPubSubDomain(true); bean.setConnectionFactory(givenConnectionFactory); return bean; } }

这里设置消息发送失败时重发次数为2次，其系统默认值为6次，源码如下：

 

5.手动确认和重发消息

   JmsTemplate是消息处理核心类（线程安全），用于发送和接收消息，在发送或者是消费消息的同时也会对所需资源进行创建和释放。消息消费采用receive方式（同步、阻塞的），这里不做具体描述。关于消息的发送，常用的方式为send()以及convertAndSend()两种，其中send()发送需要指定消息格式，使用convertAndSend可以根据定义好的MessageConvert自动转换消息格式。

   消息生产者代码比较简单，主要是调用 jmsMessagingTemplate 的 convertAndSend() 方法。发送的消息如果是对象类型的，可以将其转成 json 字符串。

@Service("producer") public class Producer { /** * 也可以注入JmsTemplate，JmsMessagingTemplate对JmsTemplate进行了封装 */ @Autowired private JmsMessagingTemplate jmsTemplate; /** * 发送消息 * * @param destination 指定消息目的地 * @param message 待发送的消息 */
    public void sendMessage(Destination destination, final String message) { jmsTemplate.convertAndSend(destination, message); } }

其中，Student 类的定义如下：

/** * @author Wiener */ @Getter @Setter @ToString @Component public class Student implements Serializable { private static final long serialVersionUID = 4481359068243928443L; private Long id; /** 姓名 */
    private String name; private String birthDay; public Student() { } public Student(Long id, String name, String birthDay) { this.id = id; this.name = name; this.birthDay = birthDay; } }

   在消费者监听类中，我们使用@JmsListener监听队列消息。大家请注意，如果我们不在@JmsListener中指定containerFactory，那么将使用默认配置，默认配置中Session是开启了事务的，即使我们设置了手动Ack也是无效的。

   定义两个名为 "east7-queue" 的消费者，在消费消息时，二者默认会轮询消费。在消费者消费完一条消息之后，调用 message.acknowledge() 进行消息的手动确认。如果在消费者中未进行手动确认，由于 ActiveMQ 进行了持久化消息，那么在下次启动项目的时候还会再次发送该消息。

import com.eg.wiener.dto.Student; import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; import org.apache.activemq.command.ActiveMQMessage; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.jms.annotation.JmsListener; import org.springframework.stereotype.Component; import javax.jms.JMSException; import javax.jms.Session; import javax.jms.TextMessage; @Component public class ConsumerListener { private static Logger logger = LoggerFactory.getLogger(ConsumerListener.class); // 记录重发总次数
    private static int num = 0; /** * east7-queue普通队列：消费者1 */ @JmsListener(destination = "east7-queue", containerFactory = "queueListener") public void receiveQueueTest1(ActiveMQMessage message, Session session) throws JsonProcessingException, JMSException { logger.info("receiveQueueTest:1-mode {}", session.getAcknowledgeMode()); String text = null; if (message instanceof TextMessage) { TextMessage textMessage = (TextMessage) message; ObjectMapper objectMapper = new ObjectMapper(); Student student = objectMapper.readValue(textMessage.getText(), Student.class); logger.info("queue1接收到student：{}", student); // 手动确认
              try { message.acknowledge(); } catch (JMSException e) { // 此不可省略 重发信息使用
 session.recover(); } } } /** * east7-queue普通队列：消费者2 */ @JmsListener(destination = "east7-queue", containerFactory = "queueListener") public void receiveQueueTest2(ActiveMQMessage message, Session session) throws JMSException, JsonProcessingException { logger.info("receiveQueueTest:2-transacted mode {}", session.getTransacted()); String msg = null; if (message instanceof TextMessage) { TextMessage textMessage = (TextMessage) message; msg = textMessage.getText(); ObjectMapper objectMapper = new ObjectMapper(); Student student = objectMapper.readValue(msg, Student.class); logger.info("queue2接收到student：{}", student); try { if (msg.contains("典韦5") || msg.contains("典韦6") || msg.contains("典韦7")) { throw new JMSException("故意抛出的异常"); } message.acknowledge(); } catch (JMSException e) { ++ num; System.out.println(String.format("触发重发机制，num = %s, msg = %s", num, msg)); session.recover(); } } } }

   这里为了验证消息重发机制，故意在消费消息的时候抛出了一个异常。我们的Junit测试用例如下：

 @Test public void ackMsgTest() throws JsonProcessingException { Destination destination = new ActiveMQQueue("east7-queue"); Student student = new Student(1L, "张三", DateUtils.parseDateSM(new Date())); ObjectMapper objectMapper = new ObjectMapper(); for (int i = 0; i < 10; i++) { student.setId(i * 10L); student.setName("典韦" + i); producer.sendMessage(destination, objectMapper.writeValueAsString(student)); } sleep(5000L); //给消费者足够的时间消费消息
 } private void sleep(long time) { try { Thread.sleep(time); } catch (InterruptedException e) { e.printStackTrace(); } }

   执行测试函数后，可以发现两个消费者交替消费消息。而且可以发现，每个错误消息重发次数为3次，与我们的预期吻合。

触发重发机制，num = 1, msg = {"id":500,"name":"典韦5","birthDay":"2020-09-20 16:31:39"} 触发重发机制，num = 2, msg = {"id":500,"name":"典韦5","birthDay":"2020-09-20 16:31:39"} 触发重发机制，num = 3, msg = {"id":500,"name":"典韦5","birthDay":"2020-09-20 16:31:39"} 触发重发机制，num = 4, msg = {"id":700,"name":"典韦7","birthDay":"2020-09-20 16:31:39"} 触发重发机制，num = 5, msg = {"id":700,"name":"典韦7","birthDay":"2020-09-20 16:31:39"} 触发重发机制，num = 6, msg = {"id":700,"name":"典韦7","birthDay":"2020-09-20 16:31:39"}

Reference

https://www.cnblogs.com/liuyuan1227/p/10776189.html

https://blog.csdn.net/chinese_cai/article/details/108342611

https://blog.csdn.net/p_programmer/article/details/88384138