RocketMQ消息如何进行幂等处理

痛定思痛。 2024-03-24 16:44 55阅读 0赞

#### 1. 背景 ####

笔者在排查生产问题的时候，发现了一个RocketMQ消息重复消费的问题。

**生产者发送的消息：**

{
        "code":"4500000237",
        "ctrlOrgCode":"JT",
        "isInternalSupplier":false,
        "name":"AAAAAA",
        "supplierCategoryCode":"0105",
        "supplierFinanceList":[
            {
                "comOrgCode":"JT",
                "ctrlOrgCode":"JT"
            }
        ],
        "supplierPurchaseList":[
    
        ],
        "tenantId":"717070740494139392"
    }
    复制代码

上面是生产者发送的消息，但是消息生产者在极短的时间内发送了相同的消息。

> Tips: 发送的消息体内容一样，但是RocketMQ消息的MessageId的不同

**消费者一张表：**

CREATE TABLE `auth_allocation` (
      `id` bigint NOT NULL AUTO_INCREMENT COMMENT '主键ID',
      `biz_code` varchar(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL COMMENT '业务编码(用户ID,使用业务方编码)',
      `deleted` int NOT NULL COMMENT '0表示删除，1表示正常'
      `description` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT '' COMMENT '说明',
      `create_time` timestamp NOT NULL COMMENT '创建时间',
      `update_time` timestamp NOT NULL ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
      PRIMARY KEY (`id`)
    ) ENGINE=InnoDB AUTO_INCREMENT=4 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;
    复制代码

消费消息，首先会去 **`auth_allocation`** 表中的根据消息 **`code`** 查询 **`biz_code`** 是否存在。如果不存在就插入数据。

> Tips: auth\_allocation数据删除是软删除。通过deleted字段的值来控制。

所以在这里就会存在并发问题。如下图：

![format_png][]

生产环境代码处理逻辑：

获取到消费消息后，首先查询auth\_allocation中是否存在

SELECT * FROM auth_allocation WHERE deleted = 1 AND biz_code = 'xxxx'
    复制代码

然后处理其他的逻辑，再往表中插入数据。由于项目中的数据适用的软删除，所以不能设置biz\_code为唯一索引。在多个消费服务下就有可能出现本来biz\_code只能是一个值但是由于不能软删除不能设置唯一索引。所以可能出现上图所示的情况。

#### 2. 消息重复场景 ####

在互联网应用中，尤其在网络不稳定的情况下，消息队列RocketMQ版的消息有可能会出现重复。也可能是出现消息ID不同消息内容相同的情况。

消息重复的场景：

*  发送时消息重复
    
    当一条消息已被成功发送到服务端并完成持久化，此时出现了网络闪断或者客户端宕机，导致服务端对客户端应答失败。 如果此时生产者意识到消息发送失败并尝试再次发送消息，消费者后续会收到两条内容相同但Message ID不同的消息。或者发送消息者主动(bug导致)消息体一样Message ID不同的消息。
 *  投递时消息重复
    
    消息消费的场景下，消息已投递到消费者并完成业务处理，当客户端给服务端反馈应答的时候网络闪断。为了保证消息至少被消费一次，消息队列RocketMQ版的服务端将在网络恢复后再次尝试投递之前已被处理过的消息，消费者后续会收到两条内容相同并且Message ID也相同的消息。
 *  负载均衡时消息重复（包括但不限于网络抖动、Broker重启以及消费者应用重启）
    
    当消息队列RocketMQ版的Broker或客户端重启、扩容或缩容时，会触发Rebalance，此时消费者可能会收到少量重复消息。

#### 3. 处理方法 ####

处理方法可以从消息生产者和消息消费者入手，因为不同的消息Message ID可能对应相同的消息内容。可能出现重现重复情况，使用Message ID不建议。RocketMQ 提供了可以设置消息的Key, key可以由用户自定义，上面的案例code就是唯一值，那么code就能作为处理消息幂等的依据。

**消息生产者处理：**

Message message = new Message();
    message.setKey("4500000237");
    SendResult sendResult = producer.send(message);           
    复制代码

> Tips: Message也不用设置Key,直接通过消息体中的唯一值字段处理

消费者接收到消息时可以根据消息key(或者消息体中的唯一值)。

**消费者处理：**

consumer.subscribe("ons_test", "*", new MessageListener() {
        public Action consume(Message message, ConsumeContext context) {
            String key = message.getKey()
            // 根据业务唯一标识的Key做幂等处理。
        }
    }); 
    //或者
    consumer.subscribe("ons_test", "*", new MessageListener() {
        public Action consume(Message message, ConsumeContext context) {
             String body = message.getBody();
            // 解析出body的唯一值做幂等处理
        }
    });    
    复制代码

接下来就是消费者如何根据唯一值做幂等处理。

3.1 消费者消息幂等处理

消息的幂等处理需要看消费者服务的部署情况，这里需要区分是单机部署还是集群两种情况。

**单服务部署处理方式：**

*  数据库对唯一值的入库字段设唯一索引，如果存在相同的唯一值存在插入数据就会报错。只需要处理相对应的错误即可。
 *  通过锁处理，对插入数据的步骤加锁(本地锁或者数据库锁)
    
        SELECT * FROM auth_allocation WHERE deleted = 1 AND biz_code = 'xxxx' for update
        复制代码

**集群消费服务部署：**

*  使用数据库的行锁处理
 *  利用分布式锁处理不同服务间的并发。
 *  数据库对唯一值的入库字段设唯一索引。

对应上述案例，如果不能设置数据库唯一索引，只能通过分布式锁或者数据库的行锁来处理消息的幂等。

#### 4. 总结 ####

*  消息消费失败做好回滚处理。
 *  一些无法做到幂等的操作，需要发送警告给相关人员进行手动处理。

[format_png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/24/e1afce6e1cfa42cbbcd71e035f26d571.png