乐趣区

关于rocketmq:大写的服看完这篇你还不懂RocketMQ算我输

jiezi

目录

  1. RocketMQ 介绍
  2. RocketMQ 概念

  3. 为什么要用 RocketMQ？

    1. 异步解耦
    2. 削峰填谷
    3. 分布式事务最终一致性
    4. 数据散发
  4. RocketMQ 架构

  5. RocketMQ 音讯类型

    1. 一般音讯
    2. 程序音讯
    3. 定时音讯
    4. 事务音讯

  6. 最佳实际

    1. 音讯重试
    2. 音讯过滤
    3. 生产模式
    4. 生产幂等
  7. 本地事务音讯封装
  8. 参考代码

RocketMQ 介绍

Apache RocketMQ 是一款 低提早、高并发、高可用、高牢靠的分布式消息中间件。音讯队列 RocketMQ 可为分布式应用零碎提供异步解耦和削峰填谷的能力，同时也具备互联网利用所需的海量音讯沉积、高吞吐、牢靠重试等个性。

RocketMQ 概念

  • Topic：音讯主题，用于将一类的音讯进行归类，比方订单主题，就是所有订单相干的音讯都能够由这个主题去承载，生产者向这个主题发送音讯。
  • 生产者：负责生产音讯并发送音讯到 Topic 的角色。
  • 消费者：负责从 Topic 接管并生产音讯 的角色。
  • 音讯：生产者向 Topic 发送的内容，会被消费者生产。
  • 音讯属性：生产者发送的时候能够为音讯自定义一些业务相干的属性，比方 Message Key 和 Tag 等。
  • Group：一类生产者或消费者，这类生产者或消费者通常生产或生产同一类音讯，且音讯公布或订阅的逻辑统一。

为什么要应用 RocketMQ？

异步解耦

随着微服务架构的风行，服务之间的关系梳理十分重要。异步解耦能够升高服务之间的耦合水平，同时也能进步服务的吞吐量。

应用异步解耦的业务场景十分多，因为每个行业的业务都会不太一样，以一些比拟通用的业务来阐明置信大家都能了解。

比方电商行业的下单业务场景，以最简略的下单流程来说，下单流程如下：

  1. 锁库存
  2. 创立订单
  3. 用户领取
  4. 扣减库存
  5. 给用户发送购买短信告诉
  6. 给用户减少积分
  7. 告诉商家发货

咱们以下单胜利后，用户进行领取，领取实现会有个逻辑叫领取回调，在回调外面须要去做一些业务逻辑。首先来看下同步解决须要破费的工夫，如下图：

下面的下单流程从 3 到 5 都是能够采纳异步流程进行解决，对于用户来说，领取实现后他就不须要关注前面的流程了。后盾缓缓解决就行了，这样就能简化三个步骤，进步回调的解决工夫。

削峰填谷

削峰填谷指的是在大流量的冲击下，利用 RocketMQ 能够抗住刹时的大流量，爱护零碎的稳定性，晋升用户体验。

在电商行业，最常见的流量冲击就是秒杀流动了，利用 RocketMQ 来实现一个残缺的秒杀业务还是与很多须要做的工作，不在本文的范畴内，前面有机会能够独自跟大家聊聊。想通知大家的是像诸如此类的场景能够利用 RocketMQ 来扛住高并发，前提是业务场景反对异步解决。

分布式事务最终一致性

家喻户晓，分布式事务有 2PC，TCC，最终一致性等计划。其中应用音讯队列来做最终一致性计划是比拟罕用的。

在电商的业务场景中，交易相干的外围业务肯定要确保数据的一致性。通过引入音讯队列 RocketMQ 版的分布式事务，既能够实现零碎之间的解耦，又能够保障最终的数据一致性。

数据散发

数据散发指的是能够将原始数据散发到多个须要应用这份数据的零碎中，实现数据异构的需要。最常见的有将数据散发到 ES, Redis 中为业务提供搜寻，缓存等服务。

除了手动通过音讯机制进行数据散发，还能够订阅 Mysql 的 binlog 来散发，在散发这个场景，须要应用 RocketMQ 的程序音讯来保证数据的一致性。

