这篇文章，我们聊聊消息队列中非常重要的最佳实践之一：消费幂等。
1 基础概念
当出现 RocketMQ 消费者对某条消息重复消费的情况时，重复消费的结果与消费一次的结果是相同的，并且多次消费并未对业务系统产生任何负面影响，那么这个消费者的处理过程就是幂等的。
例如，在支付场景下，消费者消费扣款消息，对一笔订单执行扣款操作，扣款金额为100元。
如果因网络不稳定等原因导致扣款消息重复投递，消费者重复消费了该扣款消息，但最终的业务结果是只扣款一次，扣费100元，且用户的扣款记录中对应的订单只有一条扣款流水，不会多次扣除费用。那么这次扣款操作是符合要求的，整个消费过程实现了消费幂等。
2 适用场景
RocketMQ 消息重复的场景如下：
发送时消息重复
当一条消息已被成功发送到服务端并完成持久化，此时出现了网络闪断或者客户端宕机，导致服务端对客户端应答失败。
如果此时生产者意识到消息发送失败并尝试再次发送消息，消费者后续会收到两条内容相同但 Message ID 不同的消息。
投递时消息重复
消息消费的场景下，消息已投递到消费者并完成业务处理，当客户端给服务端反馈应答的时候网络闪断。为了保证消息至少被消费一次，Broker 服务端将在网络恢复后再次尝试投递之前已被处理过的消息，消费者后续会收到两条内容相同并且 Message ID 也相同的消息。
负载均衡时消息重复（包括但不限于网络抖动、Broker 重启以及消费者应用重启）
Broker 端或客户端重启、扩容或缩容时，会触发 Rebalance ，此时消费者可能会收到少量重复消息。
3 处理策略
因为不同的 Message ID 对应的消息内容可能相同，有可能出现冲突（重复）的情况，所以真正安全的幂等处理，不建议以 Message ID 作为处理依据。
最好的方式是以业务唯一标识作为幂等处理的关键依据，业务的唯一标识可以通过消息 Key 设置。
Message msg = new Message(TOPIC /* Topic */,
TAG /* Tag */,
("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
);
message.setKey("ORDERID_100"); // 订单编号
SendResult sendResult = producer.send(message);
消费者收到消息时可以根据消息的 Key ，即订单号来实现消息幂等 ：
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
for (MessageExt message : msgs) {
String key = message.getKeys();
// 根据业务唯一标识的Key做幂等处理
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
注意：我们也可以在消息 body 体设置业务唯一标识，使用 key 是为了编程简易。
常见的幂等策略包含：
- 数据库去重表
- Redis 添加标志位
- 业务状态机判断
4 数据库去重表
举一个电商场景的例子：用户购物车结算时，系统会创建支付订单。用户支付成功后支付订单的状态会由未支付修改为支付成功，然后系统给用户增加积分。
我们可以使用 RocketMQ 事务消息的方案，该方案能够发挥 MQ 的优势：异步和解耦，以及事务的最终一致性的特性。
在消费监听器逻辑里，幂等非常重要 。积分表 SQL 如下：
CREATE TABLE `t_points` (
`id` bigint(20) NOT NULL COMMENT '主键',
`user_id` bigint(20) NOT NULL COMMENT '用户id',
`order_id` bigint(20) NOT NULL COMMENT '订单编号',
`points` int(4) NOT NULL COMMENT '积分',
`remarks` varchar(128) COLLATE utf8mb4_bin NOT NULL COMMENT '备注',
`create_time` datetime NOT NULL,
PRIMARY KEY (`id`),
UNIQUE KEY `unique_order_Id` (`order_id`) USING BTREE COMMENT '订单唯一'
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;
当收到订单信息后，首先判断该订单是否有积分记录，若没有记录，才插入积分记录。
就算出现极端并发场景下，订单编号也是唯一键，数据库中也必然不会存在相同订单的多条积分记录。
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
try {
for (MessageExt messageExt : msgs) {
String orderJSON = new String(messageExt.getBody(), "UTF-8");
logger.info("orderJSON:" + orderJSON);
OrderPO orderPO = JSON.parseObject(orderJSON, OrderPO.class);
// 首先查询是否处理完成
PointsPO pointsPO = pointsMapper.getByOrderId(orderPO.getId());
if (pointsPO == null) {
Long id = SnowFlakeIdGenerator.getUniqueId(1023, 0);
pointsPO = new PointsPO();
pointsPO.setId(id);
pointsPO.setOrderId(orderPO.getId());
pointsPO.setUserId(orderPO.getUserId());
// 添加积分数 30
pointsPO.setPoints(30);
pointsPO.setCreateTime(new Date());
pointsPO.setRemarks("添加积分数 30");
pointsMapper.insert(pointsPO);
}
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
} catch (Exception e) {
logger.error("consumeMessage error: ", e);
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
}
5 Redis 添加标志位
在消费者接收到消息后，首先判断 Redis 中是否存在该业务主键的标志位，若存在标志位，则认为消费成功，否则，则执行业务逻辑，执行完成后，在缓存中添加标志位。
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
try {
for (MessageExt messageExt : msgs) {
String bizKey = messageExt.getKeys(); // 唯一业务主键
//1. 判断是否存在标志
if(redisTemplate.hasKey(RedisKeyConstants.WAITING_SEND_LOCK + bizKey)) {
continue;
}
//2. 执行业务逻辑
//TODO do business
//3. 设置标志位
redisTemplate.opsForValue().set(RedisKeyConstants.WAITING_SEND_LOCK + bizKey, "1", 72, TimeUnit.HOURS);
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
} catch (Exception e) {
logger.error("consumeMessage error: ", e);
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
}
6 业务状态机判断
笔者曾经服务于神州专车，乘客在用户端点击立即叫车，订单服务创建订单，首先保存到数据库后，然后将订单信息同步保存到缓存中。
在订单的载客生命周期里，订单的修改操作先修改缓存，然后发送消息到 MetaQ ，订单落盘服务消费消息，并判断订单信息是否正常（比如有无乱序)，若订单数据无误，则存储到数据库中。
这种设计会有小的概率，因为网络问题，消费者会收到乱序的消息。
订单状态机按顺序分别是：创建、已分配司机、司机已出发、司机已到达、司机已接到乘客、已到达。
当订单状态是司机已到达时，消费者收到司机已出发的消息，此时只要业务数据正确，就可以忽略旧的消息，打印相关日志即可。
