Java中的RocketMQ消费幂等详解

目录

• 什么是消息幂等
• 消费幂等的必要性
• 通用解决方案
• 设置业务key
• 支付实例场景

什么是消息幂python等

当出现消费者对某条消息重复消费的情况时，重复消费的结果与消费一次的结果是相同的，并且多次消费并未对业务系统业务产生任何负面影响，那么这个消费过程就是消息幂等，在互联网应用中尤其在网络不稳定的情况下，消息很有可能会出现重复发送或者重复消费，如果重复的消息可能会影响业务处理，那么就应该应用消息幂等处理

消费幂等的必要性

在互联网应用中，尤其在网络不稳定的情况下，消息队列 RocketMQ 的消息有可能会出现重复，这个重复简单可以概括为以下情况：

• 发送时消息重复

当一条消息已被成功发送到服务端并完成持久化，此时出现了网络闪断或者客户端宕机，导致服务端对客户端应答失败。 如果此时生产者意识到消息发送失败并尝试再次发送消息，消费者后续会收到两条内容相同并且 Message ID 也相同的消息。

• 投编程递时消息重复

消息消费的场景下，消息已投递到消费编程客栈者并完成业务处理，当客户端给服务端反馈应答的时候网络闪断。 为了保证消息至少被消费一次，消息队列 RocketMQ 的服务端将在网络恢复后再次尝试投递之前已被处理过的消息，消费者后续会收到两条内容相同并且 Message ID 也相同的消息。

• Rebalance时消息重复（包括但不限于网络抖动、Broker 重启以及订阅方应用重启）

当消息队列 RocketMQ 的 Broker 或客户端重启、扩容或缩容时，会触发 Rebalance，此时消费者可能会收到重复消息。

通用解决方案

两要素

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

• 幂等令牌：是生产者和消费者两者中的既定令牌，通常具有唯一业务标识的字符串
• 唯一性处理：服务端通过采用一定的算法策略，保证同一个业务逻辑下不会被重复执行成功多次

解决方案

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

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

设置业务key

因为 Message ID 有可能出现冲突（重复）的情况，所以真正安全的幂等处理，不建议以 Message ID 作为处理依据。 最好的方式是以业务唯一标识作为幂等处理的关键依据，而业务的唯一标识可以通过消息 Key 进行设置：

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

订阅方收到消息时可以根据消息的 Key 进行幂等处理：

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

注意：

消息重复消费的情况是不能去避免的，我们需要考虑的就是在有重复消息的情况之下，怎么取保证幂等性，那么在保证幂等性的有一个关键，就是在发送消息的时候携带一个业务Key，然后在接收到消息后先去获得这编程客栈个业务Key，然后在消费方的数据库当中判断一下这个业务Key所对应的消息有没有消费过，如果没有消费过就接着消费，消费完了在数据中存储一下，或者缓存数据库中存储也行，如果当前业务Key对应的消息已经消费过，那么直接舍弃即可！

支付实例场景

• 当支付请求到达后，首先在Redis中获取key作为支付流水号的缓存value，若value不为空，则说明本次支付是重复操作，业务系统直接返回调用侧重复支付标识，若value为空，则进入下一步操作
• DBMS中根据支付流水号查询是否存在相应的实例，若存在，则说明本次支付是重复操作，业务系统直接返回调用侧支付标识，若不存在，则说明本次操作是首次操作，进入下一步完成唯一性处理
• 在分布式系统中完成三项操作
  • 完成支付任务
  • 将当前支付流水号作为key，任意字段作为value，写入到Redis缓存中
  • 将当前支付流水号作为主键，与其他相关数据共同写入到DBMS中

到此这篇关于Java中的RocketMQ消费幂等详解的文章就介绍到这了,更多相关RocketMQ消费幂等内容请搜索编程客栈(www.devze.com)以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程客栈(www.devze.com)！