消息队列中间件是分布式系统中重要的组件，已经逐渐成为企业系统内部通信的核心手段。主要功能包括松耦合、异步消息、流量削锋、可靠投递、广播、流量控制、最终一致性等。实现高性能，高可用，可伸缩和最终一致性架构。消息形式支持点对点和订阅-发布。
消息队列中间件常见的应用场景包括应用解耦、异步处理、流量错峰与流控、日志处理等等。目前常见的消息队列中间件有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对象中定义了如下四种：
- SESSION_TRANSACTED= 0：事务提交并确认
- AUTO_ACKNOWLEDGE= 1 ：自动确认
- CLIENT_ACKNOWLEDGE= 2：客户端手动确认
- 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