RocketMQ 架构

图片起源阿里云官网文档

  • Name Server：是一个简直无状态节点，可集群部署，在音讯队列 RocketMQ 版中提供命名服务，更新和发现 Broker 服务。就是一个注册核心。
  • Broker：音讯直达角色，负责存储音讯，转发音讯。分为 Master Broker 和 Slave Broker，一个 Master Broker 能够对应多个 Slave Broker，然而一个 Slave Broker 只能对应一个 Master Broker。Broker 启动后须要实现一次将本人注册至 Name Server 的操作；随后每隔 30s 定期向 Name Server 上报 Topic 路由信息。
  • 生产者：与 Name Server 集群中的其中一个节点（随机）建设长链接（Keep-alive），定期从 Name Server 读取 Topic 路由信息，并向提供 Topic 服务的 Master Broker 建设长链接，且定时向 Master Broker 发送心跳。
  • 消费者：与 Name Server 集群中的其中一个节点（随机）建设长连贯，定期从 Name Server 拉取 Topic 路由信息，并向提供 Topic 服务的 Master Broker、Slave Broker 建设长连贯，且定时向 Master Broker、Slave Broker 发送心跳。Consumer 既能够从 Master Broker 订阅音讯，也能够从 Slave Broker 订阅音讯，订阅规定由 Broker 配置决定。

RocketMQ 音讯类型

RocketMQ 反对丰盛的音讯类型，能够满足多场景的业务需要。不同的音讯有不同的利用场景，上面为大家介绍罕用的四种音讯类型。

一般音讯

一般音讯是指 RocketMQ 中无个性的音讯。当没有非凡的业务场景，应用一般音讯就够了。如果有非凡的场景，就能够应用非凡的音讯类型，比方程序，事务等。

同步发送

同步发送：音讯发送方发送进来一条音讯，会同步失去服务端返回的后果。

异步发送

异步发送：音讯发送方收回去一条音讯，不必期待服务端返回后果，能够接着发送下一条音讯。发送方能够通过回调接口接管服务端响应，并解决响应后果。

单向发送

单向发送：音讯发送方只负责发送音讯，发送进来后就不论了，这种形式发送速度十分快，存在失落音讯的危险。

程序音讯

程序音讯是指生产者依照肯定的先后顺序公布音讯；消费者依照既定的先后顺序订阅音讯，即先公布的音讯肯定会先被消费者接管到。

比方数据散发的场景，如果咱们订阅了 Mysql 的 binlog 来进行数据异构。音讯要是没有程序，就会呈现数据错乱问题。

比方新增一条 id=1 的数据，而后马上删除。这样就产生了两条音讯。失常的生产程序是先新增，而后删除，此时数据是没有的。如果音讯没有程序，删除的先被生产了，而后生产新增的，此时数据还在，没被删除掉，就会导致不统一。

定时音讯

定时音讯是指音讯具备定时发送的性能，当音讯发送到服务端后，不会立刻投递给消费者。而是要等到音讯指定的工夫后才会投递给消费者进行生产。

提早音讯也就是定时音讯，定时音讯是定在某个工夫点进行发送，比方 2020-11-11 12:00:00 发送。

提早音讯个别是在以后发送工夫的根底上提早多久进行发送，比方以后工夫是 2020-09-10 12:00:00，提早 10 分钟，那么音讯发送胜利后将在 2020-09-10 12:10:00 进行投递给消费者。

定时音讯能够在订单超时未领取主动勾销等场景应用。

事务音讯

RocketMQ 提供相似 X/Open XA 的分布式事务性能，通过 RocketMQ 事务音讯能达到分布式事务的最终统一。

交互流程：

