消费幂等

Posted 2021-04-24 程序架道

什么是幂等操作

我做一次调用请求A，无论调用多少次，影响的结果都是相同的。这就是幂等操作。很显然读操作是天然的幂等操作，如果是更新操作包括增加、修改、删除，在进行这些操作的时候一定要注意幂等性。尤其是涉及支付等计算活动的时候更要重视不是幂等操作带来的严重后果。

是什么引起我们要关注消费幂等

在大多数消息中间件产品都要确保消息一定投递成功，但同时又要确保不重复投递就非常困难。在这种情况下消息中间件为了保证消息不丢失宁可让消息重复。在分布式系统环境下最大的不稳定因素之一就是网络因素，从网络因素造成消息重复的原因有如下两种情况。

发送消息时重复

消息生产者将消息MSG1发送到MQ服务端，服务端在返回生产者响应的时候网络闪断，那么生产者认为消息MSG1没有发送成功，则会尝试第二次发送。造成服务端存储了两份MSG1消息。如下图：

消费消息时重复

消息消费端第一次在消费服务端上面的MSG1消息的时候，出现网络闪断。为了确保消息一定投递，在网络恢复的时候MQ服务端会尝试第二次投递，则消费端会收到两份MSG1消息。如下图：

上面是两种被动的情况，还有一种情况是消息业务逻辑处理系统A在处理消息过程中需要调用系统B。那么在这种情况下有可能出现系统A给BROKER返回失败，实际上系统B已经处理成功，根据消息的重试机制，系统A会再次调用系统B。这里面也会有幂等的问题，联机幂等。这个时候就需要系统A和系统B两者来协商保证因为消费消息引起的幂等问题。

如何处理消费幂等

根据RocketMQ的官方文档，它给出了处理消费幂等的建议，不能考虑使用Message ID，因为Message ID有可能是会重复。最好的方式是跟进业务ID来判断。如果是使用RocketMQ可以参照如下方式：

Message message = new Message();
message.setKey("ORDERID_1001");
SendResult sendResult = producer.send(message);

消费消息的时候可以跟进消息key来进行幂等处理，如下代码：

consumer.subscribe("msg_test", "*", new MessageListener() {
    public Action consume(Message message, ConsumeContext context) {
        String key = message.getKey()
        // 根据业务标识的 key 做幂等处理,key是唯一的
    }
});

小节

由于网络这一最大的不稳定因素，我们在使用消息中间的时候，经常会面对消息重复的情况，这种情况就需要关注幂等性操作。如果是读性质的操作我们不用关心，因为它具备天然的幂等性。我们以RocketMQ为例介绍了处理消费幂等的操作，利用业务上的ID 来处理而不是用中间件自带的Message ID。这样是最安全的处理方式。