RabbitMQ 消息确认机制、补偿机制、消息幂等性实践

喜欢ヅ旅行 2023-10-01 19:46 62阅读 0赞

### 1. 场景 ###

先看这么几个面试题：

1.  如何保证消息的可靠性投递？即如何确定消息是否发送成功？
2.  如果失败如何处理（补偿机制）？
3.  如何保证消息不被重复消费？或者说，如何保证消息消费时的幂等性？

### 2. [消息的可靠性投递][Link 1] ###

**消息确认**

消息确认包括主要**生产者发送确认**和**消费者接收确认**，因为发送消息的过程中我们是无法确认消息是否能路由等，一旦消息丢失我们就无法处理，所以需要确认消息，避免消息丢失。

#### 2.1 生产者确认 ####

我们知道生产者与消费者完全隔离的，不做任何配置的情况下，生产者是不知道消息是否真正到达 RabbitMQ，也就是说消息发布操作不返回任何消息给生产者。

那么怎么保证我们消息发布的**可靠性投递**？有以下几种常用机制。

![image-20220223172851176][]

由于之前的[文章][Link 1]对上面都有过介绍，所以这里不一一介绍，而一般采用的方式就是**发布者确认模式（生产者确认模式）**。

**原理**：生产者将信道设置成 confirm 模式，一旦信道进入 confirm 模式，所有在该信道上面发布的消息都将会被指派一个唯一的 ID（从 1 开始），由这个 id 在生产者和 RabbitMQ 之间进行消息的确认。

这里的**唯一 ID** 能够唯一标识消息，在消息不可达的时候触发回调时可以获取该值，进行对应的错误处理，即对应的消息补偿机制。（记住这个唯一 ID，且是全局唯一，分布式系统中可采用雪花算法等方式）

confirm 模式最大的好处在于他可以是异步的，一旦发布一条消息，生产者应用程序就可以在等信道返回确认的同时继续发送下一条消息，当消息最终得到确认之后，生产者应用便可以通过回调方法来处理该确认消息，如果 RabbitMQ 因为自身内部错误导致消息丢失，就会发送一条 nack 消息，生产者应用程序同样可以在回调方法中处理该 nack 消息决定下一步的处理。

注：这里描述的场景都是能**正确路由到队列中**的，也就是不考虑失败通知（ReturnCallback）的情况。

由于现在大家开发基本都是通过 Spring Boot 的方式进行开发，所以，这里直接提供其基本配置类参考，如下：

/**
     * @description : 消息生产者
     */
    @Component
    @Slf4j
    public class RabbitmqProducer {
        
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        public void sendMessage(Map<String, Object> headers, Object message, String messageId, String exchangeName, String key) {
        
            // 自定义消息头
            MessageHeaders messageHeaders = new MessageHeaders(headers);
            // 创建消息
            Message<Object> msg = MessageBuilder.createMessage(message, messageHeaders);
            /* 确认的回调 确认消息是否到达 Broker 服务器 其实就是是否到达交换器
             * 如果发送时候指定的交换器不存在 ack 就是 false 代表消息不可达
             */
            rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
        
                log.info("correlationData：{} , ack:{}", correlationData.getId(), ack);
                if (!ack) {
        
                    System.out.println("进行对应的消息补偿机制");
                }
            });
            // 在实际中ID 应该是全局唯一 能够唯一标识消息 消息不可达的时候触发ConfirmCallback回调方法时可以获取该值，进行对应的错误处理
            CorrelationData correlationData = new CorrelationData(messageId);
            rabbitTemplate.convertAndSend(exchangeName, key, msg, correlationData);
        }
    }

#### 2.2 消费者确认 ####

说了生产者如何保证消息的确认，而对于消费者来说，同样需要确认。

前面也说了目前编码都是基于 Spring 的，那么对于消费者来说，同样也有一个接收消息的配置类，如下：

/**
     * @description : 消息消费者
     */
    @Component
    @Slf4j
    public class RabbitmqConsumer {
        
    
        @RabbitListener(bindings = @QueueBinding(
                value = @Queue(value = RabbitInfo.QUEUE_NAME, durable = RabbitInfo.QUEUE_DURABLE),
                exchange = @Exchange(value = RabbitInfo.EXCHANGE_NAME, type = RabbitInfo.EXCHANGE_TYPE),
                key = RabbitInfo.ROUTING_KEY)
        )
        @RabbitHandler
        public void onMessage(Message message, Channel channel) throws Exception {
        
            MessageHeaders headers = message.getHeaders();
            // 获取消息头信息和消息体
            log.info("msgInfo:{} ; payload:{} ", headers.get("msgInfo"), message.getPayload());
        }
    }

对于上面接收消息的配置并没有做任何配置，当我们发送消息的时候，消费者接收消息并进行对应的逻辑处理，并且 **Spring 的处理是自动 ack 的**，但其实它也有配置，如下：

spring:
      rabbitmq:
        addresses: 127.0.0.1:5672
        # RabbitMQ 默认的用户名和密码都是guest 而虚拟主机名称是 "/"
        # 如果配置其他虚拟主机地址，需要预先用管控台或者图形界面创建 图形界面地址 http://主机地址:15672
        username: admin
        password: admin
        virtual-host: /
        listener:
          simple:
            # 为了保证信息能够被正确消费,建议签收模式设置为手工签收,并在代码中实现手工签收
            acknowledge-mode: manual
            # 侦听器调用者线程的最小数量
            concurrency: 10
            # 侦听器调用者线程的最大数量
            max-concurrency: 50

也就是上面的`acknowledge-mode`，他有三个值，如下：

![image-20220223194810055][]