图片起源阿里云官网文档

  1. 发送方首先发送半事务音讯到 RocketMQ 服务端。
  2. RocketMQ 服务端接管到音讯，而后将音讯长久化胜利之后，向发送方返回 Ack 确认音讯曾经发送胜利，此时音讯为半事务音讯，不会投递给生产方。
  3. 收到半事务音讯的 Ack 后，发送方开始执行本地事务逻辑。
  4. 发送方依据本地事务执行后果向服务端提交二次确认，如果本地事务执行成则进行音讯的 Commit，如果执行失败则进行音讯的 Rollback，服务端收到 Commit 状态则将半事务音讯标记为可投递，生产方最终将收到该音讯；服务端收到 Rollback 状态则删除半事务音讯，生产方将不会收到该音讯。
  5. 如果出现意外状况，步骤 4 没有进行音讯的二次确认，期待固定工夫后服务端将对该音讯发动音讯回查。
  6. 发送方收到音讯回查后，须要查看对应音讯的本地事务执行的最终后果。发送方依据查看失去的本地事务的最终状态再次提交二次确认，服务端仍依照步骤 4 对半事务音讯进行操作。

最佳实际

音讯重试

音讯在生产方生产失败后，RocketMQ 服务端会从新进行音讯的投递，晓得消费者胜利生产音讯，当然重试有次数限度，默认 16 次。

音讯重试在肯定水平上保障了音讯不失落，通过重试来达到最终被生产的目标。须要留神的是消费者在生产的时候肯定要等本地业务胜利后能力进行 ACK(生产确认)，不然就会呈现生产失败，然而曾经 ACK，音讯将不会反复投递。

如果采取异步生产的形式，须要进行异步转同步，等异步操作完才进行 ACK，具体能够参考我之前写的一篇文章 https://mp.weixin.qq.com/s/Bbh1GDpmkLhZhw5f0POJ2A。

最初须要做好对应的监控，如果重试了 4，5 次还是失败的，基本上前面重试也是失败的。这个时候须要让开发人员晓得，该人工解决的就人工染指。或者间接监控死信队列。

音讯过滤

音讯主题，个别用于一类音讯的对立分类。比方订单主题，然而订单下的音讯会分为很多种。比方创立订单，勾销订单等。

不同类型的音讯有不同的业务解决，咱们能够对立定义音讯格局，而后通过一个字段去辨别音讯类型来做不同的业务逻辑。不好的点在于所有音讯都会推送到生产方，不能按需生产。

在 RocketMQ 中能够给音讯指定 tag，通过 tag 来辨别音讯类型。消费者能够依据 Tag 在 RocketMQ 服务端实现音讯过滤，以确保消费者最终只生产到其关注的音讯类型。

我已经遇到过一个 tag 没有正确应用的形式，只有一个 MQ 实例，用 tag 来辨别环境。所有音讯都在一个主题中，测试环境生产测试环境的 tag，线上生产线上的 tag。

这种形式的问题在于音讯没做隔离，线上线下的音讯都在一起。另一个就是 tag 被固定成了环境的辨别，无奈用于音讯类型场景，导致只能建多个 topic 来承载多个业务音讯类型。

生产模式

RocketMQ 生产模式有两种，集群生产和播送生产。

集群生产：

消费者部署了多个实例咱们称之为一个集群，集群生产只会被其中的某一个实例进行生产。

适宜大部分的业务场景，大部分的场景咱们的音讯只容许被生产一次，而且只能有一个消费者去生产，比方领取回调场景，如果一个音讯被多个实例同时生产，那么就会呈现同时去批改订单状态，同时去扣减库存的状况。

播送生产：

播送生产会让集群中每个实例都生产一次。

比方咱们应用了本地缓存，当数据变更的时候，咱们须要刷新每个节点本地的缓存，所以每个节点都须要收到音讯。

生产幂等

幂等问题，无论是在 API 申请场景还是在音讯生产场景，都会遇到。一条音讯不能反复生产屡次这个必定是要保障的，因为咱们不能保障音讯发送方不发送屡次，也不能保障音讯不反复投递。

RocketMQ 的 Exactly-Once 投递语义，就是用于解决幂等问题。Exactly-Once 是指发送到音讯零碎的音讯只能被生产端解决且仅解决一次，即便生产端重试音讯发送导致某音讯反复投递，该音讯在生产端也只被生产一次。

