学习目标
- 能够测试生产者重试机制
- 能够测试生产者确认机制
- 能够说出生产者确认机制的两种方式
- 能够说出发送失败处理机制
- 能够说出消息持久化机制
- 能够测试消费者确认机制
- 能够说出消费失败重试机制
- 能够说出MQ消息幂等性方案
- 能够说出延迟消息方案
- 能够实现自动取消超时未支付订单功能
- 能够说出保证消息的可靠性的完整方案
1 消息可靠性
1.1. 思路分析
在昨天的练习作业中，我们改造了余额支付功能，在支付成功后利用RabbitMQ通知交易服务，更新业务订单状态为已支付。但是大家思考一下：如果这里MQ通知失败，支付服务中支付流水显示支付成功，而交易服务中的订单状态却显示未支付，数据出现了不一致。
首先，我们一起分析一下消息丢失的可能性有哪些。
消息从发送者发送消息，到消费者处理消息，需要经过的流程是这样的：
消息从生产者到消费者的每一步都可能导致消息丢失：
- 发送消息时丢失：
- 生产者发送消息时连接MQ失败
- 生产者发送消息到达MQ后未找到
Exchange - 生产者发送消息到达MQ的
Exchange后，未找到合适的Queue - 消息到达MQ后，处理消息的进程发生异常
- MQ导致消息丢失：
- 消息到达MQ，保存到队列后，尚未消费就突然宕机
- 消费者处理消息时：
- 消息接收后尚未处理突然宕机
- 消息接收后处理过程中抛出异常
综上，我们要解决消息丢失问题，保证MQ的可靠性，就必须从3个方面入手：
- 保证生产消息的可靠性
- 确保MQ不会将消息弄丢
- 保证消费消息的可靠性
注意：使用MQ并不是所有场景对消息的可靠性要求都很高，比如上图中，支付成功短信通知的流程对消息可靠性要求就不高，通常都可以保证消息正常到达消费者，即使个别没有成功通知用户也不影响主体业务流程，所以在设计技术方案时一定要根据业务需求具体分析。
1.2 生产消息可靠性
1.2.1. 生产者重试机制
1.2.1.1 配置
首先第一种情况，就是生产者发送消息时，出现了网络故障，导致与MQ的连接中断。
为了解决这个问题，SpringAMQP提供的消息发送时的重试机制。即：当RabbitTemplate与MQ连接超时后，多次重试。
从课程资料中找到mq-demo-v2.zip，并解压，使用IDEA打开mq-demo-v2工程。
修改publisher模块的application.yaml文件，添加下面的内容：
spring: rabbitmq: host: 192.168.101.68 # 你的虚拟机IP port: 5672 # 端口 virtual-host: /hmall # 虚拟主机【注意这里可能不对】 username: hmall # 用户名【注意这里可能不对】 password: 123 # 密码【注意这里可能不对】 connection-timeout: 1s # 设置MQ的连接超时时间 template: retry: enabled: true # 开启超时重试机制 initial-interval: 1000ms # 失败后的初始等待时间 multiplier: 1 # 失败后下次的等待时长倍数，下次等待时长 = 上次等待时长 * multiplier max-attempts: 3 # 总共尝试次数 datasource: url: jdbc:mysql://192.168.101.68:3306/hmall?useUnicode=true&characterEncoding=UTF-8&autoReconnect=true&serverTimezone=Asia/Shanghai driver-class-name: com.mysql.cj.jdbc.Driver username: root password: mysql mybatis-plus: configuration: default-enum-type-handler: com.baomidou.mybatisplus.core.handlers.MybatisEnumTypeHandler global-config: db-config: update-strategy: not_null id-type: auto logging: pattern: dateformat: MM-dd HH:mm:ss:SSS level: com.itheima: debug rabbit-mq: enable: true persistence: enable: true
配置参数解释
initial-interval: 失败后的初始等待时间multiplier: 倍增器，每次重试的等待时间是前一次几倍。
失败后下次等待时长 =上次等待时长 * multiplier
max-attempts: 最大重试次数（包括第一次尝试）
举例：
由于multiplier设置为1，这意味着每次重试之间的间隔是固定的，不会增加。
假设在t=0时刻首次尝试发送消息，如果发送失败，则会按照以下时间点进行重试：
- 第一次尝试（也是首次发送）：t=0（假设即时失败）
- 第一次重试：等待1秒后重试，t=1秒（首次失败后等待1秒）
- 第二次重试：等待1秒*1=1秒 后重试，t=2秒（从第一次重试再等待1秒）
如果设置如下：
initial-interval:1000ms multiplier:2 max-attempts: 5
由于multiplier设置为2，这意味着每次重试之间的间隔会翻倍。
假设在t=0时刻首次尝试发送消息，如果发送失败，则会按照以下时间点进行重试：
- 第一次尝试（也是首次发送）：t=0（假设即时失败）
- 第一次重试：等待1秒后重试，t=1秒（首次失败后等待1秒）
- 第二次重试：等待1*2=2秒 后重试，t=3秒
- 第三次重试：等待2*2=4秒 后重试，t=7秒
- 第四次重试：等待4*2=8秒 后重试，t=15秒
1.2.1.2 测试
我们利用命令停掉RabbitMQ服务：
docker stop mq
然后测试发送一条消息，会发现会每隔1秒重试1次，总共重试了3次。消息发送超时重试机制配置成功！
01-15 17:37:58:252 INFO 18792 --- [ main] o.s.a.r.c.CachingConnectionFactory : Attempting to connect to: [192.168.101.68:5672] 01-15 17:38:00:281 INFO 18792 --- [ main] o.s.a.r.c.CachingConnectionFactory : Attempting to connect to: [192.168.101.68:5672] 01-15 17:38:02:306 INFO 18792 --- [ main] o.s.a.r.c.CachingConnectionFactory : Attempting to connect to: [192.168.101.68:5672] org.springframework.amqp.AmqpIOException: java.net.SocketTimeoutException: connect timed out
注意：当网络不稳定的时候，利用重试机制可以有效提高消息发送的成功率。不过SpringAMQP提供的重试机制是阻塞式的重试，也就是说多次重试等待的过程中，当前线程是被阻塞的。
如果对于业务性能有要求，建议禁用重试机制。如果一定要使用，请合理配置等待时长和重试次数，当然也可以考虑使用异步线程来执行发送消息的代码。
1.2.2. 生产者确认机制
1.2.2.1 两种机制介绍
一般情况下，只要生产者与MQ之间的网路连接顺畅，基本不会出现发送消息丢失的情况，因此大多数情况下我们无需考虑这种问题。
不过，在少数情况下，也会出现消息发送到MQ之后丢失的现象，比如：
- MQ内部处理消息的进程发生了异常
- 生产者发送消息到达MQ后未找到
Exchange - 生产者发送消息到达MQ的
Exchange后，未找到合适的Queue，因此无法路由
针对上述情况，RabbitMQ提供生产者确认机制，包括Publisher Confirm和Publisher Return两种。在开启确认机制的情况下，当生产者发送消息给MQ后，MQ会根据消息处理的情况返回不同的回执。
具体如图所示：
生产者确认机制：
1.Publisher Return
消息投递成功但路由失败会调用Publisher Return回调方法返回异常信息。
2.Publisher Confirm
消息投递成功返回ack，投递失败返回nack。
注意：消息投递成功但可能路由失败了，此时会通过Publisher Confirm返回ack，通过Publisher Return回调方法返回异常信息。
默认两种机制都是关闭状态，需要通过配置文件来开启。
生产者确认机制：确保消息发送到MQ的一种机制，它通过2个函数来确认
（1）【异常】找不到交换机：publisher-confirm：nack
（2）【异常】找到交换机，匹配不到队列：publisher-return：ack
（3）【正常】找到交换机，匹配到队列：publisher-cofirm：ack
1.2.2.2 开启生产者确认
在publisher模块的application.yaml中添加配置：
spring: rabbitmq: publisher-confirm-type: correlated # 开启publisher confirm机制，并设置confirm类型 publisher-returns: true # 开启publisher return机制
这里publisher-confirm-type有三种模式可选：
none：关闭confirm机制simple：同步阻塞等待MQ的回执(回调方法)correlated：MQ异步回调返回回执
一般我们推荐使用correlated，回调机制。
1.2.2.3 实现方法
ReturnCallback
每个RabbitTemplate只能配置一个ReturnCallback，因此我们可以在配置类中统一设置。我们在publisher模块定义一个配置类：
内容如下：
package com.itheima.publisher.config; import lombok.AllArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.amqp.core.ReturnedMessage; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.context.annotation.Configuration; import javax.annotation.PostConstruct; @Slf4j @AllArgsConstructor @Configuration public class MqConfig { private final RabbitTemplate rabbitTemplate; @PostConstruct public void init(){ rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() { @Override public void returnedMessage(ReturnedMessage returned) { log.error("触发return callback,"); log.debug("exchange: {}", returned.getExchange()); log.debug("routingKey: {}", returned.getRoutingKey()); log.debug("message: {}", returned.getMessage()); log.debug("replyCode: {}", returned.getReplyCode()); log.debug("replyText: {}", returned.getReplyText()); } }); } }
ConfirmCallback
由于每个消息发送时的处理逻辑不一定相同，因此ConfirmCallback需要在每次发消息时定义。具体来说，是在调用RabbitTemplate中的convertAndSend方法时，多传递一个参数：
这里的CorrelationData中包含两个核心的东西：
id：消息的唯一标示，MQ对不同的消息的回执以此做判断，避免混淆SettableListenableFuture：回执结果的Future对象
将来MQ的回执就会通过这个Future来返回，我们可以提前给CorrelationData中的Future添加回调函数来处理消息回执：
我们测试下边的方法，向系统自带的交换机发送消息，并且添加ConfirmCallback：
注意：此代码不用编写直接测试即可，稍后我们会用工具类替代。
@Test void testPublisherConfirm() { // 1.创建CorrelationData CorrelationData cd = new CorrelationData(); // 2.给Future添加ConfirmCallback cd.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() { @Override public void onFailure(Throwable ex) { // 2.1.Future发生异常时的处理逻辑，基本不会触发 log.error("send message fail", ex); } @Override public void onSuccess(CorrelationData.Confirm result) { // 2.2.Future接收到回执的处理逻辑，参数中的result就是回执内容 if(result.isAck()){ // result.isAck()，boolean类型，true代表ack回执，false 代表 nack回执 log.debug("发送消息成功，收到 ack!"); }else{ // result.getReason()，String类型，返回nack时的异常描述 log.error("发送消息失败，收到 nack, reason : {}", result.getReason()); } } }); // 3.发送消息,故意指定一个错误的rontingKey rabbitTemplate.convertAndSend("hmall.direct", "q", "hello", cd); }
1.2.2.4 测试
测试报错：
原因：每个RabbitTemplate只支持一个ReturnCallback。
解决：
屏蔽common-rabbitmq依赖，在common-rabbitmq中对RabbitTemplate设置了ReturnCallback
测试步骤：
可以看到，由于传递的RoutingKey是错误的，路由失败后，触发了return callback，同时也收到了ack。
当我们修改为正确的RoutingKey以后，就不会触发return callback了，只收到ack。
当我们把交换机名称修改错误则只会收到nack。
1.2.3. 发送失败处理机制
1.2.3.1 失败处理机制
在ConfirmCallback中收到nack表示消息投递失败，ReturnCallback异常表示路由失败
高频面试题：消息投递失败怎么处理/你们怎么保证消费成功
可以将消息记录到失败消息表，由定时任务进行发布，每隔10秒钟(可设置)执行获取失败消息重新发送，发送一次则在失败次数字段加一，达到3次停止自动发送由人工处理(如钉钉告警)。
在commonn-rabbitmq模块中实现了发送消息的工具方法，此方法实现了发送失败处理机制。
ReturnCallback回调逻辑在com.itheima.common.rabbitmq.config.RabbitMqConfiguration中，核心代码如下：
@Configuration @ConditionalOnProperty(prefix = "rabbit-mq", name = "enable", havingValue = "true") @Import({RabbitClient.class, FailMsgDaoImpl.class}) @Slf4j public class RabbitMqConfiguration implements ApplicationContextAware { /** * 并发数量 */ public static final int DEFAULT_CONCURRENT = 10; @Autowired(required = false) private FailMsgDao failMsgDao; @Override public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { // 获取RabbitTemplate RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class); //定义returnCallback回调方法 rabbitTemplate.setReturnsCallback( new RabbitTemplate.ReturnsCallback() { @Override public void returnedMessage(ReturnedMessage returnedMessage) { byte[] body = returnedMessage.getMessage().getBody(); //消息id String messageId = returnedMessage.getMessage().getMessageProperties().getMessageId(); String content = new String(body, Charset.defaultCharset()); log.info("消息发送失败，应答码{}，原因{}，交换机{}，路由键{},消息id{},消息内容{}", returnedMessage.getReplyCode(), returnedMessage.getReplyText(), returnedMessage.getExchange(), returnedMessage.getRoutingKey(), messageId, content); if (failMsgDao != null) { // 失败消息落库（后续定时任务重试，达最大次数人工干预） failMsgDao.save(messageId, returnedMessage.getExchange(), returnedMessage.getRoutingKey(), content, 0, DateUtils.getCurrentTime()+10, "returnCallback"); } } } ); } }
ApplicationContextAware 的作用：如果 Bean 实现了 ApplicationContextAware 接口，Spring 容器会调用 setApplicationContext 方法，将 ApplicationContext 传递给该 Bean。
Bean 创建：Spring 容器首先创建 Bean 实例。
属性注入：Spring 容器对 Bean 的属性进行依赖注入。
Aware 接口回调：如果 Bean 实现了 ApplicationContextAware 接口，Spring 容器会调用 setApplicationContext 方法，将 ApplicationContext 传递给该 Bean。
初始化方法：如果 Bean 配置了初始化方法（例如通过 @PostConstruct 注解或 init-method 属性），Spring 容器会调用这些初始化方法。
以上可以详见：Spring源码解析
ConfirmCallback的逻辑在RabbitClient 工具类的sendMsg方法中，通过sendMsg方法发送消息，在发送消息时指定correlationData对象，在correlationData对象中指定了ConfirmCallback回调方法的逻辑，当返回nack会将消息写入失败表，如果消息重发成功会将该记录从失败表删除。
核心代码如下：
@Slf4j @Service public class RabbitClient { .... @Resource private RabbitTemplate rabbitTemplate; @Autowired(required = false) private FailMsgDao failMsgDao; /** * 发送消息 重试3次 * * @param exchange 交换机 * @param routingKey 路由key * @param msg 消息对象，会将对象序列化成json字符串发出 * @param delay 延迟时间 秒 * @param msgId 消息id * @param isFailMsg 是否是失败消息 * @return 是否发送成功 */ @Retryable(value = MqException.class, maxAttempts = 3, backoff = @Backoff(value = 3000, multiplier = 1.5), recover = "saveFailMag") public void sendMsg(String exchange, String routingKey, Object msg, Integer delay, String msgId, boolean isFailMsg) { // 1.发送消息前准备 // 1.1获取消息内容，如果非字符串将其序列化 String jsonMsg = JsonUtils.toJsonStr(msg); // 1.2.全局唯一消息id，如果调用者设置了消息id，使用调用者消息id，如果为配置，默认雪花算法生成消息id if(StrUtil.isBlank(msgId)){ msgId = IdUtil.getSnowflakeNextIdStr(); } // 1.3.设置默认延迟时间，默认立即发送 delay = NumberUtils.null2Default(delay, -1); log.debug("消息发送！exchange = {}, routingKey = {}, msg = {}, msgId = {}", exchange, routingKey, jsonMsg, msgId); // 1.4.构建回调 RabbitMqListenableFutureCallback futureCallback = RabbitMqListenableFutureCallback.builder() .exchange(exchange) .routingKey(routingKey) .msg(jsonMsg) .msgId(msgId) .delay(delay) .isFailMsg(isFailMsg) .failMsgDao(failMsgDao) .build(); // 1.5.CorrelationData设置 CorrelationData correlationData = new CorrelationData(msgId.toString()); correlationData.getFuture().addCallback(futureCallback); // 1.6.构造消息对象 Message message = MessageBuilder.withBody(StrUtil.bytes(jsonMsg, CharsetUtil.CHARSET_UTF_8)) //持久化 .setDeliveryMode(MessageDeliveryMode.PERSISTENT) //消息id .setMessageId(msgId.toString()) .build(); try { // 2.发送消息 this.rabbitTemplate.convertAndSend(exchange, routingKey, message, new DelayMessagePostProcessor(delay), correlationData); } catch (Exception e) { log.error("send error:" + e); // 3.构建异常回调，并抛出异常 MqException mqException = new MqException(); mqException.setMsg(ExceptionUtil.getMessage(e)); mqException.setMqId(msgId); throw mqException; } } public class RabbitMqListenableFutureCallback implements ListenableFutureCallback<CorrelationData.Confirm> { //记录失败消息service private FailMsgDao failMsgDao; private String exchange; private String routingKey; private String msg; private String msgId; private Integer delay; //是否是失败消息 private boolean isFailMsg=false; @Override public void onFailure(Throwable ex) { if(failMsgDao == null) { return; } // 执行失败保存失败信息 failMsgDao.save(msgId, exchange, routingKey, msg, delay, DateUtils.getCurrentTime() + 10, ExceptionUtil.getMessage(ex)); } @Override public void onSuccess(CorrelationData.Confirm result) { if(failMsgDao == null){ return; } if(!result.isAck()){ // 执行失败保存失败信息，如果已经存在保存信息，如果不在信息信息 failMsgDao.save(msgId, exchange, routingKey, msg, delay,DateUtils.getCurrentTime() + 10, "MQ回复nack"); }else if(isFailMsg && msgId != null){ // 如果发送的是失败消息，当收到ack需要从fail_msg删除该消息 failMsgDao.removeById(msgId); } } }
当发送消息失败会入库到失败消息表。
我们可以启动定时任务去扫描失败消息表的记录，重新发送，当达到最大失败次数后由人工处理。
1.2.3.4 测试
下边测试发送失败入库功能：
- 首先在publisher模块添加common-rabbitmq依赖
<dependency> <groupId>com.itheima</groupId> <artifactId>common-rabbitmq</artifactId> <version>0.0.1-SNAPSHOT</version> </dependency>
- 屏蔽MqConfig类中设置ReturnCallback的代码
由于RabbitTemplate只能设置一次ReturnCallback，而在common-rabbitmq中设置了ReturnCallback，所以屏蔽MqConfig类中设置ReturnCallback的代码，如下图：
- 在hmall中创建失败消息表
查看虚拟机对应hmall数据库，如果已存在fail_msg表需要删除原表再通过下边的语句新建表。
create table fail_msg ( id varchar(255) not null comment '消息id' primary key, exchange varchar(255) not null comment '交换机', routing_key varchar(255) not null comment '路由key', msg text not null comment '消息', reason varchar(255) not null comment '原因', delay_msg_execute_time int null comment '延迟消息执行时间', next_fetch_time int null comment '下次拉取时间', create_time datetime null comment '创建时间', update_time datetime null comment '更新时间', fail_count int default 0 null comment '失败次数' ) comment '失败消息记录表' charset = utf8mb4;
- 在publisher模块进行配置
在启动类中添加@MapperScan("com.itheima.common.rabbitmq.dao.mapper")
在application.yaml添加：直接复制粘贴为一级配置
rabbit-mq: enable: true persistence: enable: true
- 使用RabbitClient工具类发送消息
@Resource private RabbitClient rabbitClient; @Test void testPublisherReturn() { rabbitClient.sendMsg("hmall.directa", "q", "hello"); }
分别测试ReturnCallback和ConfirmCallback不同情况下失败消息入库。
测试ReturnCallback时注意：单元测试方法运行完就完毕了数据库连接池，而ReturnCallback是回调方法，是在单元测试方法执行完再执行，在ReturnCallback中操作数据库时报没有可用的数据库连接的错误，需要在单元测试方法最后添加休眠代码，保证ReturnCallback执行完成再结束整个单元测试方法。
@Test void testPublisherReturn() throws InterruptedException { rabbitClient.sendMsg("hmall.directa", "q", "hello"); // 增加一点时间，确保ReturnCallback执行完成 Thread.sleep(5000); }
1.2.4 失败消息定时任务重发(作业)
我们完成了失败消息的存储，请使用任意一个你熟悉的定时任务完成消息的重发
思路大致如下：
- 选定一个定时任务框架，如：SpringTask、Quartz、xxljob等
- 编写实现代码
- 扫描表：fail_msg
- 扫描条件：where fail_count < 3
- 调用rabbitClient重发消息
- 发送完成后，删除表：fail_msg中的数据
- 参考实现思路：首先问AI，然后就可以得到详尽的步骤
- 大致代码路径如下：这里我就不再写查找数据库那部分代码了，较为简单
1.2.5 小结
如何保证生产消息可靠性？
首先在发送消息时可以开启重试机制，避免因为短暂的网络问题导致发送消息失败。
RabbitMQ还提供生产者确认机制保证发送消息到MQ的可靠性。
生产者确认机制包括两种：
1.Publisher Return
消息投递成功但路由失败会调用Publisher Return回调方法返回异常信息。
2.Publisher Confirm
消息投递成功返回ack，投递失败返回nack。
注意：消息投递成功但可能路由失败了，此时会通过Publisher Confirm返回ack，通过Publisher Return回调方法返回异常信息。
我们在发送消息时给每个消息指定一个唯一ID，设置回调方法，如果Publisher Return失败或Publisher Confirm返回nack，我们在回调方法中解析失败消息，并记录到失败表由定时任务去异步重新发送。
1.3.消息持久化
为了提升性能，默认情况下MQ的数据都是在内存存储的临时数据，重启后就会消失。为了保证数据的可靠性，必须配置持久化，包括：
- 交换机持久化
- 队列持久化
- 消息持久化
我们以控制台界面为例来说明。
1.3.1 交换机持久化
交换机持久化是指将交换机的定义信息(元数据)持久化到RabbitMQ的数据库（mnesia）中，RabbitMQ重启后交换机定义仍然存在。
若交换机不设置持久化，在rabbitmq服务重启之后，相关的交换机元数据会丢失，但消息不会丢失，只是不能将消息发送到这个交换机中，所以通常要设置交换机持久化。
在控制台中声明交换机是默认设置持久化的。
设置方法：
在控制台的Exchanges页面，添加交换机时可以配置交换机的Durability参数：
设置为Durable就是持久化模式，Transient就是临时模式。
设置持久化后在交换机列表会有一个"D"标识
1.3.2 队列持久化
队列持久化也是将队列的定义信息(元数据)持久化到RabbitMQ的数据库中。
如果队列不设置持久化，在RabbitMQ重启后队列的元数据丢失。
设置队列持久化可以保证队列本身的元数据不会因异常情况而丢失，队列中存储的是消息的在队列中的位置、消息的ID、存储位置等，消息会存储在独立的rdq数据文件中，队列持久化不能保证消息数据不会丢失。
在控制台的Queues页面，添加队列时，同样可以配置队列的Durability参数：
除了持久化以外，你可以看到队列还有很多其它参数，有一些我们会在后期学习。
设置持久化后在队列列表会有一个"D"标识
1.3.3 消息持久化
要确保消息不会丢失需要将消息设置为持久化。
在控制台发送消息的时候，可以添加很多参数，而消息的持久化是要配置一个properties：
查看持久消息的内容：
delivery_mode=2 表示持久化
1.4 消费消息可靠性
当RabbitMQ向消费者投递消息以后，需要知道消费者的处理状态如何。因为消息投递给消费者并不代表就一定被正确消费了，可能出现的故障有很多，比如：
- 消息投递的过程中出现了网络故障
- 消费者接收到消息后突然宕机
- 消费者接收到消息后，因处理不当导致异常
- ...
一旦发生上述情况，消息也会丢失。因此，RabbitMQ必须知道消费者的处理状态，一旦消息处理失败能重新投递消息。
但问题来了：RabbitMQ如何得知消费者的处理状态呢？
本章我们就一起研究一下消费者处理消息时的可靠性解决方案。
1.4.1. 消费者确认机制
1.4.1.1 介绍
为了确认消费者是否成功处理消息，RabbitMQ提供了消费者确认机制（Consumer Acknowledgement）。即：当消费者处理消息结束后，应该向RabbitMQ发送一个回执，告知RabbitMQ消息处理状态。回执有三种可选值：
- ack：成功处理消息，RabbitMQ从队列中删除该消息
- nack：消息处理失败，RabbitMQ需要再次投递消息
- reject：消息处理失败并拒绝该消息，RabbitMQ从队列中删除该消息
一般reject方式用的较少，除非是消息格式有问题，那就是开发问题了。因此大多数情况下我们需要将消息处理的代码通过try catch机制捕获，消息处理成功时返回ack，处理失败时返回nack.
SpringAMQP帮我们实现了消息确认，并可以通过配置文件设置消息确认的处理方式，有三种模式：
none：不处理。即消息投递给消费者后消息会立刻从MQ删除。非常不安全，不建议使用manual：手动模式。需要自己在业务代码中调用api，发送ack或reject，存在业务入侵，但更灵活auto：自动模式。当业务正常执行时则自动返回ack. 当业务出现异常时，根据异常判断返回不同结果：
- 如果是业务异常，会自动返回
nack； - 如果是消息处理或校验异常，自动返回
reject，返回的异常包括：MessageConversionException、MethodArgumentTypeMismatchException等
1.4.1.2 auto模式测试
通过下面的配置可以修改消息确认的处理方式为auto：
spring: rabbitmq: listener: simple: acknowledge-mode: auto # 自动ack
修改consumer服务的SpringRabbitListener类中的方法，模拟一个消息处理的异常：
@RabbitListener(queues = "simple.queue") public void listenSimpleQueueMessage2(String msg) throws InterruptedException { log.info("spring 消费者接收到消息：【" + msg + "】"); if (true) { throw new RuntimeException("故意的"); } log.info("消息处理完成"); }
在此方法内第一句打断点，我们向队列“simple.queue”发一条消息，此时进入断点
可以发现此时有一条消息状态为unacked（未确定状态）：
放行以后，由于抛出的是业务异常，消息处理失败后，会回到RabbitMQ，并重新投递到消费者。
1.4.1.3 manual模式测试(自学)
通常在应用程序中会使用auto自动模式，手动模式在生产中不常用可以自行学习。
通过 api指定是否重新入队，具体返回ack或nack通过手动编程实现。
由于手动模式需要通过api编程，需要在监听方法添加Channel、Message类型的参数，如下：
Message：是spring AMQP封装的底层消息对象。
Channel：是消费端与MQ基于通道的操作对象。
@RabbitListener(queues = "simple.queue") public void listenSimpleQueueMessage(String msg, Channel channel, Message message) throws InterruptedException, IOException { log.info("spring 消费者接收到消息：【" + msg + "】"); //返回nack //每个参数的意义：1.消息的标记 2.是否确认之前所有未确认的消息 3.是否重新入队 channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true); // log.info("消息处理完成"); // //返回ack,每个参数的意义：1.消息的标记 2.是否确认之前所有消息 // channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); }
1.4.1.4 小结
消息者确认机制怎么实现？
消费者处理消息结束后，应该向RabbitMQ发送一个回执，告知RabbitMQ消息处理状态：
- ack：成功处理消息，RabbitMQ从队列中删除该消息
- nack：消息处理失败，RabbitMQ需要再次投递消息
- reject：消息处理失败并拒绝该消息，RabbitMQ从队列中删除该消息
具体实现方法：
SpringAMQP提供消息确认配置，有三种模式：
none：不处理。即消息投递给消费者后消息会立刻从MQ删除。非常不安全，不建议使用。manual：手动模式。需要自己在业务代码中调用api，发送ack或reject。auto：自动模式。当业务正常执行时则自动返回ack. 当业务出现异常时会自动返回nack.