1.  当为 `NONE` 的时候，即默认值，即`autoAck=true`，消费者接收消息后自动确认，此时，MQ 队列中的消息会移除；
2.  当为 `MANUAL` 的时候，需要我们手动确认，即`channel.basicAck`，如下：
    
    ![image-20220223195153898][]
3.  当为 `AUTO` 的时候，经测试发现和`NONE`没什么不同 。

此时，有人就会说那么如果在默认情况下（自动确认），我们的业务代码抛出异常了怎么办？

Spring 的做法是会抛出异常，并且消息不会被 ack，如下面这种情况：

![image-20220223195953222][]

`1/0` 肯定会抛出异常，此时会一直打印日志，如下：

![image-20220223200128118][]

再去看 MQ 客户端，状态如下：

![image-20220223200207390][]

此时，可能会造成重复消费，怎么理解？

假如我的业务代码没有事务，或者在参数的传递过程中某个方法没有事务的控制，当异常业务代码之前入库了，那么这条消息实际上是没有被确认的，还在队列中，因此，当下次程序启动，则会再次消费这条消息，尽管业务代码出现了异常。

自动确认会在消息发送给消费者后立即确认，但存在丢失消息的可能，如果消费端业务代码抛出异常，也就是消费端没有处理成功这条消息，那么就**相当于丢失了消息**。

如果消息已经被处理，但后续代码抛出异常，使用 Spring 进行管理的话消费端业务代码会进行回滚，这也同样造成了实际意义的消息丢失。

### 3. 补偿机制 ###

在回到前面的问题，如何确定消息是否发送成功？**生产者确认机制**确实能帮我们解决这个问题，但如果生产者就是接收不到 ack 这个指令怎么办，比如消费者处理时间太长或者网络超时，等等情况，导致生产者一直接收不到这个 ack ，此时怎么办？

生产者与消费者之间应该约定一个超时时间，比如 5 分钟，对于超出这个时间没有得到响应的消息，可以设置一个定时重发的补偿机制：通过消息落库 + 定时任务来实现。

怎么做？这里讲讲思路，如下：

1.  发送消息之前，先把消息入库，我这里的表设计如下：
    
        CREATE TABLE `t_cap_published_message` (
          `id` varchar(40) COLLATE utf8mb4_bin NOT NULL DEFAULT '' COMMENT '标识。',
          `version` varchar(20) COLLATE utf8mb4_bin NOT NULL DEFAULT '' COMMENT '版本',
          `exchange` varchar(200) COLLATE utf8mb4_bin DEFAULT '' COMMENT '交换机。',
          `topic` varchar(200) COLLATE utf8mb4_bin NOT NULL DEFAULT '' COMMENT '话题。',
          `content` longtext COLLATE utf8mb4_bin NOT NULL COMMENT '消息内容。',
          `retries` int(11) NOT NULL COMMENT '重试次数，一般为 3 次。',
          `expiry` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '过期时间。',
          `status` varchar(40) COLLATE utf8mb4_bin NOT NULL COMMENT '状态，成功则消息ack成功，其他状态都要重试。',
          `created_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间。',
          `last_modified_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '最后更新时间，可以用作数据版本。',
          PRIMARY KEY (`id`)
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT='发布的消息。';
2.  入库之后在发送消息；
3.  如果在规定时间不能 `ack` 或者 `ack=false`，即`confirmCallback`回调的`ack=false`，则按照定时规则重新发送消息；
4.  然后对于发布成功的消息，如果业务操作完成，实际上它的作用已经发挥完成，一段时间对数据库做清理即可，根据业务的具体情况。

### 4. 消息幂等性 ###

首先我们要知道什么是幂等性，比如一个转账系统，A 要转给 B 100元，当 A 发出消息后，B 接收成功，然后给 MQ 确认的时候出现网络波动，MQ 并没有接收到 ack 确认，那 MQ 为了保证消息被消费，就会继续给消费者投递之前的消息，如果再重复投递 5 次，则 B 在处理 5 次，加上之前的一次，B 的余额增加了 600 元，很明显是不合理的。

所以幂等性简单来说就是：**重复调用多次产生的业务结果与调用一次产生的业务结果相同**；

为了避免相同消息的重复处理，必须要采取一定的措施。RabbitMQ 服务端是没有这种控制的，因为它不知道你是不是就要把一条消息发送两次，所以只能在消费端控制。

回到前面生产者确认模式中讲到了一个**全局唯一 ID**，我们可以通过他来保证消息的幂等性，如下：

1.  消费者获取到消息后先根据这个全局**唯一 ID**去查询 redis/db 是否存在该消息；
2.  如果不存在，则正常消费，消费完毕后写入 redis/db；
3.  如果存在，则证明消息被消费过，直接丢弃，不做处理。

[Link 1]: https://www.javatv.net/pages/cbc8b9/
[image-20220223172851176]: https://img-blog.csdnimg.cn/img_convert/2054b5e5b2e807d52d3d3a29ddb423b1.png
[image-20220223194810055]: https://img-blog.csdnimg.cn/img_convert/f7e86f92f240b7bf523df5f3d1ab836e.png
[image-20220223195153898]: https://img-blog.csdnimg.cn/img_convert/e95e806e6511df63ebaed6021c72eab7.png
[image-20220223195953222]: https://img-blog.csdnimg.cn/img_convert/22c18f546cdc1b0849706116e950e986.png
[image-20220223200128118]: https://img-blog.csdnimg.cn/img_convert/965d2a4b9523e9885abdd7e858843996.png
[image-20220223200207390]: https://img-blog.csdnimg.cn/img_convert/bd2fb17dfe5bfbc4c62fe6bf954af6a1.png