最佳的幂等解决形式还是须要有一个惟一的业务标识，尽管每条音讯都有 MessageId，然而不倡议用 MessageId 来做幂等判断，在发送音讯的时候，能够为每条音讯设置一个 MessageKey，这个 MessageKey 就能够用来做业务的惟一标识。

对于幂等怎么解决，就不细讲了。能够参考我之前写的一篇文章 https://mp.weixin.qq.com/s/9fhqnbeXPz7-7x0Eadd8DA，通用的幂等实现计划。

本地事务音讯封装

下面介绍了事务音讯，RocketMQ 的事务音讯采纳了二阶段提交的形式。并且联合了音讯反查的机制来确保最终一致性。

从应用层面来说，每个业务场景都要去实现一个反查的逻辑，有点烦。

上面介绍另一种常常被应用的形式，就是本地事务音讯。本地音讯表这个计划最后是 ebay 提出的，本地事务音讯须要在服务对应的数据库中创立一个音讯表，发送音讯的时候不是真正的将音讯发送给 MQ，而是往音讯表中插入一条音讯数据。

插入的动作跟本地的业务逻辑是同一个事务，如果本地事务执行胜利，音讯才会落表胜利，才会发送给 MQ, 本地事务失败，音讯数据回滚。

而后须要有一个专门的程序去拉取音讯表中未发送的音讯投递给 MQ，如果投递失败，能够始终重试，直到胜利或者人工染指。

音讯写到音讯表，而后会始终给 MQ 发送，这个步骤没问题。如果 MQ 收到音讯后，音讯还在 PageCache 中的时候，Broker 宕机了，这个时候是会呈现音讯失落。当然你也能够应用同步刷盘等形式来防止失落。如果咱们就是异步刷盘，有方法保障音讯不失落吗？

后面咱们提到，RocketMQ 的事务音讯会有回查的机制，音讯表的形式，也须要有一个机制来保障音讯被生产了，否则就须要一直的重试去发送音讯，直到音讯被生产。

在音讯表中须要有一个字段来标识以后这条音讯的状态，比方 未发送，已发送，已生产。当音讯还是未发送的时候就会被发送到 MQ, 如果发送胜利了，状态就是已发送。然而过了几分钟，状态还是已发送，这个时候就要去做一些动作了。

这个场景下，有可能是消费者跟不上生产的速度，音讯沉积了，导致音讯始终没被生产。另一种可能就是音讯是不是失落了？

能够获取对应的音讯沉积数据来判断是否音讯沉积了，如果不是就从新发送音讯给 MQ，晓得音讯被生产。

问题是音讯被生产了，我怎么晓得？

像我是用的云服务，是有对应的 Open API 能够间接查问音讯轨迹。开源的应该也有，没有认真去钻研，跟商业版应该差不多。

依据音讯轨迹就能够晓得音讯有没有被生产，到此为止流程完结。音讯发送给 MQ 如果失败会重试，音讯如果长时间没生产，也会从新发送，即便最初进入了死信队列，也能够通过死信队列的监控来人工干预，肯定会是最终一致性。

跟自带的事务音讯比，本地音讯表的形式不须要实现回查逻辑，然而要减少音讯表，同时也要配套各种发送，查看等逻辑，也挺麻烦了。特地是当音讯量大的时候，如何疾速的将音讯表中的音讯发送进来，也须要做很多解决，简略的查表轮询在量大的状况下不太实用。

两种形式都能够应用，能实现咱们要的目标即可。

参考代码

本地事务音讯相干的代码能够微信搜寻「猿天地」，回复关键字「kitty」获取即可。

码字不易，能够的话来个三连击，感激！

对于作者 ：尹吉欢，简略的技术爱好者，《Spring Cloud 微服务 - 全栈技术与案例解析》,《Spring Cloud 微服务 入门 实战与进阶》作者, 公众号 猿天地 发起人。

我整顿了一份很全的学习材料，感兴趣的能够微信搜寻「猿天地 」，回复关键字「 学习材料」获取我整顿好了的 Spring Cloud，Spring Cloud Alibaba，Sharding-JDBC 分库分表，任务调度框架 XXL-JOB，MongoDB，爬虫等相干材料。

退出移动版