【RocketMQ工作原理】

悠悠 2022-08-28 15:54 21阅读 0赞

## 什么是消费幂等 ##

当出现消费者对某条消息重复消费的情况时，重复消费的结果与消费一次的结果是相同的，并且多次消  
费并未对业务系统产生任何负面影响，那么这个消费过程就是消费幂等的。

> 幂等：若某操作执行多次与执行一次对系统产生的影响是相同的，则称该操作是幂等的。

在互联网应用中，尤其在网络不稳定的情况下，消息很有可能会出现重复发送或重复消费。如果重复的  
消息可能会影响业务处理，那么就应该对消息做幂等处理。

## 消息重复的场景分析 ##

什么情况下可能会出现消息被重复消费呢？最常见的有以下三种情况：

### 发送时消息重复 ###

当一条消息已被成功发送到Broker并完成持久化，此时出现了网络闪断，从而导致Broker对Producer应  
答失败。 如果此时Producer意识到消息发送失败并尝试再次发送消息，此时Broker中就可能会出现两  
条内容相同并且Message ID也相同的消息，那么后续Consumer就一定会消费两次该消息。

### 消费时消息重复 ###

消息已投递到Consumer并完成业务处理，当Consumer给Broker反馈应答时网络闪断，Broker没有接收  
到消费成功响应。为了保证消息至少被**消费一次**的原则，Broker将在网络恢复后再次尝试投递之前已  
被处理过的消息。此时消费者就会收到与之前处理过的内容相同、Message ID也相同的消息。

### Rebalance时消息重复 ###

当Consumer Group中的Consumer数量发生变化时，或其订阅的Topic的Queue数量发生变化时，会触  
发Rebalance，此时Consumer可能会收到曾经被消费过的消息。

## 通用解决方案 ##

### 两要素 ###

幂等解决方案的设计中涉及到两项要素：幂等令牌，与唯一性处理。只要充分利用好这两要素，就可以  
设计出好的幂等解决方案。

*  幂等令牌：是生产者和消费者两者中的既定协议，通常指具备唯⼀业务标识的字符串。例如，订  
    单号、流水号。一般由Producer随着消息一同发送来的。
 *  唯一性处理：服务端通过采用⼀定的算法策略，保证同⼀个业务逻辑不会被重复执行成功多次。  
    例如，对同一笔订单的多次支付操作，只会成功一次。

## 解决方案 ##

对于常见的系统，幂等性操作的通用性解决方案是：

1.  首先通过缓存去重。在缓存中如果已经存在了某幂等令牌，则说明本次操作是重复性操作；若缓  
    存没有命中，则进入下一步。
2.  在唯一性处理之前，先在数据库中查询幂等令牌作为索引的数据是否存在。若存在，则说明本次  
    操作为重复性操作；若不存在，则进入下一步。
3.  在同一事务中完成三项操作：唯一性处理后，将幂等令牌写入到缓存，并将幂等令牌作为唯一索  
    引的数据写入到DB中。

> 第1步已经判断过是否是重复性操作了，为什么第2步还要再次判断？能够进入第2步，说明已经  
> 不是重复操作了，第2次判断是否重复？  
> 当然不重复。一般缓存中的数据是具有有效期的。缓存中数据的有效期一旦过期，就是发生缓  
> 存穿透，使请求直接就到达了DBMS。

### 解决方案举例 ###

1.  当支付请求到达后，首先在Redis缓存中却获取key为**支付流水号**的缓存value。若value不空，则  
    说明本次支付是重复操作，业务系统直接返回调用侧重复支付标识；若value为空，则进入下一步  
    操作
2.  到DBMS中根据支付流水号查询是否存在相应实例。若存在，则说明本次支付是重复操作，业务  
    系统直接返回调用侧重复支付标识；若不存在，则说明本次操作是首次操作，进入下一步完成唯  
    一性处理
3.  在分布式事务中完成三项操作：

*  完成支付任务
 *  将当前支付流水号作为key，任意字符串作为value，通过set(key, value, expireTime)将数  
    据写入到Redis缓存
 *  将当前支付流水号作为主键，与其它相关数据共同写入到DBMS

### 4 消费幂等的实现 ###

消费幂等的解决方案很简单：为消息指定不会重复的唯一标识。因为Message ID有可能出现重复的情  
况，所以真正安全的幂等处理，不建议以Message ID作为处理依据。最好的方式是以业务唯一标识作为  
幂等处理的关键依据，而业务的唯一标识可以通过消息Key设置。  
以支付场景为例，可以将消息的Key设置为订单号，作为幂等处理的依据。具体代码示例如下：

Message message = new Message();
    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 msg:msgs){
            String key = msg.getKeys();
                // 根据业务唯一标识Key做幂等处理
                // ……
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    });

> RocketMQ能够保证消息不丢失，但不能保证消息不重复