从 RocketMQ 5.0 开始，社区引入了 POP 这种消费模式。客户端发起 POP 请求后，Broker 从 queue 中取出消息返回，并在 Broker 侧维护不可见期、ACK、续租和超时恢复这套状态——尤其在非 FIFO 场景下，它允许多个消费者并发处理同一 queue 上的消息。 在实际业务里，POP 往往不是因为“想换一种消费接口”才被拿出来，而是在消费并发已经碰到瓶颈时，才真正体现出价值。 一个很典型的场景是：某个 Topic 的 queue 数已经拉到很高，比如 500；消费客户端也已经拉到 500；但一个客户端处理一个 queue，吞吐还是不够。写入侧没有明显瓶颈，真正卡住的是消费侧并发。这个时候，业务真正需要的，往往不是继续把 Topic 切得更碎，而是让多个客户端共同处理同一个 queue 上的数据。 从这个场景往下看，POP 可以先理解成一种“消息被 Broker 借出一段时间”的消费模型： + 消息取走后，不会立刻算完成 + 消息会先进入不可见期 + 处理成功后再 ACK + 处理时间不够时可以续租 + 超时后还要重新进入后续消费流程 顺着这个语义再来看源码，后面的 invisibleTime、ACK、changeInvisibleTime、revive，以及 Broker 里那套“已投递但未完成”的状态管理，就比较容易串起来。这里把 Broker 为消息保存的这类“消费中”状态统称为“投递记录”。 图注：POP 消费的核心语义——消息被 Broker 借出一段时间，取走后进入不可见期，成功后 ACK，不够则续租，超时则 Revive 重投到 Retry Topic。 一、为什么需要 POP ▍1.1 传统 Pull 为什么不太够用 传统 Pull 模式下，消费进度主要由消费位点驱动；在同一消费组内，一个队列在同一时刻通常只会稳定地分配给一个消费者实例。 这套模型很适合“按位点往前推进”的流式消费，但在下面这类场景里，局限会暴露得更明显： + 单条消息处理时间明显长于一次普通消费循环 + 处理成功与否不能只靠“位点推进了没有”来表达 + 客户端需要显式告诉 Broker：我处理完了、还没处理完、需要再给我一点时间 + 更希望消费语义接近“消息先领出处理、完成后再确认”，而不是单纯按 queue 上的位点顺序往前推进 问题不再只是“怎么把消息拉出来”，而是“怎么让消息被领出来处理一段时间，同时还能继续放大消费并发”。 单靠继续加 queue，本质上还是在用更多 queue 换并发；但当 queue 数已经很高时，很多时候业务更需要的，其实是允许多个客户端共同处理同一个 queue 上的数据。 这也正是非 FIFO POP 真正有意义的地方：它把原来更接近“一个 queue 稳定分给一个消费者实例”的消费方式，往“同一 queue 上可以存在多条被不同消费者领走、但尚未最终确认的消息”这个方向推进了一步。 ▍1.2 POP 的关键变化在哪里 POP 真正补上的，不是一个新的拉消息动作，而是 Broker 侧一层明确的“处理中状态”。 消息被取走后，不会立刻算消费完成，而是先进入一段受 Broker 管理的处理中窗口：这段时间里消息不可见，成功后 ACK，处理不完可以续租，超时则重新进入后续消费流程。 这样一来，Broker 就可以同时跟踪同一 queue 上多条“已经投递、但尚未最终完成”的消息，POP 能支撑更高消费并发的关键也在这里。 后面会反复出现的 invisibleTime、ACK、changeInvisibleTime、revive，本质上都是围绕这层状态展开的。 更详细的对比如下图： 二、POP 的几个核心概念 下面讨论 RocketMQ 5.4.0 里的 KV 路径，前面说的“投递记录”在这里具体指 PopConsumerRecord。旧路径和 FIFO 放到后文再补。 ▍2.1 invisibleTime invisibleTime 是 POP 最核心的语义。消息被 POP 成功后，在 popTime + invisibleTime 之前，这条消息对该消费组后续的 POP 请求不可见。 如果只抓住一件事，那就是：Broker 把消息借出去之后，不会立刻把它算作完成，而是先给它挂上一段“处理中窗口”。这段窗口就是 invisibleTime。 ▍2.2 ACK ACK 的核心语义不是简单推进消费位点，而是把“这条消息已经完成”这件事落到 Broker 侧状态里。 更具体一点说，就是把这条消息对应的 PopConsumerRecord 删除掉。它不是“把 queue 往前推一步”，而是把这条消息从“已投递但未完成”的集合里移出去。 ▍2.3 changeInvisibleTime 如果业务处理超过原来的 invisibleTime，客户端可以调用 changeInvisibleTime 续租。从语义上看，它不是在改消息内容，而是在把这次投递的可见时间整体往后推。 放到实现上，它体现为： + 写一条新的 PopConsumerRecord + 再删除旧记录 所以它本质上是在切换“这次投递对应的超时点”。 ▍2.4 revive revive 主要对应超时恢复机制。 如果一条消息在不可见窗口内一直没有被 ACK，Broker 后续会把它重新投递到该消费组对应的 retry topic，后续 POP 请求再按 popFromRetryProbability 的轮转策略从 retry topic 取消息，这个过程就是 revive。也就是说，revive 不是把消息直接放回原始 queue，而是走重试投递路径。 需要特别注意的是，revive 并不是一个“附加的重试能力”，而是这条消费模型下的必然结果。 从 Broker 视角看，一旦消息返回给客户端后，后续读取位置已经继续前移，而这条消息本身又仍处于“已投递但未确认”的状态，Broker 就不能再只靠 offset 把它重新取回来。 因此，如果客户端在不可见期内一直没有 ACK，Broker 就必须依赖这层消息级未确认状态的扫描与 revive 机制，把消息重新投递到 retry topic，再进入后续 POP 消费流程。 这也是为什么 revive 本质上是在弥补“offset 已前移，但消息尚未完成”的状态缺口。 ▍2.5 POP 还有 offset 吗？ 有。 Broker 至少同时维护两层状态： + pullOffset / consumeOffset —— 决定后续从哪里继续读或推进 + PopConsumerRecord —— 表达“哪些消息已经投递出去、但还没有最终完成” 所以 offset 前移不等于消息已经真正消费完成，只有对应的 PopConsumerRecord 被 ACK 删除、续租切换或超时恢复后，这条消息的状态才算真正闭环。 很多人第一次看 POP，会误以为“既然已经按消息 ACK 了，那是不是就没有 offset 了”。实际上并不是。POP 里 offset 一直都在，只是进入 POP 之后，offset 不再足以单独表达全部消费状态。 如果把 Pull 和 POP 放在一起看，差别不在“有没有 offset”，而在“offset 是否还足够表达全部消费状态”： + 对 Pull 来说，offset 基本就能同时表达“从哪里继续读”和“消费推进到了哪里” + 但对 POP 来说，Broker 侧还多了一层“已投递但未完成”的消息级状态 后面整篇文章，其实就是在把这层状态是怎么建立、怎么变更、怎么收尾讲清楚。 三、先按一条主线读源码 RocketMQ 5.4.0 里新旧两条实现路径仍然并存，但如果一开始就把两套状态模型交叉着讲，读者很容易在“当前这一步到底落在哪套实现里”上迷路。 所以下面只顺着一条线往下走：PopMessageProcessor → PopConsumerService → PopConsumerRecord → ACK / changeInvisibleTime / revive。旧路径和 FIFO，都放到后面再统一说。 图注：POP 源码的三层架构——请求入口（PopMessageProcessor）、核心服务（PopConsumerService）、状态存储（PopConsumerCache + PopConsumerKVStore）。核心投递记录 PopConsumerRecord 贯穿全流程。 ▍3.1 统一入口：PopMessageProcessor POP 请求统一从 PopMessageProcessor.processRequest() 进入。它负责： + 校验 Broker / Topic / ConsumerGroup 权限 + 校验 maxMsgNums，当前限制最大 32 + 校验 queueId + 校验 timerWheelEnable，没开时 POP 直接拒绝 + 构造过滤表达式和 MessageFilter + 根据配置决定走哪条 POP 实现路径 + 在没取到消息时进入 PopLongPollingService 代码位置：broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java，入口方法 processRequest(...) 它是 POP 的入口，但不是全部逻辑所在。 ▍3.2 核心组件：PopConsumerService 当brokerConfig.isPopConsumerKVServiceEnable() 为 true 时，POP 主流程会委托给 PopConsumerService.popAsync(...)。 这一套是这里真正的主角，核心组件包括： PopConsumerContext 封装单次 POP 的上下文，保存本次返回的 GetMessageResult 和本次生成的 PopConsumerRecord，汇总返回给客户端的 startOffsetInfo、msgOffsetInfo。 PopConsumerRecord 表示“一条消息被 POP 出去但还没最终确认”的记录。核心字段：popTime / groupId / topicId / queueId / retryFlag / invisibleTime / offset / attemptTimes / attemptId。其中 attemptTimes 是 revive 重试次数，每次 revive 失败后递增，用于退避策略。visibilityTimeout = popTime + invisibleTime。 PopConsumerKVStore 默认是 RocksDB 存储实现。 PopConsumerCache enablePopBufferMerge 打开时的内存缓冲，缓冲区定期刷盘或触发 revive。 PopConsumerLockService 按 groupId + topicId 做互斥。 代码位置：broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java，入口方法 popAsync(...) 四、POP 请求主流程 下面把一次 POP 请求从入口到返回完整走一遍。 图注：一次完整 POP 请求的 10 步主流程 + 长轮询备选路径（A/B/C）。注意长轮询本质仍是“请求驱动的拉取”，唤醒时重新调用 processRequest()。 ▍4.1 请求进入 Broker 客户端发 POP_MESSAGE 请求后，PopMessageProcessor 先做一轮硬校验： + Broker 是否可读 + Topic 是否存在且可读 + ConsumerGroup 是否存在且允许消费 + requestHeader.getMaxMsgNums() ≤ 32 + Store 是否开启 timer wheel + queueId 是否合法 如果请求带过滤表达式，还会构建普通 Topic 和对应 retry topic 的 SubscriptionData，以及对应的 ExpressionMessageFilter。 ▍4.2 创建 PopConsumerContext 进入 PopConsumerService.popAsync(...) 后，会先创建 PopConsumerContext，里面记录：客户端地址、popTime、invisibleTime、消费组、attemptId、本次累积返回的消息结果和投递记录。这个对象贯穿单次 POP。 ▍4.3 加锁粒度：groupId + topicId PopConsumerService 会先尝试对 groupId + topicId 加锁。注意这里不是“每个队列一把锁”，而是消费组和 Topic 粒度的互斥。 这么做的目的，主要是保证一次 POP 紧密相关的状态变更不会互相踩踏：取消息、建投递记录、提交 offset，都在这把锁保护下串起来。 ▍4.4 先取重试还是先取普通消息 POP 不是永远先取普通 Topic。PopConsumerService 里会根据配置决定是否优先取重试消息： + popFromRetryProbability + popFromRetryProbabilityForPriority 并结合以下条件算出 preferRetry：Topic 类型是否为 PRIORITY，且 requestCount % 100 拿到 probability 后，preferRetry = (probability 0) && (requestCount % (100 / probability) == 0)。 取消息的实际顺序也与 preferRetry 和 queueId 模式有关： + preferRetry=true：先取 retry topic（v1/v2），再取普通 Topic，不再重复取 retry + preferRetry=false 且 queueId==1：先取普通 Topic，再取 retry topic（v1/v2） + preferRetry=false 且 queueId!=1：只取指定普通队列，这次请求里不会再补读 retry topic 也就是说，它不是简单的“普通队列空了再去重试队列”，而是带概率控制、并且受 queueId 模式约束的优先级轮转策略。 ▍4.5 支持 retry topic v1/v2 这里会根据配置决定是否从这些重试主题取消息：KeyBuilder.buildPopRetryTopicV1(...) 和 KeyBuilder.buildPopRetryTopicV2(...)。并且 retryFlag 会记在 PopConsumerRecord 里，后面 revive/返回消息时都要用到。 ▍4.6 queueId 的两种模式 指定队列：如果 queueId != 1，只从指定队列取。 全队列 POP：如果 queueId == 1，会遍历 Topic 的所有读队列。遍历起点和方向不是固定的：结合请求计数做轮转，尽量避免每次都从固定队列开始；同时受 priorityOrderAsc 控制，决定从低 queueId 向高递增，还是从高 queueId 向低递减。 ▍4.7 真正取消息 每个队列最终都会调用 getMessageAsync(...) 去读存储。这里有几个关键点： + POP 使用的是 pullOffset + 如果发现 offset 过小/过大/无效，会修正 offset 后重试读取 + 读到消息后，会同步更新 commitPullOffset(...) 和 commitOffset(...) 要注意：commitPullOffset 更像“下一次从哪里继续试探着拉”；打开 enablePopBufferMerge 时，commitOffset 可能会被压到 PopConsumerCache 里最小的未确认 offset，而不是这次读取的 nextBeginOffset。 commitOffset 仍然是 POP 在 Broker 侧维护的一部分消费进度，但真正决定一条消息是否已经消费完成的，仍然是后面的 ACK、续租和 revive。 ▍4.8 inflight 背压：isPopShouldStop 在每个队列真正取消息之前，PopConsumerService 还会先调用 isPopShouldStop(groupId, topicId, queueId) 做背压判断。 当 enablePopMessageThreshold=true 且该队列当前 inflight 消息数（由 PopConsumerCache 统计的未 ACK 记录数）已经达到 popInflightMessageThreshold 时，isPopShouldStop 返回 true，这次队列迭代直接跳过，不再读存储。 这和长轮询里的 POLLING_FULL 是两个独立的流控层： + POLLING_FULL：长轮询挂起队列容量已满，新请求无法挂起 + isPopShouldStop：某个队列的 inflight 消息过多，主动限制继续发送 ▍4.9 读到消息后如何建模 PopConsumerContext.addGetMessageResult(...) 会先在本次请求上下文里，为每条命中的消息挂一条 PopConsumerRecord。 可以把它理解成：Broker 在说“这条 offset=xxx 的消息，我在 popTime 时刻交给了这个消费组，它要在 invisibleTime 后才算超时。” 这是 POP 跟 Pull 最大的不同。但要注意，这一步只是先把“投递记录”挂在 PopConsumerContext 里；真正写到缓存或 KV Store，是下一步的事。 ▍4.10 投递记录写到哪里 如果这次 POP 确实取到了消息，这些 PopConsumerRecord 就会被真正保存下来： + 如果 enablePopBufferMerge=true 且缓存没满，先写 PopConsumerCache + 否则直接写 PopConsumerKVStore（默认 RocksDB） 所以这里真正的核心投递记录就是 PopConsumerRecord。 投递记录写入路径图 图注：PopConsumerRecord 从创建到持久化的完整写入路径。enablePopBufferMerge=true 时经内存缓冲刷盘，否则直写 RocksDB。存储键为 visibilityTimeout + groupId + topicId + queueId + offset。 ▍4.11 返回给客户端前的重试消息改写 如果取到的是 retry topic 上的消息，返回客户端前，Broker 可能会对“客户端看到的 topic”做一层重编码；但这里要特别注意两件事： + 这一层更多是“响应呈现层”的行为，不等于 ACK/续租时真正使用的 topic + 客户端后续 ACK、续租最终依赖的仍然是 PROPERTY_POP_CK / receipt handle 里携带的 retry 标记，客户端会据此恢复 real topic 所以从客户端视角看，“日志里看到的消息 topic”和“Broker 这次实际从哪个主题取到消息”，不一定始终一一对应。这个问题后面再单独展开。 ▍4.12 没取到消息时的长轮询 如果本次 POP 没取到消息，不一定立刻返回空。只要请求里的 pollTime 0，请求就会进入 PopLongPollingService.polling(...)。 长轮询服务会把请求按 topic + consumerGroup + queueId 挂到 pollingMap 里。已经挂起的请求，主要会在下面几种场景被重新处理： + 新消息到达 + 挂起超时 + 服务停止时的清理唤醒 这里的“重新处理”不是 Broker 手里已经攥着一批消息，等条件满足后直接把它推给挂起请求，而是重新调用一次 processor.processRequest(...)，也就是再走一遍 POP 处理流程。 需要特别注意的是，POP 的长轮询本质上仍然是“请求驱动的拉取”，而不是“服务端主动推送”。 要额外注意的是，POLLING_FULL 不是唤醒场景。如果长轮询队列本身已经满了，请求不会进入挂起队列，而是会直接返回 POLLING_FULL。 五、ACK、续租和 revive 消息返回客户端之后，还剩三件事要闭环：处理完成怎么 ACK，时间不够怎么续租，超时之后又怎么恢复。 ▍5.1 ACK ACK 请求入口是 AckMessageProcessor。最重要的是它最终会落到 PopConsumerService.ackAsync(...)。它的主流程可以理解成这样： 1. AckMessageProcessor.appendAckNew(...) 先从 extraInfo 或 BatchAck 里还原这次 POP 的关键信息 2. 然后调用 PopConsumerService.ackAsync(...) 3. ackAsync(...) 会根据 popTime/groupId/topicId/queueId/invisibleTime/offset 构造对应的 PopConsumerRecord 作为查找 key；底层存储键最终会折算成 visibilityTimeout + groupId + topicId + queueId + offset 4. 然后先尝试从 PopConsumerCache 删除；删不到再去 PopConsumerKVStore 删除 所以这里的 ACK，就是把这条消息对应的“进行中记录”删掉。 ▍5.2 为什么 ACK 不是简单提交 offset 因为 POP 是“消息级确认”，不是“队列级顺序推进”。同一批次返回的多条消息里，可能有的已经处理完成，有的还在处理中，所以 Broker 必须保留消息级的未确认状态，而不能只靠一个队列 offset 表达完成状态。 放在这里看，这层状态就是 PopConsumerRecord。 ▍5.3 changeInvisibleTime ChangeInvisibleTimeProcessor 处理续租请求。它最终会进入 PopConsumerService.changeInvisibilityDuration(...)。这条分支会做两件事： + 写入一条新的 PopConsumerRecord（changedPopTime + changedInvisibleTime） + 删除旧的 PopConsumerRecord 本质上等于把原来的“超时点”整体往后挪。 要注意，这里新生成的记录不是重新写回 PopConsumerCache，而是直接写 PopConsumerKVStore，然后再删除旧记录。 ▍5.4 超时未 ACK 后如何 revive PopConsumerService 自身就是一个背景线程服务，会定期扫描 PopConsumerKVStore 里的过期记录： + 实现上通过 scanExpiredRecords(...) 扫描已经进入过期窗口的 PopConsumerRecord + 再去读原始消息 + 把原始消息重写到 retry topic + 删除旧记录 如果打开了 enablePopBufferMerge，还要再补一层理解：部分记录会先进入 PopConsumerCache，缓存线程会先把未过期记录刷到 store，把已过期记录直接交给 revive 逻辑处理。所以“后台线程只扫 KVStore”是主路径，但不是全部。 如果 revive 失败，不是无限重试，而是会按 attemptTimes 做退避重试。超过最大尝试次数后，日志里会出现 message may be lost。 ▍5.5 revive 重投到哪里 如果原消息本来就来自重试主题，则继续回到对应重试主题。如果原消息来自普通 Topic，则会重投到 POP retry topic（v1 或 v2，取决于 Broker 配置），并且会带上 PROPERTY_FIRST_POP_TIME、PROPERTY_ORIGIN_GROUP 和最新的重试次数。 图注：ACK 删除 PopConsumerRecord 标记完成，changeInvisibleTime 写新记录+删旧记录将超时点后推。两者操作路径、写入目标、最终效果均不同。 六、最后补三个容易混的点 ▍6.1 FIFO 为什么不是这套模型 前面讲的这套模型里，核心一直是“每条消息对应一条 PopConsumerRecord，再围绕它做 ACK、续租和 revive”。FIFO 不是这套思路。 FIFO 更像“带不可见期的顺序消费”。Broker 关心的不是每条消息各自有没有一条独立投递记录，而是同一队列上的顺序窗口有没有被前面的未 ACK 消息卡住。对应到源码上，核心状态从 PopConsumerRecord 换成了 OrderInfo，由 ConsumerOrderInfoManager 维护。 所以 FIFO 不走前面那套消息级 retry/revive 主流程。它的超时恢复也不是后台扫描，而是后续 POP 通过 OrderInfo.needBlock() 判断阻塞窗口是否已经过期；一旦阻塞解除，因为 committedOffset 还没前进，消息会从原始 queue 原地重新投递，而不是进 retry topic。 ▍6.2 旧路径在做什么 旧路径和前面讲的是同一个问题的另一套实现。如果 isPopConsumerKVServiceEnable() 为 false，Broker 不再用 PopConsumerRecord 表示“已投递未确认”，而是换成 PopCheckPoint / AckMsg / BatchAckMsg 这一组对象。 大致可以把它理解成： + POP 取消息后生成 PopCheckPoint + ACK 通过 AckMsg / BatchAckMsg + PopBufferMergeService 表达 + changeInvisibleTime 会追加新的 PopCheckPoint，再去 ACK 旧 checkpoint + 超时恢复依赖 revive topic 和 PopReviveService 旧路径和前文解决的是同一个问题，但中间状态模型已经完全换了一套。理解到 checkpoint、ack 日志和 revive topic 这一层就够了。 ▍6.3 涉及重试消息时，客户端看到的 topic 为什么会不同 这里最容易混淆的一点是：客户端看到的消息 topic，不等于 ACK / 续租真正作用的 real topic。 POP 返回消息时，Broker 可能会对响应里的 topic 做重编码，尤其是消息实际来自 retry topic 时更容易让人误会。但客户端后续 ACK / changeInvisibleTime 依赖的并不是当前消息对象上的 topic，而是 PROPERTY_POP_CK / receipt handle 里的 retry 标记，再据此恢复 real topic。 所以仅凭客户端日志里看到的 topic，不能反推 Broker 这次一定是从普通 topic 读到的，还是从 retry topic 读到的。把这件事分开看就不容易绕：日志里的 topic 更像是响应呈现层，ACK / 续租用的 topic 才是语义层。 图注：客户端日志中的 topic 属于“呈现层”（可能被 Broker 重编码），而 ACK/续租使用的 real topic 属于“语义层”（来自 PROPERTY_POP_CK 中的 retry 标记）。两者不能混为一谈。 七、再往外看：为什么 Kafka 也在演进这类能力 如果把 POP 从 RocketMQ 自己的实现细节里跳出来看，它解决的其实不是一个 RocketMQ 独有的问题，而是一类更普遍的消息消费诉求： + 共享消费 + 显式确认 + 处理中窗口控制 + 超时后的再次投递 RocketMQ POP 是其中一种实现方式。它仍然建立在 RocketMQ 自己的 Topic、queue、retry topic、FIFO/非 FIFO 分支之上，但消费语义上已经明显引入了“消息被借出一段时间、完成后确认、超时后恢复”的队列式特征。 Kafka 近几个版本也在沿着这个方向演进。按照 Apache Kafka 官方文档和 KIP932 的表述： + Kafka 4.1 把 share groups 作为 preview 形态引入 + Kafka 4.2 把 Queues for Kafka / share groups 标记为 productionready + 它提供的是一种不同于传统 consumer group 的共享消费模型 + 不再要求每个 partition 只能分配给一个消费者实例处理 + 消费者可以做逐条 ACK，并跟踪 delivery attempts；同时系统仍然针对 batch processing 做了优化 + 官方文档也明确说，这类能力更适合“records are processed one at a time”，而不是典型的 ordered stream 如果只看高层语义，会发现它和 POP 想解决的问题其实很像：都在补“共享消费 + 显式确认 + 失败后继续处理”这类能力。当然，两者并不是一套实现： + RocketMQ POP 的关键语义是 invisibleTime、ACK、续租、revive，以及围绕这些语义建立的 Broker 侧投递记录 + Kafka share groups 的关键语义是共享消费组、逐条 ACK、delivery attempts，以及围绕 share group 建立的新协调与状态管理 这也说明，“共享消费、逐条确认、处理中窗口控制、失败后继续处理”并不是某一个消息系统偶然长出来的特性，而是消息系统里真实存在的一类业务需求。不同消息系统都在沿着各自的体系往这个方向演进，只是具体实现路径和抽象方式并不相同。 八、关键源码索引 以下以 release5.4.0 分支、提交 b5da00ad0 为代码基线。 | 功能 | 关键类 / 方法 | | :: | | | POP 请求入口 | PopMessageProcessor.processRequest() | | KV 路径主流程 | PopConsumerService.popAsync() | | 重试消息重编码 | PopConsumerService.recodeRetryMessage() | | ACK 主流程 | AckMessageProcessor.processRequest()→ appendAckNew()→ PopConsumerService.ackAsync() | | 续租主流程 | ChangeInvisibleTimeProcessor.processRequestAsync()→ PopConsumerService.changeInvisibilityDuration() | | revive 主流程 | PopConsumerService.revive(PopConsumerRecord) / revive(AtomicLong, int) | | 旧路径buffer merge | PopBufferMergeService | | 旧路径 revive | PopReviveService | | FIFO 顺序状态 | QueueLevelConsumerManager | | 长轮询挂起 | PopLongPollingService.polling() | | 长轮询唤醒 | PopLongPollingService.wakeUp() | 总结 POP 消费机制的核心可以概括为三句话： 第一，POP 补上了 Broker 侧的“处理中状态”。 消息被取走后不立刻算完成，而是进入 invisibleTime 不可见窗口，Broker 通过 PopConsumerRecord 跟踪每条“已投递但未完成”的消息。 第二，消费进度不再仅靠 offset 表达。 offset 决定从哪里继续读，PopConsumerRecord 决定哪些消息还在处理中。ACK 删除记录、续租切换超时点、revive 处理超时恢复，三者共同构成状态闭环。 第三，这是消息系统的一类共性需求。 从 RocketMQ 的 POP 到 Kafka 的 share groups，不同系统都在用各自的方式实现“共享消费 + 显式确认 + 处理中窗口控制”，因为这是实际业务中的真实诉求。

随着大语言模型（LLM）、语音识别（ASR）、语音合成（TTS）等能力逐步成熟，AI Agent 开始从文本交互走向语音交互，典型场景包括 AI 教师、AI 情感聊天、AI 助手等。相比文本输入，语音更自然、更实时，用户可以直接通过说话完成提问、练习、任务触发与多轮对话，这也让“和 Agent 用语音对话”真正进入实际业务场景。 但当 Agent 语音交互进入高并发场景后，很多团队会发现：最先遇到瓶颈的，往往不是模型本身，而是支撑实时交互的消息链路。海量会话管理、高频小包传输、异步结果回推、会话生命周期管理等问题会随之集中出现。要让和 Agent 语音交互真正做到更稳、更快，底层链路设计往往才是关键。 本文结合一个典型的高并发智能语音交互场景，介绍如何基于阿里云 RocketMQ LiteTopic 构建一套更稳定、更可靠、更高效的实时语音消息链路架构。 高并发 Agent 语音交互对技术架构提出哪些关键要求？ 在 Agent 语音交互场景中，系统并不只是“接收一句话、返回一句话”这么简单。一次完整交互背后，往往涉及客户端、网关、业务处理系统，以及 LLM、ASR、TTS 等多个服务之间的协同。 因此，智能语音交互业务会对技术架构提出更高要求： + 海量会话管理：随着业务规模增长，并发连接数和活跃会话数会迅速上升。每个用户的语音交互都是一个独立会话（Session），系统需要同时维持数万甚至数十万个长连接。 + 高频小包传输：用户按下录音键到松开，形成一个独立会话（Session）。在此期间，客户端会将音频流切片成小包持续传输，需要保证语音包连续、不丢失，避免影响业务正确性。 + 严苛的时效性：客户端对延迟极度敏感，若较长时间内未收到响应，用户体验会明显下降。这对 LLM 在高并发场景下的吞吐能力，以及系统的实时响应通知能力，都提出了更高要求。 正因如此，很多系统在低并发时看起来“可以跑通”，但一旦进入高并发实时语音场景，底层消息架构中的问题就会被迅速放大。 传统消息架构在实时语音场景下面临哪些核心挑战？ 在智能语音交互业务的实际落地过程中，传统消息架构在支撑高并发、低延迟的实时语音场景时，往往会暴露出几类典型问题。 ▍1. 全链路 Session Sticky 的精准路由 语音交互的消息流转路径通常贯穿：APP Gateway BizProcessSystem (Route) LLM/ASR/TTS。其中，APP与 Gateway、BizProcessSystem与 LLM之间均维持着 WebSocket 长连接。 在这种架构下，整个链路必须严格保持会话粘滞（Session Sticky）。也就是说，某个用户的上行音频流和下行反馈结果，必须精准路由到其当前连接的特定网关节点，以及对应的后端处理实例。 问题在于，在分布式环境下，维护“Session ID 到物理节点 IP”的动态映射表本身就非常复杂。一旦网关节点扩容、重启，或者发生网络波动，路由表同步延迟极易导致消息被投递到错误节点，进而造成连接断裂、数据丢失，破坏交互连续性。 ▍2. 大模型异步结果的实时、精准回推 大模型（LLM）的推理过程通常耗时较长且波动明显（秒级甚至分钟级）。若采用同步等待模式，会长时间占用网关和业务线程，导致系统吞吐量急剧下降，甚至引发资源耗尽的雪崩效应。 因此，为了提升吞吐，LLM 调用通常需要改造成异步处理。但异步之后，新的难点也随之出现：最终计算结果（如 ASR 文本、TTS 音频）如何实时、准确地回推给发起请求的那个用户连接？ 如果依赖复杂的回调轮询或状态查询，不仅实现复杂，还会进一步增加延迟和维护成本。这也是语音交互架构设计中的核心难点之一。 ▍3. 海量临时通道带来的元数据爆炸 从业务逻辑上看，为每个独立语音会话（Session）建立隔离的通信通道，是避免数据串扰的理想方案。 但如果为每个 Session 都创建一个标准 RocketMQ Topic，就会带来明显的元数据（Metadata）爆炸问题。海量临时 Topic 会严重消耗 NameServer 和 Broker 的内存与 CPU 资源，导致集群性能急剧下降，甚至影响可用性。 此前，一些场景会采用广播消息的方案来规避这一问题。虽然实现简单，但也存在两个明显缺陷： + 消息会在所有节点重复投递和过滤，造成大量无效流量与计算资源浪费。 + 所有节点都需要处理全量消息，单节点处理能力容易成为整体容量上限，系统水平扩展受限。 因此，广播模式很难支撑持续增长的高并发语音业务。 ▍4. 会话生命周期的自动化管理缺失 语音会话通常具有很强的临时性，并不是永久存在的资源。它的生命周期可能只是一次几分钟的对话，也可能仅存在于一个业务周期之内。 但在传统架构下，会话结束后的路由记录、缓存状态、临时通道等资源，往往需要依赖定时任务扫描或手动清理。这会带来两个典型问题： + 清理不及时：无效资源长期堆积，占用系统内存和计算资源。 + 清理过早：可能切断仍在进行中的合法交互。 因此，系统需要一种真正适合临时会话场景的机制，能够实现通道的自动创建、自动过期和自动销毁。 基于 RocketMQ LiteTopic 的消息链路重构 针对上述问题，可以基于阿里云云消息队列 RocketMQ 的轻量主题（LiteTopic）模型，构建一套更适合高并发智能语音交互场景的消息中间件架构。 LiteTopic 支持动态创建海量轻量主题，天然具备会话隔离能力，并内置 TTL 自动清理机制。这些特性与 Agent 语音交互场景对“高并发、低延迟、强隔离、易回收”的要求高度契合。 ▍1. RocketMQ LiteTopic 方案设计 1.1 请求保序与响应隔离的核心架构 + 请求侧：分片音频包采用分区顺序 Topic 上传 同一个会话的语音包用 SessionID 作为分区顺序的 Key，发送到分区顺序 Topic 中，相同 Key 的消息会按照发送顺序投递给业务处理系统，从而保证同一会话内消息处理有序。 + 响应侧：模型结果通过 LiteTopic 异步通知 为每个会话创建一个 LiteTopic：直接使用 SessionID 作为 LiteTopic 名称。每个语音会话拥有独立的消息通道，天然实现消息隔离。 每个节点订阅不同的 LiteTopic 集合：应用服务端节点作为 Consumer，只订阅与当前节点相关会话的 LiteTopic 集合，确保消息“点对点”精准投递，无需维护复杂路由表。 节点动态订阅 LiteTopic：会话断连后，可以动态删除对应 SessionID 的 LiteTopic 订阅；新会话建立时，可以动态新增对应 SessionID 的 LiteTopic 订阅。当网络异常或者服务节点重启时，也可以利用动态订阅能力续订 LiteTopic 消息，从而保障会话内容连续性。 LiteTopic 自动创建：LiteTopic 无需预先创建。当生产者发送消息时，如果发现 LiteTopic 不存在，会自动创建，且不影响消息发送耗时。 配置合适的 TTL 时间：当 LiteTopic 距离最近一次消息写入超过设定时长后，会被自动删除。可结合实际业务特点，设置合适的 TTL 时长。 1.2 面向运维和问题定位的可观测性 RocketMQ LiteTopic 基于云监控建立了全方位、细粒度的监控、告警与排查体系： + 智能告警：配置 LiteTopic 的消息堆积量阈值。一旦某会话链路延迟超过预期，立即触发告警。 + 快速定位：收到告警后，运维人员可直接在控制台 Group 详情页查看堆积量最高的 LiteTopic 列表及对应的消费者 IP。借助这种细粒度透视能力，原本大海捞针般的故障定位变成了分钟级的精准修复。 ▍2. RocketMQ LiteTopic 方案优势 基于 RocketMQ LiteTopic 的轻量化、灵活订阅、自动创建及 TTL 自动删除等原生特性，这套方案在架构层面主要具备以下三方面优势： 2.1 保障长耗时会话的极致连续性 借助 LiteTopic 的自动创建、“一会话一通道”和动态订阅机制，系统可以为每个语音会话建立独立的响应通道。无论 LLM 推理耗时多久，消息都能在专属通道中有序流转，避免多会话间的消息串扰。 同时，即使后端服务发生扩缩容或网络波动，响应消息仍能回到用户当前连接的网关节点，保证长链路交互中的 Session Sticky 和会话连续性。 此外，通过 LiteTopic 的异步通知机制，系统可以避免长耗时线程阻塞，进一步提升整体吞吐能力，让用户在高峰期也能获得流畅的语音交互体验。 2.2 推动应用架构进一步无状态化 在传统方案中，应用通常需要维护“Session ID 到 Node IP”的路由映射，以及配套的心跳保活和异常清理逻辑，状态管理复杂、维护成本高。 引入 LiteTopic 后，路由逻辑下沉到消息中间件层，业务代码只需围绕 SessionID 发送和接收消息。应用节点因此更接近无状态计算单元，不再强依赖本地连接状态表。这样不仅能够降低状态管理复杂度，也让应用更容易进行弹性伸缩和故障恢复，从而提升整体可维护性与容灾能力。 2.3 降低无效调用带来的模型成本 在传统架构中，消息错投或超时导致的“无响应”往往会触发客户端重试，导致同一段音频被重复发送给 LLM 推理，带来额外的 Token 消耗。 通过更精准的会话路由和可靠的投递机制，这套方案能够更好地保障“一次请求，必达响应”，显著减少因链路问题导致的重复调用，从而直接降低 LLM 的无效 Token 成本。 消息链路优化后带来的业务价值 从业务效果来看，引入 RocketMQ LiteTopic 之后，高并发智能语音交互链路通常可以在以下几个方面获得明显提升： + 用户体验更稳定：可以显著减少因连接状态不一致导致的“无响应”问题，提升语音交互成功率。即使在网络波动场景下，也能更好保障无感知重连，确保交互连续性。 + 系统复杂度更低：不再需要维护复杂的自定义路由表和状态同步逻辑，而是借助 LiteTopic 的原生能力完成会话管理，整体架构更简单，也更易扩展。 + 运维定位更高效：借助细粒度监控与告警，潜在性能瓶颈可以在影响用户前被发现和处理，问题定位与修复效率明显提升。 + 资源成本更可控：借助云消息队列 RocketMQ 版的弹性能力，业务可以按量付费，无需提前预留峰值容量，同时减少重复调用带来的额外模型消耗。 + 业务扩展更从容：更轻量、可扩展的链路设计，也为后续拓展更多实时互动场景打下基础，能够更从容地应对流量增长。 结语 很多团队在构建 Agent 时，往往会优先关注模型能力、推理效果和成本控制。但在高并发实时语音场景中，想把这些能力稳定地交付给用户，消息链路的稳定性、精准性和可扩展性同样不可忽视。 基于 RocketMQ LiteTopic 的这套方案，本质上解决的是几个关键问题： + 海量会话如何隔离。 + 长链路如何保持 Session Sticky。 + 异步结果如何精准回推。 + 临时通道如何自动管理。 + 系统如何在高并发下保持可观测与可扩展。 RocketMQ LiteTopic 提供了一种更适合高并发实时交互场景的消息架构思路。对于正在推进 Agent、实时互动、大模型应用工程化落地的团队来说，尤其是需要支撑海量动态会话、低延迟响应和灵活扩展的业务，这类能力正在从“加分项”逐步变成“必选项”。 欢迎钉钉搜索_（__群号：__110085036316）_或扫码加入 RocketMQ for AI 用户交流群，与我们交流探讨。

引言 随着大模型推理服务成为主流，消息队列在 AI 场景下的精细化流量治理，正面临前所未有的挑战。 传统互联网应用的业务流程固定、请求耗时短，消息队列的限流机制已相对成熟。然而，在 AI 推理场景下，业务流程高度动态、单次任务可持续数分钟甚至更久。这让传统方法显得力不从心，并引发两大核心痛点： + 队列头部阻塞：单个用户的慢任务，会阻塞队列中其他用户的消息处理。 + 并发效率受损：简单粗暴的限流措施，会导致整个系统吞吐量急剧下降。 为解决这些问题，Apache RocketMQ 5.x 版本推出了专为 AI 场景设计的核心特性——轻量主题模型 LiteTopic。它支持百万级轻量主题的创建和高性能动态订阅。基于 LiteTopic 的精细化流量治理方案，既能实现毫秒级的实时限流，又能支持分钟级的忙闲调度，真正做到了“千人千面”的个性化流量治理。 AI 推理场景下的消息队列新挑战 AI 应用与传统互联网应用存在本质差异，在于其执行模式和任务耗时。传统应用流程固定可预测、耗时短（秒级）、多为单向一次性交互；AI 应用更偏主动执行，会自主拆解目标并动态调整策略，流程不确定，单次任务耗时长（分钟级且不可预测），还常伴随多轮对话交互。 这种差异，导致消息队列在 AI 推理场景下面临两大严峻挑战： 1. 队列头部阻塞 传统业务中，不同用户的请求耗时较均衡（通常为秒级），即便多租户共享队列，也不会长期占用队列头部，阻塞问题不明显。因此，只需设置几个队列即可满足需求。 但在 AI 推理场景下，不同用户的请求耗时差异巨大（几秒到几十分钟不等且不可预测）。多租户共享队列时，一条长耗时消息（如复杂推理任务）占据队列头部，会阻塞后续所有消息的处理，导致同队列其他用户的正常消息无法被及时处理。若某个用户密集提交慢任务，可能长期抢占全部队列头部位置，形成资源独占，导致其他用户延迟飙升，破坏系统公平性。 2. 并发效率受损 在 AI 推理场景中，当某个用户短时间内密集提交大量推理请求时，系统需要对该用户实施流量控制。然而，传统的限流措施（如 Thread.sleep()）会阻塞消费者线程，这会导致一个严重的问题： 即使队列中还有其他健康用户的消息等待处理，但由于所有消费线程都在处理限流用户的请求而被阻塞，这些健康用户的正常消息也无法得到处理。随着被限流的用户增多，大量线程陷入阻塞状态，整个系统的并发处理能力将急剧下降。 传统方案为何在 AI 推理场景中失效？ 面对 AI 推理场景的流量洪峰，业界通常采用两种“老套路”来限流，但都“治标不治本”。 ▍方案一：消费失败重试法 简单粗暴地让消息失败，并自动重回队列排队。这听起来似乎很取巧，实则埋下了“定时炸弹”： + 重试机制不可控：依赖中间件内置重试机制，缺乏时间精度控制，易造成延迟放大； + 服务质量不稳定：无法保证时效性，消息可能在队列里躺上好几轮才被处理，影响业务 SLA； + 资源浪费严重：失败重试会消耗额外的网络、磁盘和 CPU 资源，增加系统整体负载，降低系统稳定性。 ▍方案二：线程阻塞限流法 当检测到某个用户短时间内请求频率过高或资源消耗过大时，通过Thread.sleep()等同步阻塞 API 暂停消息处理线程，直接让处理线程“睡一会儿”。这看似控制住了消息处理频率，实则是在“饮鸩止渴”： + 资源利用率低：大量线程被无效阻塞，不仅占用内存，还增加调度开销，导致并发能力下降，长期运行有资源耗尽风险； + 租户隔离失效：在共享线程池中，对某个队列的限流会波及由同一线程处理的其他队列，从而破坏多用户间的隔离性； + 吞吐量受损：阻塞机制与高性能设计的初衷背道而驰，严重损害了系统整体的消息处理能力。 这两种传统方法，要么过度依赖中间件机制，要么牺牲系统性能，都无法从根本上解决多租户环境下的精细化流量控制难题。 RocketMQ LiteTopic 流量治理：千人千面，优雅调度 ▍1. 毫秒级实时限流：让每个用户都有“专属 VIP 通道” AI 推理请求可能在毫秒级内剧烈波动，需要毫秒级的精细化限流能力来应对瞬时流量洪峰。 RocketMQ 基于 LiteTopic 打造了一套精细化限流方案，通过构建完整的资源隔离与调度体系来实现高效的流量治理： + 物理隔离：为每个用户/会话创建独立 LiteTopic，从物理层面实现用户级资源隔离，彻底消除交叉干扰。 + 弹性扩容：LiteTopic 支持百万级规模的按需创建，无论是小批量测试还是大规模生产，都能从容应对。 + 精准流控：每个 LiteTopic 可独立执行限流策略，支持按用户配置差异化阈值，真正实现“千人千面”的个性化流量治理。 + 消费挂起：当检测到用户请求超限时，不是简单地拒绝（失败重试）或等待（阻塞线程），而是优雅地“请用户稍等片刻”（挂起），既保护了系统资源，又不影响用户体验。 在实际应用中，流量处理流程如下图所示： 1. 消息分流：上游业务消息根据用户标识（如 userId）分流到每个独立用户对应的专属 LiteTopic，实现物理隔离。 2. 并行拉取：消费者通过长轮询并行拉取各 LiteTopic 的消息，在限流窗口中对每个 LiteTopic 独立执行限流判断。 3. 限流判断： 未超限：当某用户请求未触发阈值时，正常消费并输出流量； 已超限：当检测到请求超限时，返回 Suspend 挂起状态。 4. 消费挂起：该 LiteTopic 立即挂起，消费者释放处理线程并暂停服务端对该用户的拉取，支持毫秒级精确控制挂起时长，确保限流策略的灵活性和响应速度。 5. 线程复用：释放的线程即时转交其他用户请求，实现资源的弹性调度与高效复用。 6. 自动恢复：挂起的 LiteTopic 将在指定时间后自动恢复消费。 以下消费代码示例展示了如何在实际业务中实现这套机制： ```plain LitePushConsumer litePushConsumer = PROVIDER.newLitePushConsumerBuilder() .setClientConfiguration(clientConfiguration) .bindTopic(TOPIC) .setConsumerGroup(GROUP) .setMessageListener(messageView { //【物理隔离】以userId作为liteTopic名称，实现用户级物理隔离 // 每个用户独享一个独立的物理队列，确保资源完全独立，避免相互干扰 String userId = messageView.getLiteTopic(); //【精准流控】根据业务规则判断是否需要触发限流 // 支持按用户配置差异化阈值，实现"千人千面"的个性化流量治理 if (shouldThrottle(userId)) { //【消费挂起】返回suspend，立即释放当前处理线程 // 服务端暂停对该用户的拉取，避免无效资源消耗 // 支持毫秒级精确控制，100ms后自动重投递，释放的线程可被重新分配给其他用户请求 return ConsumeResultSuspend.of(Duration.ofMillis(100)); } // 正常处理消息 processMessage(messageView); return ConsumeResult.SUCCESS; }) .build(); ``` 上述代码的核心是引入了“消费挂起”机制。 与传统消息队列仅支持“消费成功”与“消费失败”两种状态不同，这里新增了第三种消费状态——Suspend，实现了精准的时间窗口控制： + 状态扩展：消费者返回 ConsumeResultSuspend 状态时，可携带下次可见时间戳，指定消息在时间窗口内的不可见期； + 资源释放：系统立即释放处理线程，清理该队列的本地缓存，避免资源占用； + 自动恢复：服务端维护定时调度器，到达指定时间后自动唤醒队列，重新参与拉取消费。 这一机制让瞬时限流不再阻塞线程，既保护了系统资源，又确保了其他用户请求的正常处理，完美契合 AI 推理场景下的实时流量治理需求。 ▍2. 分钟级忙闲调度：让延迟任务“错峰出行” 除了毫秒级的瞬时流量控制，RocketMQ LiteTopic 的消费挂起机制同样适用于分钟级甚至小时级的长时间窗口调度，实现延迟不敏感任务的错峰调度。 在实际业务场景中，可能存在大量延迟不敏感的任务，如： + 跑批任务：数据统计、报表生成等批量处理作业； + 异步处理：非核心链路的异步通知、日志分析等； + 资源消耗型任务：模型训练、离线推理等计算密集型操作。 这类任务无需实时处理，但可能占用大量计算资源。通过消费挂起机制，我们可以将这些任务智能调度到业务空闲时段执行： 1. 长时间窗口挂起：设置秒级甚至分钟级的挂起时长（如 Duration.ofMinutes(30)），将任务延迟到低峰期处理； 2. 动态感知业务负载：实时监控系统负载，当检测到资源紧张时，主动挂起低优先级任务的消费； 3. 轻量级任务调度：在无需引入额外调度系统的情况下，通过消息队列本身实现任务的延迟执行和资源错峰，降低系统复杂度。 ```plain LitePushConsumer litePushConsumer = PROVIDER.newLitePushConsumerBuilder() .setClientConfiguration(clientConfiguration) .bindTopic(TOPIC) .setConsumerGroup(GROUP) .setMessageListener(messageView { String taskType = messageView.getUserProperty("taskType"); //【忙闲调度】识别延迟不敏感任务 if ("BATCH".equals(taskType) || "LOW_PRIORITY".equals(taskType)) { // 检测系统是否处于繁忙状态 if (isSystemBusy()) { //【长时间挂起】将任务延后到空闲时段处理 // 挂起30分钟后自动恢复，实现错峰调度 return ConsumeResultSuspend.of(Duration.ofMinutes(30)); } } // 正常处理消息 processMessage(messageView); return ConsumeResult.SUCCESS; }) .build(); ``` 这种忙闲调度能力，让 RocketMQ LiteTopic 在消息队列的基础上，扩展了延迟任务处理能力。无需引入额外的调度组件，即可在保障核心业务 SLA 的同时，最大化系统资源利用率。 RocketMQ LiteTopic 技术揭秘：如何实现百万级物理隔离？ LiteTopic 是 Apache RocketMQ 专为 AI 场景设计的轻量主题模型，具备轻量资源、自动化生命周期管理、高性能订阅和顺序性保障等特点。 其底层基于创新的存储架构和分发机制，支撑了百万级 LiteTopic 的高效管理，在不牺牲性能的前提下，实现了海量 LiteTopic 资源的物理隔离，为 AI 场景下的精细化流量治理提供了坚实的技术基础。 关键技术点包括： + 统一存储、多路分发：所有消息数据统一存储在底层 CommitLog 文件中且仅存储一份，采用追加写入模式避免磁盘碎片化，保障极致写入性能。同时，通过多路分发机制为不同 LiteTopic 生成独立的消费索引。 + RocksDB KV 存储引擎：摒弃传统文件型 CQ 结构，替换为高性能的 KV 存储引擎 RocksDB，将队列索引信息和消息物理偏移量作为键值对存储，充分发挥 RocksDB 顺序写入的高性能优势，实现对百万级元数据的高效管理。 + 订阅关系管理：Broker 负责管理消费者的订阅关系集，支持增量更新，能够实时、主动地感知消息与订阅的匹配状态。 + 事件驱动与就绪集维护：每当新消息写入时立即触发订阅匹配，将符合条件的消息聚合到就绪集中。 + 高效批量拉取：消费者只需一次 poll 请求即可批量拉取来自多个 LiteTopic 的消息，显著降低网络交互频率，确保在海量订阅场景下的低延迟与高吞吐。 _百万级 LiteTopic 高并发性能的发送和消费流程_ 结语 随着 AI 推理日益普及，传统消息队列限流方式已难以满足精细化流量控制需求。 基于 RocketMQ LiteTopic 的精细化流量治理方案，通过物理隔离、弹性扩容、精准流控和消费挂起四大核心特性，系统性解决了队列头部阻塞和并发效率受损两大痛点，为 AI 推理场景提供了从毫秒级实时限流到分钟级忙闲调度的全方位消息处理保障，实现了真正意义上的“千人千面”个性化流量治理。 值得一提的是，该方案已与阿里云大模型服务平台百炼网关达成深度合作，利用 RocketMQ LiteTopic 的精细化流控能力，帮助其更好地管理 AI 推理请求的流量峰值与资源调度。 目前，LiteTopic 的核心能力已在阿里云云消息队列 RocketMQ 版 5.x 系列实例中发布，若要在实际业务中使用，请点击下方阅读原文链接查看帮助文档。 未来，我们将继续探索更多创新技术，推动消息队列在 AI 时代的演进与发展。 欢迎钉钉搜索_（群号：110085036316）_或扫码加入 RocketMQ for AI 用户交流群，与我们交流探讨～

作者简介：艾阳坤，Apache RocketMQ PMC Member/Committer，CNCF OpenTelemetry Member，CNCF Envoy contributor。 在分布式系统中，多个服务之间的交互涉及到复杂的网络通信和数据传输，其中每个服务可能由不同的团队或组织负责维护和开发。因此，在这样的环境下，当一个请求被发出并经过多个服务的处理后，如果出现了问题或错误，很难快速定位到根因。分布式全链路追踪技术则可以帮助我们解决这个问题，它能够跟踪和记录请求在系统中的传输过程，并提供详细的性能和日志信息，使得开发人员能够快速诊断和定位问题。对于分布式系统的可靠性、性能和可维护性起到了非常重要的作用。 RocketMQ 5.0 与分布式全链路追踪 Apache RocketMQ 5.0 版本作为近几年来最大的一次迭代，在整个可观测性上也进行了诸多改进。其中，支持标准化的分布式全链路追踪就是一个重要的特性。 RocketMQ 5.0 可观测 而由 Google、Microsoft、Uber 和 LightStep 联合发起的 CNCF OpenTelemetry 作为 OpenTracing 和 OpenCensus 的官方继任者，已经成为可观测领域的事实标准，RocketMQ 的分布式全链路追踪也围绕 OpenTelemetry 进行展开。 分布式链路追踪系统的起源可以追溯到 2007 年 Google 发布的论文。这篇论文详细介绍了 Google 内部使用的链路追踪系统 Dapper，其中使用的 span 概念被广泛采用，并成为后来开源链路追踪系统中的基础概念之一。 Dapper Trace Tree 在 Dapper 中，每个请求或事务被追踪时都会创建一个 span，记录整个请求或事务处理过程中的各个组件和操作的时间和状态信息。这些 span 可以嵌套，形成一个树形结构，用于表示整个请求或事务处理过程中各个组件之间的依赖关系和调用关系。后来，很多开源链路追踪系统，如 Zipkin 和 OpenTracing，也采用了类似的 span 概念来描述分布式系统中的链路追踪信息。现在，合并了 OpenTracing 和 OpenCensus 的 CNCF OpenTelemetry 自然也一样采用了 span 概念，并在此基础上进行了进一步发展。 OpenTelemetry 为 messaging 相关的 span 定义了，旨在制定一套与特定消息系统无关的 specification，而 OpenTelmetry 自身的开发其实也都是由 specification 驱动进行展开。 Specification Driven Development Messaging Span 定义 Specifaition 中描述了 messaging span 的拓扑关系，包括代表消息发送、接收和处理的不同 span 之间的父子和链接关系。关于具体的定义可以参考：。对应到 RocketMQ 中，有三种不同的 span： | Span | Description | | | | | send | 消息的发送过程。span 以一次发送行为开始，成功或者失败/抛异常结束。消息发送的内部重试会被记录成多条 span。 | | receive | 消费者中接收消息的长轮询过程，与长轮询的生命周期保持一致。 | | process | 对应 PushConsumer 里 MessageListener 中对消息的处理过程，span 以进入 MessageListener 为开始，离开 MessageListener 为结束。 | 特别地，默认情况下，receive span 是不启用的。在 receive span 启用和不启用的两种情况下，span 之间的组织关系是不同的： 启用 receive span 前后的 span 关系 在没有启用 receive span 的情况下，process span 会作为 send span 的 child；而当 receive span 启用的情况下，process span 会作为 receive span 的 child，同时 link 到 send span。 Messaging Attributes 定义 语义约定中规定了随 span 携带的通用属性的统一名称，这包括但不限于： messaging.message.id: 消息的唯一标识符。 messaging.destination：消息发送的目的地，通常是一个队列或主题名称。 messaging.operation：对消息的操作类型，例如发送、接收、确认等。 具体可以查看 。 特别地，不同的消息系统可能会有自己特定的行为和属性，，这包括： | Attribute | Type | Description | | | | | | messaging.rocketmq.namespace | string | RocketMQ 资源命名空间，暂未启用 | | messaging.rocketmq.client_group | string | RocketMQ producer/consumer 负载均衡组，5.0 只对 consumer 生效 | | messaging.rocketmq.client_id | string | 客户端唯一标识符 | | messaging.rocketmq.message.delivery_timestamp | int | 定时消息定时时间，只对 5.0 生效 | | messaging.rocketmq.message.delay_time_level | int | 定时消息定时级别，只对 4.0 生效 | | messaging.rocketmq.message.group | string | 顺序消息分组，只对 5.0 生效 | | messaging.rocketmq.message.type | string | 消息类型，可能为 normal/fifo/delay/transaction，只对 5.0 生效 | | messaging.rocketmq.message.tag | string | 消息 tag | | messaging.rocketmq.message.keys | string[] | 消息 keys，可以有多个 | | messaging.rocketmq.consumption_model | string | 消息消费模型，可能为 clustering/broadcasting，5.0 broadcasting 被废弃 | 快速开始 在 OpenTelemetry 中有两种不同的方式可以为应用程序添加可观测信息： Automatic Instrumentation：无需编写任何代码，只需进行简单的配置即可自动生成可观测信息，包括应用程序中使用的类库和框架，这样可以更方便地获取基本的性能和行为数据。 Manual Instrumentation：需要编写代码来创建和管理可观测数据，并通过 exporter 导出到指定的目标。这样可以更灵活自由地控制用户想要观测的逻辑和功能。 在 Java 类库中，前者是一种更为常见的使用形式。RocketMQ 5.0 客户端的 trace 也依托于 automatic instrumentation 进行实现。在 Java 程序中，automatic instrumentation 的表现形式为挂载 Java agent。在过去的一年里，我们将 推入了 OpenTelemetry 官方社区。现在，只需要在 Java 程序运行时挂载上 OpenTelemetry agent，即可实现对应用程序透明的分布式全链路追踪。 除此之外，Automatic Instrumentation 和 Manual Instrumentation 并不冲突，Automatic Instrumentation 中所使用的关键对象会被注册成全局对象，在 Manual Instrumentation 的使用方式中也可以非常方便的获取。实现两个 Instrumentation 共用一套配置，非常灵活和方便。 首先准备好 RocketMQ 5.0 Java 客户端，可以参考 进行消息的收发。关于 RocketMQ 5.0 的更多细节，欢迎大家参考和关注 和 。 然后准备好 OpenTelemetry agent jar，可以从 OpenTelemetry 官方，在应用程序启动时增加 javaagent:yourpath/opentelemetryjavaagent.jar 即可。可以通过设置 OTEL_EXPORTER_OTLP_ENDPOINT 环境变量来设置 OpenTelemetry collector 的接入点。 默认情况下，按照 OpenTelemetry 中关于 messaging 的规范，只有 send 和 process 的 span 会被启用，receive 的 span 是默认不启用的，如果想要启用 receive span，需要手动设置 Dotel.instrumentation.messaging.experimental.receivetelemetry.enabled=true。 场景最佳实践 目前，主流的云服务供应商都为 OpenTelemetry 提供了良好的支持，阿里云上的 SLS 和 ARMS 两款可观测产品都提供了基于 OpenTelemetry 的分布式全链路追踪服务。 为了更好地展示分布式全链路追踪的过程，这里提供了一个代码示例： 。在这个代码示例中，会启动三个不同的进程，涉及三种不同类库和业务逻辑之间的相互调用，展示了一个在分布式环境较复杂中间件之间进行交互的典型案例。 请求首先会从 gRPC 客户端发往 gRPC 服务端，在 gRPC 服务端收到请求之后，会向 RocketMQ 5.0 的 Producer 往服务端发送一条消息，然后再回复对应的 response 给客户端。在 RocketMQ 5.0 的 PushConsumer 接受到消息之后，会在 MessageListener 中使用 Apache HttpClient 往淘宝网发送一条 GET 请求。 示例代码调用链路 特别地，gRPC 客户端在发起具体的调用是在一个上游业务 span 的生命周期之内进行的，这个 span 我们称之为 ExampleUpstreamSpan，RocketMQ 5.0 PushConsumer 在收到消息之后，也会在 MessageListener 里执行其他的业务操作，也会有对应的 span，我们称之为 ExampleDownstreamSpan。那么默认在 receive span 没有启用的情况下，按照开始时间的顺序，会先后存在 7 个 span。分别是： ExampleUpstreamSpan。 gRPC 客户端请求 span。 gRPC 服务端响应 span。 RocketMQ 5.0 Producer 的 send span。 RocketMQ 5.0 Producer 的 process span。 HTTP 请求 span。 ExampleDownstreamSpan。 RocketMQ 5.0 对接 SLS Trace 服务 首先在阿里云日志服务中创建 Trace 服务。然后获取接入点，项目和实例名称等信息，具体可以参考。 在补充好信息之后完成接入之后，稍等一会就可以看到对应的 Trace 信息已经被上传到 SLS trace 服务中： SLS Trace 服务分布式全链路展示 Trace 服务其实是将相关数据存储到日志中，因此这些数据也可以通过 SLS 的 SQL 语法查询得到。 通过 Trace 数据，我们可以很方便知道用户的操作系统环境，Java 版本等一系列基础信息。消息的发送延时，失败与否，消息是否准时投递到了客户端，以及客户端本地消费耗时，消费失败与否等一系列有效信息，可以帮助我们十分有效地进行问题排查。 除此之外，SLS Trace 服务的 demo 页也提供了基于 RocketMQ 5.0 定制的消息中间件大盘，生动展示了利用 Trace 数据得到的发送成功率，端到端延时等一系列指标。 ：展示利用 Trace 数据得到的包括发送延时、发送成功率、消费成功率、端到端延时在内的一系列指标。 ：可以根据上一步得到的差错长 message id 进行进一步的细粒度查询。 消息中间件分析 RocketMQ 5.0 对接应用实时监控服务（ARMS） 首先进入应用实时监控服务 ARMS 控制台，点击接入中心中的 OpenTelemetry，选择 java 应用程序下的自动探测，获取启动参数并修改至自己的 java 应用程序，具体可以参考。 配置好参数之后，启动自己的相关应用程序，稍等一会儿，就可以在 ARMS Trace Explorer 里看到对应的数据了。 Trace Explorer 还可以查看 span 之间的时序关系。 ARMS Trace Explorer 分布式全链路追踪展示 具体地，可以点进每个 span 查看详细的 attributes/resources/events 等信息。除此之外，ARMS 还支持通过使用 OpenTelemetry Collector 转发的形式来收集应用程序的 Trace 数据。 趋势与思考 随着现代应用程序架构的不断演进，可观测性的重要性日益凸显。它不仅可以帮助我们快速发现和解决系统中的问题，还提高应用程序的可靠性和性能，同时也是实现 DevOps 的关键部分。在相关领域，也陆续诞生了像 DataDog 和 Dynatrace 这样的明星公司。 近年来涌现了一些新兴技术，如 eBPF（Extended Berkeley Packet Filter）和 Service Mesh 也为可观测领域提供了一些新的思路： eBPF 可以在内核层面运行，通过动态注入代码来监控系统的行为。它被广泛应用于实时网络和系统性能监控、安全审计和调试等任务，并且性能影响很小，未来也可以作为 continuous profiling 的一种选择。Service Mesh 则通过在应用程序之间注入代理层实现流量管理、安全和可观测性等功能。代理层可以收集和报告有关流量的各种指标和元数据，从而帮助我们了解系统中各个组件的行为和性能。 Service Mesh 中反映出的技术趋势很大一部分已经在 RocketMQ 5.0 proxy 中得到了应用，我们也在更多地将可观测指标往 proxy 进行收敛。当前的 Trace 链路未来也在考虑和服务端一起进行关联，并打造用户侧，运维侧，跨多应用的全方位链路追踪体系。除此之外还可以将 Trace 数据与 Metrics 数据通过 Exemplars 等技术进行联动。实现面到线，线到点的终极排查效果。 在可观测领域，RocketMQ 也在不断探索和摸索更加领先的可观测手段，以帮助开发者和客户更快更省心地发现系统中的隐患。 特别感谢阿里云 SLS 团队的千乘同学和 ARMS 团队的垆皓同学在接入过程提供的帮助和支持！ 相关链接 RocketMQ 5.0 客户端： OpenTelemetry Instrumentation for RocketMQ 5.0： RocketMQ OpenTelemetry 示例：

消息典型应用场景 阿里云拥有丰富的消息产品家族，除了 RocketMQ 以外，还有大家熟知的对标开源的云 Kafka、支持 AMQP 协议的开源消息队列 RabbitMQ、物联网通信网关 MQTT、 对标 AWS SQS/SNS 的 Serverless 版消息 MNS（现在也是轻量版 RocketMQ）以及云上事件总线、事件中心 EeventBridge 。 阿里云所有消息产品均采用 RocketMQ 作为底层存储引擎，为用户提供稳定、可靠、高性能的消息通信能力，比如百万 TPS、百万队列、毫秒级通信延迟、分级存储、 Serverless 弹性等众多的消息产品。也带来了丰富的应用场景，分为应用集成和数据集成两大类。 应用集成以 RocketMQ 为主，应用最为广泛，本文也将分享 RocketMQ 在微服务解耦、电商交易、金融支付等场景下的最佳实践，比如银行的交易流水、保单的支付流转等。RabbitMQ 、MQTT 也主要用于应用集成场景，比如物联网、 IoT 双向通信、云产品的事件通知以及后处理等。对于新建的业务场景，一般首推 RocketMQ 作为消息选型，因为 RocketMQ 拥有最丰富的功能特性；而对于存量的业务迁移，则可以根据具体使用的消息产品来进行选择，以降低迁移成本。 数据集成以云 Kafka 为主，在大数据分析、日志采集分析等场景下应用最为广泛，比如游戏的玩家操作、广告埋点、数据分析、应用数据监控等。各种 SaaS 类的集成、聚石塔、电商数据打通等场景，则主要使用 EventBridge。更多的产品选型对比，可以参考专题页中的消息队列产品选型。 业务消息使用场景 RocketMQ 经过阿里集团内部十年锤炼，经过双 11 大促等万亿级规模的实时场景验证，SLA 最高可支持 4 个9，数据可靠性支持 9 个9。 以微服务解耦、订单交易、金融支付等场景为例，在互联网、电商、零售、金融等行业，消息的使用量非常大。尤其是在秒杀大促时，为了保障系统的稳定运行，需要 RocketMQ 进行削峰填谷。另外金融客户对每笔交易、每个订单也都要求数据不能丢失。因此在此类场景普遍对消息的可靠传输、海量并发、低延迟、稳定性与容灾等有着非常高的要求。RocketMQ 提供了丰富的消息类型，比如事务消息、定时消息、顺序消息等。 在交易系统里，为了简化交易流程，一般使用事务消息和定时消息。同时 RocketMQ 也提供了消息轨迹查询、消息 dashboard ，可以非常方便地对每个消息进行回溯，对每个 topic 或者 group 进行监测。RocketMQ 5.0 也提供了丰富的实例规格，从百级别 QPS 到百万级 QPS ，可以覆盖大部分应用场景。RocketMQ 默认提供多副本、多可用区部署，也提供了跨地域消息路由能力，支持客户构建高可用容灾或多活，且 RocketMQ 能够支持 99.9%的消息 RT 在 10ms 传输。 RocketMQ 事务消息举例 实现订单状态机异步流转 以订单状态机异步流转为例。此前，如果收到一笔订单交易，需要逐个通知下游的服务模块，比如需要更新购物车、更新积分等。每个业务模块耦合在一起会导致大促时的流量峰值非常大，需要每个服务模块保障其处理性能。而基于 RocketMQ 的事务消息能力，即可轻松实现订单子流程系统的异步解耦和流量的削峰填谷，事务消息可以确保数据库订单状态持久化和下游通知的事务性。 收到 LBL 订单交易时，可以先向 RocketMQ 发送一条半事务消息，此时 RocketMQ 会 hold 住消息，等核心交易事务完成后再向 MQ 提交确认半事务消息的状态，并执行下游服务模块的通知。假设核心交易模块失败，则会废弃之前提交的半事务消息，不通知下游。 对比此前的传统事务模块，使用 RocketMQ 可以大幅简化交易链路，缩短下单耗时。尤其是在大促场景下，可以解耦下游的服务模块，提供削峰填谷的能力。 超时中心 RocketMQ 的定时消息场景也是常见的使用方式。 比如双 11 等大促场景存在大量预售订单、定点尾款等，会带来大量定时任务。在电商交易过程中，订单流转也存在多个超时状态的任务，处理超时状态的任务需要确保可靠及时。以传统的方案进行构建分布式调度机制实现的时候，比如基于定时器调度延迟大,可能会存在性能瓶颈。 而采用 RocketMQ 的定时消息，实现将变得非常简单。定时任务只需提交一条延迟消息到 RocketMQ ，由 RocketMQ 保障定时消息达到秒级的精度，最高可支持百万级别的 TPS 能力，同时也能支持消息的消费重试，保障任务可靠触发，相比传统的使用方式大大简化了定时的复杂度。 RocketMQ 灰度策略举例 微服务全链路灰度 微服务场景下，精准地控制灰度流量并进行灰度版本验证，是保障线上业务稳定运行的关键。大部分情况下，用户通过划分不同的环境来进行灰度发布，对应 RocketMQ 的不同实例。但是很多用户希望能够简化环境管理，尽可能复用线上资源，结合消息来提供微服务全链路灰度能力。 如上图所示，线上已经在运行的微服务模块游 A、B、C，C 模块会产生消息，并由 A 模块进行消费。此时对服务模块 A 和 C 做灰度发布，则线上会存在两条泳道，一条是正常的业务流量，一条是灰度链路。我们希望线上版本 C 模块生产的消息能够被线上版本 A 模块进行消费，灰度版本 C 模块生产的消息能够被灰度版本 A 模块进行消费。 RocketMQ 支持透传环境标签，可在生产端给消息属性添加标签，然后开启 RocketMQ 的 SQL 92 语法过滤，服务端即可完成消息的过滤和路由，从而降低客户端的压力。 本文讲述了 RocketMQ 的业务消息场景、一些功能特性的使用方法，包括事务消息、定时消息、消息全链路灰度等，欢迎大家尝试使用。

阿里云消息队列 RocketMQ 商业化历程 RocketMQ 诞生于 2012 年，诞生即开源。2012～2015 年，RocketMQ 一直在通过内部电商业务打磨自身服务能力,并在 2015 年于阿里云上线公测。2016 年，阿里云 RocketMQ 完成商业化，同时被捐赠给 Apache 基金会，同年获得了年度受欢迎中国开源软件荣誉。 在 Apache 孵化期间，Apache RocketMQ 经历了快速发展，2017 年即毕业成为了 Apache 顶级项目。同年，Apache RocketMQ TLP RocketMQ 4.0 正式发布。此后，RocketMQ 4.0 经历了长足发展，期间阿里云商业和开源相辅相成、齐头并进，直到今天，共同迈入 RocketMQ 5.0 时代。 RocketMQ 5.0 发布后，阿里云商业会持续采取 OpenCore 的发展模式，秉承上游优先的社区发展原则，与社区一起将 RocketMQ 打造为一个超融合的数据处理平台。 阿里云消息队列产品矩阵 阿里云基于 RocketMQ 消息底座，构建了多元化的消息产品系列。 RocketMQ 是阿里云主打的消息品牌，互联网新兴业务领域首选的数据通道。消息队列 Kafka 是大数据的首选数据通道，微消息队列 MQTT 是移动互联网和物联网的数据通道，消息队列 RocketMQ 是传统业务领域的数据通道。消息服务 MNS 是 RocketMQ 轻量版，主要应用于应用集成领域，为平台型应用提供简单的队列服务。事件总线 Event Bridge 定位为云上事件枢纽，旨在阿里云上构建统一的事件中心。 阿里云消息队列产品矩阵完全构建在 RocketMQ 之上，基本实现了应用场景全覆盖，包括微服务解耦、SaaS 集成、物联网、大数据或日志收集生态，同时也在内部覆盖了阿里巴巴所有业务，在云上为数万阿里云企业提供了优质的消息服务。阿里云的消息产品矩阵涵盖了互联网、大数据、移动互联网等领域业务场景，为云原生客户提供不可或缺的一站式解决方案。 RocketMQ 在阿里云商业化历程中，一直致力于探索业务消息实践，也孵化了大量业务消息特性，并持续反哺到开源社区。 RocketMQ 4.0 业务消息探索之路 RocketMQ 在商业化过程中，陆续推出了四种消息类型来满足丰富的业务场景。 普通消息：普通消息提供极致弹性、海量堆积能力，内置重试与死信队列来满足业务对失败重试的需求，同时具备高吞吐、高可用、低延迟等特性，广泛应用于应用集成、异步解耦、削峰填谷等场景。 定时消息：提供秒级定时精度， 40 天超长定时，主要面向分布式定时调度、任务超时处理等场景，目前正在开源中。  顺序消息：支持全局与局部严格有序，从发送、存储到消费，保证端到端有序。面向有序事件处理、撮合交易、数据实时增量同步等场景。 事务消息：分布式、高性能、高可用的最终一致性事务解决方案，广泛应用于电商交易系统中服务的一致性协调场景并且已经开源。  RocketMQ 4.0 期间，商业和开源都致力于全方位拓展消息接入能力，使 RocketMQ 能够非常轻松地连接应用开源和云产品生态。比如商业上提供了多语言 SDK ，开源也有相应的 SDK 能够覆盖 Java、Go、Python 、C++使用 RocketMQ。同时支持 Spring 生态，能够通过 Spring Cloud 的方式使用 RocketMQ。商业上提供了一组非常简单易用的 HTTP API，提供了 67 种语言的实现。 除了 SDK 接入，RocketMQ 也在积极拥抱社区标准，在云产品侧提供了 AMQP 和 MQTT 的接入能力，其中 MQTT 已开源。 RocketMQ 也大力在发展 connector 生态，能够通过 RocketMQ connector 接入很多数据源，包括 Redis、MongoDB、Hudi 等大数据系统。 另外，阿里云构建的事件总线 EventBridge 也已开源，通过该产品能够将阿里云的云产品、SaaS 应用、自建数据平台的数据引入 RocketMQ。 RocketMQ 4.0 版本做了大量尝试，提供了全方位的消息接入能力。 RocketMQ 在服务阿里集团用户和商业化历程中，沉淀了大量领先的业务消息处理与服务能力。比如消息订阅方面，RocketMQ 支持集群分布式消费能力，也支持广播消费。在消息处理方面支持基于 Tag 和 SQL 做灵活过滤，其中基于 SQL 过滤是电商交易中非常重要的特性，能够支持在非常订阅比的情况下实现较低的投递比。 全球消息路由能力具备性能高、实时性强的特点。在云时代，数据中心天然分布在各个地域，各个地域之间还有 VPC 网络隔离。但是通过全球消息路由功能可以将地域与网络打通，能够满足更多业务场景。比如在阿里内部基于该能力实现了异地多活、异地容灾等企业级特性。 另外，全球消息路由具备非常高的易用性，提供了可视化任务管理界面，通过简单配置即可创建复制链路。 消息治理方面，RocketMQ 提供了访问控制、命名空间、实例限流、消息回放、重试消息、死信消息、堆积治理等能力。 服务能力方面，RocketMQ 经历了非常多沉淀，它在为交易链路服务了 12  年，参加了 10 年双 11，这也保证了 RocketMQ 能够在阿里云上提供非常高的可靠性。双 11 消息收发 TPS 峰值过亿，日消息收发总量超过 3 万亿。而即使在双十一万亿级数据洪峰下，消息也能做到 99.996% 毫秒级响应能力，消息发布平均响应时间不超过 3 毫秒，最大不超过 20 毫秒，真正实现了低延迟消息发布。 商业化初期，客户遇到最大难题是在分布式环境下如何完整地追踪异步消息链路。基于此背景，我们打造了可视化全生命周期消息轨迹追踪系统，能够提供丰富的消息查询、消息下载、定点重投、轨迹追踪能力，通过可观测系统帮助用户解决分布式环境中不可观测的问题。 如上图所示，一条消息从产生、发送至服务端存储到最终投递到消费者，整个发送和消费轨迹都有迹可循，包括投递给哪些消费者、哪些消费者在什么地方成功消费或者消费失败、何时进行重投，真正帮助客户解决了分布式观测难题。 除了功能特性，RocketMQ 在稳定性方面也做了很多建设。我们始终坚持，SLA 是云原生的根本，因此整个研发运维链路都有严格的稳定性保障措施： 架构开发：每个方案设计都会面向失败设计，代码开发阶段会有严格 Code Review 阶段，也会完整经历单元测试、集成测试、性能测试和容灾测试流程。 变更管理：有着非常严格的变更制度，要做到每个变更可灰度、可监控、可回滚、可降级。 稳定性防护：提供了限流、降级、容量评估、应急方案、大促保障等能力，会定期进行故障和预案演练，定期进行风险梳理。 体系化巡检：在云上有全方位的生产环境黑盒巡检。基于用户视角，会对全地域所有功能做全功能扫描，包含高达 50 多项检测项，任意项功能出问题都能立刻被监测到。在白盒巡检方面，会对 JVM 运行时指标、内核系统、集群指标进行巡检。 故障应急：有完整地故障应急流程，包括监控报警、故障发生、快速止血、排查根因、故障复盘。 RocketMQ 5.0 云原生架构升级之路 云原生时代，云上用户对云产品服务化程度、弹性能力、可控制性能力以及韧性都有了更高的要求。在此背景之下，我们对 RocketMQ 进行了云原生架构升级，这也是 RocketMQ 5.0 的诞生背景。 轻量级 SDK：基于云原生通信标准 gRPC 开发了一组轻量级 SDK，能够与当前富客户端优势互补。   无状态消息网关：在核心数据链路推出了无状态消息网关。通过搭建无状态服务节点Proxy，再通过 LB 进行服务暴露，将存储节点数据分离来独立负责核心消息存储和高可用。Proxy 与 Store 节点分离部署，独立弹性。  Leaderless 高可用架构：Store 节点身份完全对等，完全 Leaderless 化，去 ZK 和 HA 管控节点，能够做到非常高的可用性。同时相比传统的 Raft 一致性协议，该 Leaderless 架构能够做到副本数灵活选择，同步异步自动升降级，实现秒级故障转移。高可用架构目前已经完成开源并与 Dledger 进行了融合。  云原生基础设施：可观测验能力云原生化，OpenTelemetry 标准化。整体架构走向 Kubernetes 化，能够充分利用售卖区的资源弹性能力。 RocketMQ 4.0 推荐的接入方式主要是富客户端。富客户端提供了诸如客户端侧负载均衡、消息缓存、故障转移等一系列企业级特性。但在云原生时代，轻量级、高性能的客户端更容易被云原生技术栈所集成。 因此，RocketMQ 5.0 重磅推出了全新多语言轻量级 SDK，具有以下优势： 全新极简 API 设计：不可变 API，有完善的错误处理。多语言 SDK 保障 API 在 Native 层面对齐。同时引入了全新的 Simple Consumer，能够支持按消息模型进行消费，用户不再需要关心消息队列，只需要关注消息。  通信层采用 gRPC 协议：拥抱云原生通信标准，gRPC 能够使服务更易被集成。多语言 SDK 通信代码也可以通过 gRPC 快速生成，更 Native 。  轻量级实现：采用无状态消费模式，能够大幅降低客户端的实现复杂度。客户端更轻量，采用的应用也更容易被 Serverless化、Mesh 化。  云原生可观测性：客户端实现了 OpenTelemetry 标准，能够支持以 OpenTelemetry 形式导出 Metrics 与 Tracing。 RocketMQ 5.0 的另一个重大升级是引入了全新的无状态消费模型。该消费模型完全构建在原先的队列模型之上。队列模型是与存储模型一致的消费模型，消费者完全按照队列做负载均衡，也按照队列做消息拉取，非常适合批量高速拉取以及对单条消息状态不敏感的场景，比如流计算等。 RocketMQ 5.0 推出了 PoP 机制，巧妙地在队列模型之上构建了消息模型，实现了鱼与熊掌兼得。在此消息模型的设计上，业务可以只关心消息而无需关心队列，所有 API 都能够支持单条消息级别的消费、重试、修改不可见时间、删除。 在消息模型下，消息发送过来被存储后，即对消费者可见。消费者通过 Receive Message API 对消息进行消费后，消息进入定时不可见状态。消息超时过后又会重新处于可见状态，能被其他消费者继续消费。某消费者确认消息后，服务端会对该消息进行删除，随即不可见。 基于消息系模型的消费流程下，API 完全面向消息而不是面向队列。而当 PoP 机制遇见了无状态 Proxy，除了存储层，其他节点都是无状态的；客户端、连接和消费也是无状态的，可任意在 Proxy 节点上飘移，真正做到轻量级。 经过重构，RocketMQ 5.0 的可观测性也走向了云原生标准。 Metrics 侧： 指标涵盖丰富：设计了更丰富的指标，包含消息量、堆积量、各个阶段耗时等指标，每个指标从实例、Topic、消费 GroupID 多维度做聚合和展示。 消息团队实践模板：为用户提供实践模板，并持续迭代更新。 Prometheus + Grafana：Prometheus 标准数据格式，利用 Grafana 展示。除了模板，用户也可以自定义展示大盘。 Tracing 侧： OpenTelemetry Tracing 标准：RocketMQ Tracing 标准已经合并到 OpenTelemetry 开源标准，提供了规范和丰富的 messaging tracing 场景定义。 消息领域定制化展示：按照消息维度重新组织抽象的请求 span数据，展示一对多的消费，多次消费信息直观且方便理解。 可衔接 tracing 链路上下游：消息的 tracing 可继承调用上下文，补充到完整的调用链路中，消息链路信息串联了异步链路的上游和下游链路信息。 Logging 侧： Error Code 标准化：不同的错误有唯一的 Error Code。 Error Message 完整：包含完整的错误信息和排序所需要的资源信息。 Error Level 标准化：细化了各种不同错误信息的日志级别，用户可根据 Error、Warn 等级别配置更适合的监控告警。 弹性方面，RocketMQ 5.0 商业版能够充分撬动云的计算、存储和网络的池化资源。比如在计算方面，RocketMQ 5.0 所有工作负载完全部署在 ACK 之上，充分利用了 ACK 弹性能力，撬动 ACK 弹性资源。主要依赖 ACK 的两项技术，一是弹性资源池，另一个是 HPA 支持计算能力快速弹性。同时也会在 ACK 之上做跨可用区部署以提供高可用保障。 网络层面，RocketMQ 5.0 也会充分利用阿里云网络设施，为用户提供更便捷的网络访问能力。比如 RocketMQ 5.0 实例能够支持公网随开随用，需要依赖公网做测试的时候即开即用，测试完立即关闭，安全与方便兼具。同时支持多种私网类型的网络形态，包括 Single Tunnel、Private Link，另外也基于 CEN 构建了全球互通设计网络。 存储方面，RocketMQ 5.0 商业版率先引入多级存储概念，基于 OSS 构建二级存储，能够充分利用 OSS 存储的弹性能力，存储计费也转向了按量付费。而用户能够在 RocketMQ 之上自定义消息存储时长，比如将消息从 3 天有效时长延长至 30 天，能够真正将消息变为数据资产。同时利用二级存储能力，将冷热数据分离，为用户提供一致的冷读 SLA 。 RocketMQ 5.0 商业版发布预告 RocketMQ 4.0 历经了五年发展，开源和商业版本共同迈入了 5.0 时代。7 月底，阿里云消息队列将会基于开源版发布全新的 5.0 商业化版本。注：截止发稿前，RocketMQ 5.0 已经在阿里云消息队列 RocketMQ 产品上全新发布，目前支持国内主要地域。 RocketMQ 5.0 版相对于 4.0 版实例主要有以下几大改变： 第一，新版本、新售卖，更便宜。新版本采取了全新计量方式，有包年、包月型，也有按量付费和公网流量弹性计费。也有更全的售卖体系，比如新增专业版实例，能够满足部分用户需求。同时每个商品系列都新增了测试环境专用实例，能够方便用户以低成本的方式搭建自己的开发环境。 第二，更强弹性，降本提效利器。存储完全走向弹性，能够通过 Serverless 按需使用，按量付费。预留弹性，实例基础规格支持实时升降配，用户可以很方便地在流量到来之前做弹性。此外，专业版支持突发流量弹性，能够解决线上稳定性风险。 第三，全新架构，增强可观测运维。无状态消息消费模型能够解决一些老版本的痛点。同时在可观测上全面采取了云原生接入栈。 消息的全新形态：事件总线 EventBridge 事件总线 EventBridge 已经开源到 RocketMQ 社区中。云原生时代，事件无处不在，云计算资源散落在各地，各类生态孤岛随处可见。因此，以事件和事件驱动的方式来集成这一切是大势所趋。 基于此，阿里云推出了全新事件型产品 EventBridge。该产品构建在 RocketMQ 之上，是 RocketMQ 之上的一个事件驱动架构实践。 EventBridge 的事件源包括阿里云服务的管控事件比如资源变更事件、审计事件、配置变更事件，阿里云服务的数据事件，也包括自定义应用、SaaS 应用、自建数据平台、其他云厂商服务等。 事件经过 EventBridge 处理后会投递到事件目标，事件目标包括函数计算、消息服务、自建网关、HTTP(S)、短信、邮箱、钉钉等。 事件源到事件目标之间会经历完整的事件处理，包括事件源接入到 EB 后，可以对事件进行过滤、转换、归档、回放等。事件在 EventBridge 整个流程中也有完善的可观测性设计，包括事件查询、链路追踪。事件的接入方式非常丰富，可以通过 OpenAPI 来接入、7 种多语言 SDK、CloudEvents SDK、Web Console 和 Webhook 。 EventBridge 具有如下特点： 能够大幅度减少用户开发成本，用户无需额外开发，通过创建 EventBridge 源、事件目标、事件规则等资源即可实现事件架构。用户可以编写事件规则，对事件做过滤、转换。  提供原生 CloudEvents 支持，拥抱 CNCF 社区，能够无缝对接社区 SDK 。标准协议也能统一个阿里云事件规范。  事件 Schema 支持：能够支持事件 Schema 自动探测和校验，支持 Source 和 Target 的 Schema 绑定。  全球事件任意互通：组建了全球事件任意互通网络，组件了跨地域、跨账户的事件网络，能够支持跨云、跨数据中心的事件路由。 EventBridge在云上生态已经初具规模，已经集成了 255+ 云产品事件源和 1000+ 事件类型。 EventBridge率先对消息生态做了融合。阿里云的消息产品矩阵生态均通过 EventBridge 做了完全融合。任何一款消息产品与另一款消息产品的数据都能互通。同时，依靠 EventBridge 的全球事件网络，能够为所有消息产品赋予全球消息路由的能力。 EventBridge 目前已经在内部接入钉钉 ISV、聚石塔 ISV，外部也有 50+ SaaS 系统可以通过 Webhook 的方式接入。另外，海量事件源可以触达 10 多种事件目标，已经对接了全系云产品 API ，任何事件都可以驱动全量云产品 API。 作者介绍： 周新宇 Apache Member，Apache RocketMQ PMC Member，阿里云消息队列 RocketMQ 研发负责人。

2022年，RocketMQ5.0的正式版发布。相对于4.0版本而言，架构走向云原生化，并且覆盖了更多业务场景。 一、消息队列演进史 操作系统、数据库、中间件是基础软件的三驾马车，而消息队列属于最经典的中间件之一，已经有30多年的历史。消息队列的发展主要经历了以下几个阶段： 第一阶段（19802000年）：80年代诞生了第一款消息队列The Information Bus，第一次提出发布订阅模式来解决软件之间的通信问题；90年代是国际商业软件巨头的时代，IBM、Oracle、Microsoft纷纷推出自己的MQ，其中最具代表性的为IBM MQ，价格昂贵，面向高端企业，主要是大型金融、电信等企业。该类商业MQ一般采用高端硬件，软硬件一体机交付，MQ本身的软件架构为单机架构。 第二阶段（2000~2007年）：进入00年代后，初代开源消息队列崛起，诞生了JMS、AMQP两大标准，与之对应的两个实现分别为ActiveMQ、RabbitMQ，他们引领了初期的开源消息队列技术。开源极大促进了消息队列的流行，降低了使用门槛，技术普惠化，逐渐成为企业级架构的标配。相比于今天而言，这类MQ主要面向传统企业级应用和小流量场景，横向扩展能力较弱。 第三阶段（2007~2017年）：PC互联网、移动互联网爆发式发展。由于传统的消息队列无法承受亿级用户的访问流量与海量数据传输，诞生了互联网消息中间件，核心能力是全面采用分布式架构，具备很强的横向扩展能力，开源典型代表有Kafka、RocketMQ，闭源的有淘宝Notify。Kafka的诞生还将消息中间件从消息领域延伸到了流领域，从分布式应用的异步解耦场景延伸到大数据领域的流存储与流计算场景。 第四阶段（2014~至今）：云计算、IoT、大数据引领了新的浪潮。 二、互联网时代的RocketMQ 阿里的电商系统最初是个庞大的单体巨石应用，在研发效率、稳定性方面都无法满足淘宝和天猫飞速的发展。为了解决问题，2008年，淘宝与天猫发起了一次最大规模的架构升级，启动了“五彩石”项目，将单体应用拆分为分布式应用，同时抽象淘宝、天猫的共同底座——业务中台，包括交易中心、商品中心、买家中心等。在业务中台之下，同时诞生了阿里中间件（初期三大件包括消息、RPC、分布式数据层），RocketMQ是其中之一。 虽然在当时业界已经存在不少商业或开源的消息队列，比如IBMMQ、ActiveMQ、RabbitMQ，但无一例外，它们都诞生于传统企业级应用的场景，无法承受互联网对于高并发、无限扩展的苛刻要求。以RabbitMQ为例，RabbitMQ的队列流量与存储负载都为单机，无法满足业务横向扩展的需求。当时另一款具备无限横向扩展能力的消息队列是Kafka，但其主要用于日志类场景，未经过大规模核心业务稳定性验证，而且偏向于简单的log型消息队列，无法满足电商对于复杂消息功能特性的诉求，比如消息过滤、延迟消息等。 另一方面，传统的消息队列无法解决电商业务对于分布式一致性的要求。通过消息队列实现应用异步解耦后，电商业务还需要保障不同上下游应用对于订单状态要达成最终一致，否则会产生大量脏数据，造成业务错误。 大规模的电商系统，既要高性能又要一致性，传统的分布式事务技术束手无策。比如IBM MQ虽然可以使用XA事务来满足分布式一致性的功能诉求，但是XA带来的延迟与成本，对于海量的互联网流量难以承受。 为了解决电商业务对于消息队列的高性能、一致性、无限扩展等需求，自研消息队列成为了当时阿里唯一的出路，最终互联网消息队列RocketMQ应运而生。 为了支持超大规模的复杂电商业务，RocketMQ面向四个方面进行了重点建设，形成了四大优势能力。 ① 支撑超大规模复杂业务的能力，具备丰富的消息特性。 每一个大型互联网公司都会有主营业务（比如阿里是交易、蚂蚁是支付、饿了么是外卖），以主营业务为中心扩展业务能力，阿里电商是围绕交易事件建设的电商操作系统，每笔交易事件都会触发不同的业务，不同细分业务会关注不同类型的交易事件，比如垂直市场只关注某个类目的交易事件、天猫超市只关注某个卖家的交易事件、购物车只关注下单成功的交易事件等。 RocketMQ的SQL订阅提供灵活的消息过滤能力，能够满足下游消费者按照不同的业务维度进行消息过滤的诉求。 在大型互联网业务中，还会有各种定时事件触发场景，最典型的是交易超时关闭机制，阿里交易或者12306订票都有类似的机制。RocketMQ的定时消息能够很方便的满足这类诉求。 ② 一致性。 无论是阿里交易还是蚂蚁支付，都天然对数据一致性有着极高要求，RocketMQ在一致性方面也打造了多个关键特性。最具代表性的是分布式事务消息，RocketMQ是第一个实现该种特性的消息队列，能够保障交易的上下游对于订单状态达到最终一致。该方案也成为异步消息一致性方案的事实标准，被多个互联网公司所采纳，甚至也有公司将移植到定制版的Kafka种。除了分布式一致性之外，RocketMQ还提供了顺序消息的特性，满足顺序一致性的需求。 ③ 稳定性。 稳定性是交易与金融场景的基石特性，也是RocketMQ的根本。RocketMQ除了具备核心服务的HA之外，还具备了全局高可用能力，在阿里内部支持同城多活、异地多活、中心容灾等高阶HA能力。同时，稳定性也不局限于数据与服务的高可用，RocketMQ从产品层面对稳定性进行了全方位的建设，如消息轨迹、消息回溯、消息死信机制。 ④ 高性能。 在双十一的极限流量下，RocketMQ写消息延迟4个9在1ms内，100%在100ms内。RocketMQ采用sharednothing分布式架构，在吞吐量方面也具备无限扩展的能力，已经连续10年支持了双十一万亿级消息洪峰，为百万级的应用实例提供低延迟消息服务。互联网的故事还在进行，云计算规模化落地的时代悄然而来。 三、云计算时代的RocketMQ5.0 2015年，RocketMQ的首个云消息服务在阿里云上线，开启了大规模的云计算实践的序幕。同时RocketMQ也是业界第一个提供公有云服务的开源消息队列。 在大规模的云计算业务场景下，RocketMQ面临着全新的挑战与机遇。 多样性：它不再仅服务于某一家公司的内部业务，不再局限于互联网或金融企业，需要实现全行业、全场景的覆盖。 标准化：对于服务企业内部的自研消息队列而言，无需考虑协议或API的标准化。但是对于云消息服务而言，因为服务对象是外部企业客户，据信通院统计，80%以上的企业客户已经采纳开源技术和标准技术。因此，作为一款云消息服务，需要提供对业界的事实标准协议、接口、SDK的兼容，才能保证客户平滑上云，同时打消客户技术绑定的担忧。 云原生：云原生理念深入人心，消息队列要更好地帮助客户实现云原生应用架构，为业务降本提效。 新趋势：各种新技术的兴起，包括IoT、5G、边缘计算、事件驱动，还有事件流技术。面向技术的新趋势与多样化的业务需求，RocketMQ进行了自我进化，演进到5.0版本。 为了充分释放云的技术红利，RocketMQ5.0在技术架构上进行了云原生的演进。从客户端到服务端都进行了全方位的改造，更高弹性、可用性、更低成本。 客户端采用轻量SDK设计理念，将原来富客户端的逻辑下沉到Broker，满足现代化应用轻量化、Serverless的趋势。 Broker彻底进行弹性架构改造，分离RocketMQ Proxy与Store层，其中Proxy是完全无状态的计算节点，专注多协议、多领域场景覆盖，可以面向不同工作负载独立弹性，如物联网、微服务、大数据不同场景有不同的资源诉求。Store层则专注消息的高可用存储，包括副本复制、主备切换与云存储集成。同时对RocketMQ的Topic资源进行三层解耦，面向消息的Topic、面向流的Topic逻辑分片、面向底层存储的Topic物理分片，每一层都可以独立弹性。 在存储层引入了Leaderless的高可用架构，Store节点身份对等，Leaderless化，0外部依赖。多副本策略可定制，可用性+可靠性+成本灵活组合，面向多可用区、多region组建Geo高可用能力。 为了满足云时代多样化的用户需求，RocketMQ5.0从原来的互联网业务消息中间件扩展到"消息、事件、流"超融合处理平台，解锁更全面的能力。 在消息领域，全面拥抱云原生技术，更好的弹性架构与高可用能力。 在事件领域，支持CloudEvent规范，以事件为中心的产品新界面，助力客户建设跨业务、跨组织的数字化商业生态。 在流领域，流存储增强批量特性，大幅度提高数据吞吐量；新增逻辑队列能力，解耦逻辑资源与物理资源，在流场景也具备无缝伸缩能力；新增流数据库RSQLDB，提供实时事件流处理、流分析能力。 RocketMQ基于端云一体化架构实现了完整的物联网消息队列的能力，从原来的连接应用扩展到连接物联网设备。同时RocketMQ5.0也继续保持极简架构的原则，能够以最低的资源消耗、运维成本搭建服务，适合边缘计算。 除了的产品核心能力之外，RocketMQ5.0积极建设开源生态。 一方面是应用架构生态的建设，既有经典的开源项目、规范的集成，比如JMS、AMQP等，也有云原生技术生态的集成，比如CloudEvents、Dapr、Envoy。同时RocketMQ也会进一步发力数据架构生态，全链路集成大数据的摄入、数据存储、数据处理、数据分析组件，从离线大数据到实时大数据。 【活动】一键体验 RocketMQ 六大生产环境 免费试用+30秒一键体验，低门槛、快速、高效、易操作，带你了解“历经万亿级数据洪峰考验”的云消息队列RocketMQ！ 点击阅读原文，立即参与活动！

设计总览 RocketMQ 多级存储旨在不影响热数据读写的前提下将数据卸载到其他存储介质中，适用于两种场景： 1. 冷热数据分离：RocketMQ 新近产生的消息会缓存在 page cache 中，我们称之为热数据；当缓存超过了内存的容量就会有热数据被换出成为冷数据。如果有少许消费者尝试消费冷数据就会从硬盘中重新加载冷数据到 page cache，这会导致读写 IO 竞争并挤压 page cache 的空间。而将冷数据的读取链路切换为多级存储就可以避免这个问题； 2. 延长消息保留时间：将消息卸载到更大更便宜的存储介质中，可以用较低的成本实现更长的消息保存时间。同时多级存储支持为 topic 指定不同的消息保留时间，可以根据业务需要灵活配置消息 TTL。 RocketMQ 多级存储对比 Kafka 和 Pulsar 的实现最大的不同是我们使用准实时的方式上传消息，而不是等一个 CommitLog 写满后再上传，主要基于以下几点考虑： 1. 均摊成本：RocketMQ 多级存储需要将全局 CommitLog 转换为 topic 维度并重新构建消息索引，一次性处理整个 CommitLog 文件会带来性能毛刺； 2. 对小规格实例更友好：小规格实例往往配置较小的内存，这意味着热数据会更快换出成为冷数据，等待 CommitLog 写满再上传本身就有冷读风险。采取准实时上传的方式既能规避消息上传时的冷读风险，又能尽快使得冷数据可以从多级存储读取。 Quick Start 多级存储在设计上希望降低用户心智负担：用户无需变更客户端就能实现无感切换冷热数据读写链路，通过简单的修改服务端配置即可具备多级存储的能力，只需以下两步： 1. 修改 Broker 配置，指定使用 org.apache.rocketmq.tieredstore.TieredMessageStore 作为 messageStorePlugIn 2. 配置你想使用的储存介质，以卸载消息到其他硬盘为例：配置 tieredBackendServiceProvider 为 org.apache.rocketmq.tieredstore.provider.posix.PosixFileSegment，同时指定新储存的文件路径：tieredStoreFilepath 可选项：支持修改 tieredMetadataServiceProvider 切换元数据存储的实现，默认是基于 json 的文件存储 更多使用说明和配置项可以在 GitHub 上查看多级存储的 技术架构 architecture 接入层：TieredMessageStore/TieredDispatcher/TieredMessageFetcher 接入层实现 MessageStore 中的部分读写接口，并为他们增加了异步语意。TieredDispatcher 和 TieredMessageFetcher 分别实现了多级存储的上传/下载逻辑，相比于底层接口这里做了较多的性能优化：包括使用独立的线程池，避免慢 IO 阻塞访问热数据；使用预读缓存优化性能等。 容器层：TieredCommitLog/TieredConsumeQueue/TieredIndexFile/TieredFileQueue 容器层实现了和 DefaultMessageStore 类似的逻辑文件抽象，同样将文件划分为 CommitLog、ConsumeQueue、IndexFile，并且每种逻辑文件类型都通过 FileQueue 持有底层物理文件的引用。有所不同的是多级存储的 CommitLog 改为 queue 维度。 驱动层：TieredFileSegment 驱动层负责维护逻辑文件到物理文件的映射，通过实现 TieredStoreProvider 对接底层文件系统读写接口(Posix、S3、OSS、MinIO 等)。目前提供了 PosixFileSegment 的实现，可以将数据转移到其他硬盘或通过 fuse 挂载的对象存储上。 消息上传 RocketMQ 多级存储的消息上传是由 dispatch 机制触发的：初始化多级存储时会将 TieredDispatcher 注册为 CommitLog 的 dispacher。这样每当有消息发送到 Broker 会调用 TieredDispatcher 进行消息分发，TieredDispatcher 将该消息写入到 upload buffer 后立即返回成功。整个 dispatch 流程中不会有任何阻塞逻辑，确保不会影响本地 ConsumeQueue 的构建。 TieredDispatcher TieredDispatcher 写入 upload buffer 的内容仅为消息的引用，不会将消息的 body 读入内存。因为多级储存以 queue 维度构建 CommitLog，此时需要重新生成 commitLog offset 字段 upload buffer 触发 upload buffer 上传时读取到每条消息的 commitLog offset 字段时采用拼接的方式将新的 offset 嵌入到原消息中 上传进度控制 每个队列都会有两个关键位点控制上传进度： 1. dispatch offset：已经写入缓存但是未上传的消息位点 2. commit offset：已上传的消息位点 upload progress 类比消费者，dispatch offset 相当于拉取消息的位点，commit offset 相当于确认消费的位点。commit offset 到 dispatch offset 之间的部分相当于已拉取未消费的消息 消息读取 TieredMessageStore 实现了 MessageStore 中的消息读取相关接口，通过请求中的逻辑位点（queue offset）判断是否从多级存储中读取消息，根据配置（tieredStorageLevel）有四种策略： DISABLE：禁止从多级存储中读取消息； NOT_IN_DISK：不在 DefaultMessageStore 中的消息从多级存储中读取； NOT_IN_MEM：不在 page cache 中的消息即冷数据从多级存储读取； FORCE：强制所有消息从多级存储中读取，目前仅供测试使用。 ${a} 需要从多级存储中读取的消息会交由 TieredMessageFetcher 处理：首先校验参数是否合法，然后按照逻辑位点（queue offset）发起拉取请求。TieredConsumeQueue/TieredCommitLog 将逻辑位点换算为对应文件的物理位点从 TieredFileSegment 读取消息。 ${b} TieredFileSegment 维护每个储存在文件系统中的物理文件位点，并通过为不同存储介质实现的接口从中读取所需的数据。 ${c} 预读缓存 TieredMessageFetcher 读取消息时会预读一部分消息供下次使用，这些消息暂存在预读缓存中 ${d} 预读缓存的设计参考了 TCP Tahoe 拥塞控制算法，每次预读的消息量类似拥塞窗口采用加法增、乘法减的机制控制： 加法增：从最小窗口开始，每次增加等同于客户端 batchSize 的消息量。 乘法减：当缓存的消息超过了缓存过期时间仍未被全部拉取，在清理缓存的同时会将下次预读消息量减半。 预读缓存支持在读取消息量较大时分片并发请求，以取得更大带宽和更小的延迟。 某个 topic 消息的预读缓存由消费这个 topic 的所有 group 共享，缓存失效策略为： 1. 所有订阅这个 topic 的 group 都访问了缓存 2. 到达缓存过期时间 故障恢复 上文中我们介绍上传进度由 commit offset 和 dispatch offset 控制。多级存储会为每个 topic、queue、fileSegment 创建元数据并持久化这两种位点。当 Broker 重启后会从元数据中恢复，继续从 commit offset 开始上传消息，之前缓存的消息会重新上传并不会丢失。 开发计划 面向云原生的存储系统要最大化利用云上存储的价值，而对象存储正是云计算红利的体现。 RocketMQ 多级存储希望一方面利用对象存储低成本的优势延长消息存储时间、拓展数据的价值；另一方面利用其共享存储的特性在多副本架构中兼得成本和数据可靠性，以及未来向 Serverless 架构演进。 tag 过滤 多级存储拉取消息时没有计算消息的 tag 是否匹配，tag 过滤交给客户端处理。这样会带来额外的网络开销，计划后续在服务端增加 tag 过滤能力。 广播消费以及多个消费进度不同的消费者 预读缓存失效需要所有订阅这个 topic 的 group 都访问了缓存，这在多个 group 消费进度不一致的情况下很难触发，导致无用的消息在缓存中堆积。 需要计算出每个 group 的消费 qps 来估算某个 group 能否在缓存失效前用上缓存的消息。如果缓存的消息预期在失效前都不会被再次访问，那么它应该被立即过期。相应的对于广播消费，消息的过期策略应被优化为所有 Client 都读取这条消息后才失效。 和高可用架构的融合 目前主要面临以下三个问题： 1. 元数据同步：如何可靠的在多个节点间同步元数据，slave 晋升时如何校准和补全缺失的元数据； 2. 禁止上传超过 confirm offset 的消息：为了避免消息回退，上传的最大 offset 不能超过 confirm offset； 3. slave 晋升时快速启动多级存储：只有 master 节点具有写权限，在 slave 节点晋升后需要快速拉起多级存储断点续传。

高可用架构演进背景 在分布式系统中不可避免的会遇到网络故障，机器宕机，磁盘损坏等问题，为了向用户不中断且正确的提供服务，要求系统有一定的冗余与容错能力。RocketMQ 在日志，统计分析，在线交易，金融交易等丰富的生产场景中发挥着至关重要的作用，而不同环境对基础设施的成本与可靠性提出了不同的诉求。在 RocketMQ v4 版本中有两种主流高可用设计，分别是主备模式的无切换架构和基于 Raft 的多副本架构（图中左侧和右侧所示）。生产实践中我们发现，两副本的冷备模式下备节点资源利用率低，主宕机时特殊类型消息存在可用性问题；而 Raft 高度串行化，基于多数派的确认机制在扩展只读副本时不够灵活，无法很好的支持两机房对等部署，异地多中心等复杂场景。RocketMQ v5 版本融合了上述方案的优势，提出 DLedger Controller 作为管控节点（中间部分所示），将选举逻辑插件化并优化了数据复制的实现。 如何实现高可用系统 副本组与数据分片 在 PrimaryBackup 架构的分布式系统中，一份数据将被复制成多个副本来避免数据丢失。处理相同数据的一组节点被称为副本组（ReplicaSet），副本组的粒度可以是单个文件级别的（例如 HDFS），也可以是分区级 / 队列级的（例如 Kafka），每个真实存储节点上可以容纳若干个不同副本组的副本，也可以像 RocketMQ 一样粗粒度的独占节点。独占能够显著简化数据写入时确保持久化成功的复杂度，因为每个副本组上只有主副本会响应读写请求，备机一般配置只读来提供均衡读负载，选举这件事儿等价于让副本组内一个副本持有独占的写锁。 RocketMQ 为每个存储数据的 Broker 节点配置 ClusterName，BrokerName 标识来更好的进行资源管理。多个 BrokerName 相同的节点构成一个副本组。每个副本还拥有一个从 0 开始编号，不重复也不一定连续的 BrokerId 用来表示身份，编号为 0 的节点是这个副本组的 Leader / Primary / Master，故障时通过选举来重新对 Broker 编号标识新的身份。例如 BrokerId = {0, 1, 3}，则 0 为主，其他两个为备。 一个副本组内，节点间共享数据的方式有多种，资源的共享程度由低到高来说一般有 Shared Nothing，Shared Disk，Shared Memory，Shared EveryThing。典型的 Shared Nothing 架构是 TiDB 这类纯分布式的数据库，TiDB 在每个存储节点上使用基于 RocksDB 封装的 TiKV 进行数据存储，上层通过协议交互实现事务或者 MVCC。相比于传统的分库分表策略来说，TiKV 易用性和灵活程度很高，更容易解决数据热点与伸缩时数据打散的一系列问题，但实现跨多节点的事务就需要涉及到多次网络的通信。另一端 Shared EveryThing 的案例是 AWS 的 Aurora，Aliyun 的 PolarStore，旁路 Kernal 的方式使应用完全运行于用户态，以最大程度的存储复用来减少资源消耗，一主多备完全共用一份底层可靠的存储，实现一写多读，快速切换。 大多数 KV 操作都是通过关键字的一致性哈希来计算所分配的节点，当这个节点所在的主副本组产生存储抖动，主备切换，网络分区等情况下，这个分片所对应的所有键都无法更新，局部会有一些操作失败。消息系统的模型有所不同，流量大但跨副本组的数据交互极少，无序消息发送到预期分区失败时还可以向其他副本组（分片）写入，一个副本组的故障不影响全局，这在整体服务的层面上额外提供了跨副本组的可用性。此外，考虑到 MQ 作为 Paas 层产品，被广泛部署于 Windows，Linux on Arm 等各种环境，只有减少和 Iaas 层产品的深度绑定，才能提供更好的灵活性。这种局部故障隔离和轻依赖的特性是 RocketMQ 选则 Shared Nothing 模型重要原因。 副本组中，各个节点处理的速度不同，也就有了日志水位的概念。Master 和与其差距不大的 Slave 共同组成了同步副本集（SyncStateSet）。如何定义差距不大呢？衡量的指标可以是日志水位（文件大小）差距较小，也可以是备落后的时间在一定范围内。在主宕机时，同步副本集中的其余节点有机会被提升为主，有时需要对系统进行容灾演练，或者对某些机器进行维护或灰度升级时希望定向的切换某一个副本成为新主，这又产生了优先副本（PriorityReplica）的概念。选择优先副本的原则和策略很多，可以动态选择水位最高，加入时间最久或 CommitLog 最长的副本，也可以支持机架，可用区优先这类静态策略。 从模型的角度来看，RocketMQ 单节点上 Topic 数量较多，如果像 kafka 以 topic / partition 粒度维护状态机，节点宕机会导致上万个状态机切换，这种惊群效应会带来很多潜在风险，因此 v4 版本时 RocketMQ 选择以单个 Broker 作为切换的最小粒度来管理，相比于其他更细粒度的实现，副本身份切换时只需要重分配 Broker 编号，对元数据节点压力最小。由于通信的数据量少，可以加快主备切换的速度，单个副本下线的影响被限制在副本组内，减少管理和运维成本。这种实现也一些缺点，例如存储节点的负载无法以最佳状态在集群上进行负载均衡，Topic 与存储节点本身的耦合度较高，水平扩展一般会改变分区总数，这就需要在上层附加额外的处理逻辑。 为了更规范更准确的衡量副本组的可用性指标，学术上就引入了几个名词： RTO（Recovery Time Objective）恢复时间目标，一般表示业务中断到恢复的时间。 RPO（Recovery Point Object）恢复点目标，用于衡量业务连续性。例如某个硬盘每天备份，故障时丢失最近备份后的所有更新。 SLA（ServiceLevel Agreement）服务等级协议，厂商以合约的形式对用户进行服务质量承诺，SLA 越高通常成本也越高。 节点数量与可靠性关系密切，根据不同生产场景，RocketMQ 的一个副本组可能会有 1，2，3，5 个副本。 1. 单副本成本最低，维护最简单，宕机时其他副本组接管新消息的写入，但已写入的数据无法读取，造成部分消息消费延迟。底层硬件故障还可能导致数据永久丢失，一般用于非关键日志，数据采集等低可靠性成本诉求较强的场景。 2. 两副本较好的权衡了数据冗余的成本与性能，RocketMQ 跨副本组容灾的特性使得两副本模式适用于绝大部分 IOPS 比较高的场景。此时备机可以分摊一定的读压力（尤其是主副本由于内存紧张或者产生冷读时）。两副本由于不满足多数派（quorum）原则，没有外部系统的参与时，故障时无法进行选举切换。 3. 三副本和五副本是业界使用最为广泛的，精心设计的算法使得多数情况下系统可以自愈。基于 Paxos / Raft 属于牺牲高可用性来保证一致性的 CP 型设计，存储成本很高，容易受到 IO 分布不均匀和水桶效应的影响。每条数据都需要半数以上副本响应的设计在需要写透（write through）多副本的消息场景下不够灵活。 日志复制还是消息复制 如何保证副本组中数据的最终一致性？那肯定是通过数据复制的方式实现，我们该选择逻辑复制还是物理复制呢？ 逻辑复制：使用消息来进行同步的场景也很多，各种 connector 实现本质上就是把消息从一个系统挪到另外一个系统上，例如将数据导入导出到 ES，Flink 这样的系统上进行分析，根据业务需要选择特定 Topic / Tag 进行同步，灵活程度和可扩展性非常高。这种方案随着 Topic 增多，系统还会有服务发现，位点和心跳管理等上层实现造成的性能损失。因此对于消息同步的场景，RocketMQ 也支持以消息路由的形式进行数据转移，将消息复制作为业务消费的特例来看待。 物理复制：大名鼎鼎的 MySQL 对于操作会记录逻辑日志（bin log）和重做日志（redo log）两种日志。其中 bin log 记录了语句的原始逻辑，比如修改某一行某个字段，redo log 属于物理日志，记录了哪个表空间哪个数据页改了什么。在 RocketMQ 的场景下，存储层的 CommitLog 通过链表和内核的 MappedFile 机制抽象出一条 append only 的数据流。主副本将未提交的消息按序传输给其他副本（相当于 redo log），并根据一定规则计算确认位点（confirm offset）判断日志流是否被提交。这种方案仅使用一份日志和位点就可以保证主备之间预写日志的一致性，简化复制实现的同时也提高了性能。 为了可用性而设计的多副本结构，很明显是需要对所有需要持久化的数据进行复制的，选择物理复制更加节省资源。RocketMQ 在物理复制时又是如何保证数据的最终一致性呢？这就涉及到数据的水位对齐。对于消息和流这样近似 FIFO 的系统来说，越近期的消息价值越高，消息系统的副本组的单个节点不会像数据库系统一样，保留这个副本的全量数据，Broker 一方面不断的将冷数据规整并转入低频介质来节约成本，同时对热数据盘上的数据也会由远及近滚动删除。如果副本组中有副本宕机较久，或者在备份重建等场景下就会出现日志流的不对齐和分叉的复杂情况。在下图中我们将主节点的 CommitLog 的首尾位点作为参考点，这样就可以划分出三个区间。在下图中以蓝色箭头表示。排列组合一下就可以证明备机此时的 CommitLog 一定满足下列 6 种情况之一。 下面对每种情况进行讨论与分析： 11 情况下满足备 Max 主 Max，可能由于主异步写磁盘宕机后又成为主，或者网络分区时双主写入造成 CommitLog 分叉。由于新主落后于备，少量未确认的消息丢失，非正常模式的选举（RocketMQ 将这种情况称为 unclean 选举）是应该尽量避免的。 33 理论上不会出现，备的数据长于主，原因可能是主节点数据丢失又叠加了非正常选举，因此这种情况需要人工介入处理。 租约与节点身份变更 前文提到 RocketMQ 每个副本组的主副本才接受外部写请求，节点的身份又是如何决定的呢？ 分布式系统一般分为中心化架构和去中心化架构。对于 MultiRaft，每个副本组包含三个或者五个副本，副本组内可以通过 Paxos / Raft 这样的共识协议来进行选主。典型的中心化架构，为了节省数据面资源成本会部署两副本，此时依赖于外部 ZK，ETCD，或者 DLedger Controller 这样的组件作为中心节点进行选举。由外置组件裁决成员身份涉及到分布式中两个重要的问题：1. 如何判断节点的状态是否正常。2. 如何避免双主问题。 对于第一个问题，kubernetes 的解决方案相对优雅，k8s 对与 Pod 的健康检查包括存活检测（Liveness probes）和就绪检测（Readiness probes），Liveness probes 主要是探测应用是否还活着，失败时重启 Pod。Readiness probes 来判断探测应用是否接受流量。简单的心跳机制一般只能实现存活检测，来看一个例子：假设有副本组中有 A、B、C 三个副本，另有一个节点 Q（哨兵） 负责观测节点状态，同时承担了全局选举与状态维护的职责。节点 A、B、C 周期性的向 Q 发送心跳，如果 Q 超过一段时间（一般是两个心跳间隔 ）收不到某个节点的心跳则认为这个节点异常。如果异常的是主副本，Q 将副本组的其他副本提升为主并广播告知其他副本。 在工程实践中，节点下线的可能性一般要小于网络抖动的可能性。我们假设节点 A 是副本组的主，节点 Q 与节点 A 之间的网络中断。节点 Q 认为 A 异常。重新选择节点 B 作为新的 Master，并通知节点 A、B、C 新的 Master 是节点 B。节点 A 本身工作正常，与节点 B、C 之间的网络也正常。由于节点 Q 的通知事件到达节点 A、B、C 的顺序是未知的，假如先达到 B，在这一时刻，系统中同时存在两个工作的主，一个是 A，另一个是 B。假如此时 A、B 都接收外部请求并与 C 同步数据，会产生严重的数据错误。上述 "双主" 问题出现的原因在于虽然节点 Q 认为节点 A 异常，但节点 A 自己不认为自己异常，在旧主新主都接受写入的时候就产生了日志流的分叉，其问题的本质是由于网络分区造成的系统对于节点状态没有达成一致。 租约是一种避免双主的有效手段，租约的典型含义是现在中心节点承认哪个节点为主，并允许节点在租约有效期内正常工作。如果节点 Q 希望切换新的主，只需等待前一个主的租约过期，则就可以安全的颁发新租约给新 Master 节点，而不会出现双主问题。这种情况下系统对 Q 本身的可用性诉求非常高，可能会成为集群的性能瓶颈。生产中使用租约还有很多实现细节，例如依赖时钟同步需要颁发者的有效期设置的比接收者的略大，颁发者本身的切换也较为复杂。 在 RocketMQ 的设计中，希望以一种去中心化的设计降低中心节点宕机带来的全局风险，（这里认为中心化和是否存在中心节点是两件事）所以没有引入租约机制。在 Controller （对应于 Q ）崩溃恢复期间，由于 Broker 对自己身份会进行永久缓存，每个主副本会管理这个副本组的状态机，RocketMQ Dledger Controller 这种模式能够尽量保证在大部分副本组在哨兵组件不可用时仍然不影响收发消息的核心流程。而旧主由于永久缓存身份，无法降级导致了网络分区时系统必须容忍双主。产生了多种解决方案，用户可以通过预配置选择 AP 型可用性优先，即允许系统通过短时分叉来保障服务连续性（下文还会继续谈谈为什么消息系统中分叉很难避免），还是 CP 型一致性优先，通过配置最小副本 ack 数超过集群半数以上节点。此时发送到旧主的消息将因为无法通过 ha 链路将数据发送给备，向客户端返回超时，由客户端将发起重试到其他分片。客户端经历一个服务发现的周期之后，客户端就可以正确发现新主。 特别的，在网络分区的情况下，例如旧主和备，Controller 之间产生网络分区，此时由于没有引入租约机制，旧主不会自动降级，旧主可以配置为异步双写，每一条消息需要经过主备的双重确认才能向客户端返回成功。而备在切换为主时，会设置自己只需要单个副本确认的同步写盘模式。此时，客户端短时间内仍然可以向旧主发送消息，旧主需要两副本确认才能返回成功，因此发送到旧主的消息会返回 SLAVE_NOT_AVAILABLE 的超时响应，通过客户端重试将消息发往新的节点。几秒后，客户端从 NameServer / Controller 获取新的路由时，旧主从客户端缓存中移除，此时完成了备节点的提升。 外置的组件可以对节点身份进行分配，上图展示了一个两副本的副本组上线流程： 1. 多个 Controller 通过选举和对 Broker 的请求进行重定向，最终由一个 Controller 做为主节点进行身份分配。 2. 如果 RocketMQ 副本组存在多个副本且需要选主，节点默认以备的身份启动，备节点会将自己注册到 Controller。 3. 节点从 Controller 获取 BrokerMemberGroup，包含了这个副本组的描述和连接信息。 1. 若分配的身份为备，解析出主节点的对外服务的地址并连接，完成日志截断后进行 HA 同步。 2. 若分配的身份为主，等待备机连接到自身的 HA 端口，并向 NameServer 再次宣告自己是主节点。 4. 主节点维护整个副本组的信息，向备发起数据复制，周期性的向 Controller 汇报主备之间水位差距，复制速度等。 RocketMQ 弱依赖 Controller 的实现并不会打破 Raft 中每个 term 最多只有一个 leader 的假设，工程中一般会使用 Leader Lease 解决脏读的问题，配合 Leader Stickiness 解决频繁切换的问题，保证主的唯一性。 Leader Lease: 租约，上一任 Leader 的 Lease 过期后，等待一段时间再发起 Leader 选举。 Leader Stickiness：Leader Lease 未过期的 Follower 拒绝新的 Leader 选举请求。 _注：Raft 认为具有最新已提交的日志的节点才有资格成为 Leader，而 MultiPaxos 无此限制。_ 对于日志的连续性问题，Raft 在确认一条日志之前会通过位点检查日志连续性，若检查到日志不连续会拒绝此日志，保证日志连续性，MultiPaxos 允许日志中有空洞。Raft 在 AppendEntries 中会携带 Leader 的 commit index，一旦日志形成多数派，Leader 更新本地的 commit index（对应于 RocketMQ 的 confirm offset）即完成提交，下一条 AppendEntries 会携带新的 commit index 通知其它节点，MultiPaxos 没有日志连接性假设，需要额外的 commit 消息通知其它节点。 计算日志分叉位点 除了网络分区，很多情况导致日志数据流分叉。有如下案例：三副本采用异步复制，异步持久化，A 为旧主 B C 为备，切换瞬间 B 日志水位大于 C，此时 C 成为新主，B C 副本上的数据会产生分叉，因为 B 还多出了一段未确认的数据。那么 B 是如何以一个简单可靠的方法去判断自己和 C 数据分叉的位点？ 一个直观的想法就是，直接将主备的 CommitLog 从前向后逐渐字节比较，一般生产环境下，主备都有数百 GB 的日志文件流，读取和传输大量数据的方案费时费力。很快我们发现，确定两个大文件是否相同的一个好办法就是比较数据的哈希值，需要对比的数据量一下子就从数百 GB 降低为了几百个哈希值，对于第一个不相同的 CommitLog 文件，还可以采取局部哈希的方式对齐，这里仍然存在一些计算的代价。还有没有优化的空间呢，那就是利用任期 Epoch 和偏移量 StartOffset 实现一个新的截断算法。这种 EpochStartOffset 满足如下原则： 1. 通过共识协议保证给定的一个任期 Epoch 只有一个Leader。 2. 只有 Leader 可以写入新的数据流，满足一定条件才会被提交。 3. Follower 只能从 Leader 获取最新的数据流，Follower 上线时按照选举算法进行截断。 下面是一个选举截断的具体案例，选举截断算法思想和流程如下： 主 CommitLog Min = 300，Max = 2500，EpochMap = {, , }备 CommitLog Min = 300，Max = 2500，EpochMap = {, , } 1. 备节点连接到主节点进行 HA 协商，获取主节点的 EpochStartOffset 信息并比较 2. 备从后向前找到任期起始点相同的那个点作为分叉任期，在上述案例里是 3. 选择这个任期里主备结束位点的最小值（如果主副本没有切换且为最大任期，则主副本的结束位点是无穷大） 实现的代码如下： ${e} 数据回发与日志截断 故障发生后，系统将会对分叉数据进行修复，有很多小小细节值得深究与探讨。 在实现数据截断的过程中，有一个很特殊的动作，当备切主的时候要把 ConsumeQueue 的 Confirm Offset 提升到 CommitLog 的 MaxPhyOffset，即使这一部分数据在主上是否被提交是未知的。回想起几年前看 Raft 的时候，当一条日志被传输到 Follower，Follower 确认收到这条消息，主再把这条日志应用到自己的状态机时，通知客户端和通知所有的 follower 去 commit 这条日志这两件事是并行的，假如 leader 先回复 client 处理成功，此时 leader 挂了，由于其他 follower 的确认位点 confirm offset 一般会略低于 leader，中间这段未决日志还没应用到 follower 的状态机上，这时就出现了状态机不一致的情况，即已经写入 leader 的数据丢失了。让我们来举一个具体的案例，假设两副本一主一备： 1. 主的 max offset = 100，主向备发送当前 confirm offset = 40 和 message buffer = [40100] 的数据 2. 备向主回复 confirm offset = 100 后主需要同时做几件事 1. 本地提交（apply） [40100] 区间的数据，用后台的 dispatch 线程异步构建这段数据的索引 2. 向 producer 响应 [40100] 这段数据是发送成功的。 3. 向多个备机异步的提交，实际上是发送了 confirm offset = 100 3. 此时主突然宕机，备机的 confirm offset 可能是 [40100] 中的值 所以当备切换为主的时候，如果直接以 40 进行截断，意味着客户端已经发送到服务端的消息丢失了，正确的水位应该被提升至 100。但是备还没有收到 2.3 的 confirm = 100 的信息，这个行为相当于要提交了未决消息。事实上新 leader 会遵守 "Leader Completeness" 的约定，切换时任何副本都不会删除也不会更改旧 leader 未决的 entry。新 leader 在新的 term 下，会直接应用一个较大的版本将未决的 entry 一起提交，这里副本组主备节点的行为共同保证了复制状态机的安全性。 那么备切换成功的标志是什么，什么时候才能接收 producer 新的流量呢？对于 Raft 来说一旦切换就可以，对于 RocketMQ 来说这个阶段会被稍稍推迟，即索引已经完全构建结束的时候。RocketMQ 为了保证构建 consume queue 的一致性，会在 CommitLog 中记录 consume queue offset 的偏移量，此时 confirm offset 到 max offset 间的数据是副本作为备来接收的，这部分消息在 consume queue 中的偏移量已经固定下来了，而 producer 新的流量时由于 RocketMQ 预计算位点的优化，等到消息实际放入 CommitLog 的再真实的数据分发（dispatch）的时候就会发现对应位置的 consume queue 已经被占用了，此时就造成了主备索引数据不一致。本质原因是 RocketMQ 存储层预构建索引的优化对日志有一些侵入性，但切换时短暂等待的代价远远小于正常运行时提速的收益。 消息中间件场景 a. 元数据变更是否依赖于日志 目前 RocketMQ 对于元数据是在内存中单独管理的，备机间隔 5 秒向当前的主节点同步数据。例如当前主节点上创建了一个临时 Topic 并接受了一条消息，在一个同步周期内这个 Topic 又被删除了，此时主备节点元数据可能不一致。又比如位点更新的时候，对于单个队列而言，多副本架构中存在多条消费位点更新链路，Consumer 拉取消息时更新，Consumer 主动向 broker 更新，管控重置位点，HA 链路更新，当副本组发生主备切换时，consumer group 同时发生 consumer 上下线，由于路由发现的时间差，还可能造成同一个消费组两个不同 consumer 分别消费同一副本组主备上同一个队列的情况。 原因在于备机重做元数据更新和消息流这两件事是异步的，这有一定概率会造成脏数据。由于 RocketMQ 单个节点上 Topic / Group 数量较多，通过日志的实现会导致持久化的数据量很大，在复杂场景下基于日志做回滚依赖 snapshot 机制也会增加计算开销和恢复时间。这个问题和数据库很像，MySQL 在执行 DDL 修改元数据时通过会创建 MDL 锁，阻塞用户其他操作访问表空间的访问。备库同步主库也会加锁，元数据修改开始点和结束点所代表的两个日志并不是一个原子操作，这意味着主库上在修改元数据的过程中如果宕机了，备库上持有的 MDL 锁就无法释放。MySQL 的解决方案是在主库每次崩溃恢复后，都写一条特殊的日志，通知所有连接的备库释放其持有的所有 MDL 排他锁。对所有操作都走日志流进行状态机复制要求存储层有多种日志类型，实现也更加复杂。RocketMQ 选择以另一种同步的模式操作，即类似 ZAB 这样二阶段协议，例如位点更新时的可以选择配置 LockInStrictMode 让备都同步这条修改。事实上 RocketMQ 为了优化上述位点跳跃的现象，客户端在未重启时，遇到服务端主备切换还会用优先采纳本地位点的方式获取消息，进一步减少重复消费。 b. 同步复制与异步复制 同步复制的含义是用户的一个操作在多个副本上都已经提交。正常情况下，假设一个副本组中的 3 个副本都要对相同一个请求进行确认，相当于数据写透 3 个副本（简称 33 写），33 写提供了非常高的数据可靠性，但是把所有从节点都配置为同步复制时任何一个同步节点的中断都会导致整个副本组处理请求失败。当第三个副本是跨可用区时，长尾也会带来一定的延迟。 异步复制模式下，尚未复制到从节点的写请求都会丢失。向客户端确认的写操作也无法保证被持久化。异步复制是一种故障时 RPO 不为 0 的配置模式，由于不用考虑从节点上的状态，总是可以继续响应写请求，系统的延迟更低，吞吐性能更好。为了权衡两者，通常只有其中一个从节点是同步的，而其他节点是异步的模式。只要同步的从节点变得不可用或性能下降，则将另一个异步的从节点提升为同步模式。这样可以保证至少有两个节点（即主节点和一个同步从节点）拥有最新的数据副本。这种模式称为 23 写，能帮助避免抖动，提供更好的延迟稳定性，有时候也叫称为半同步。 在 RocketMQ 的场景中，异步复制也被广泛应用在消息读写比极高，从节点数量多或者异地多副本场景。同步复制和异步复制是通过 Broker 配置文件里的 brokerRole 参数进行设置的，这个参数可以被设置成 ASYNC_MASTER、SYNC_MASTER、SLAVE 三个值中的一个。实际应用中要结合业务场景合理设置持久化方式和主从复制方式，通常，由于网络的速度高于本地 IO 速度，采用异步持久化和同步复制是一个权衡性能与可靠性的设置。 c. 副本组自适应降级 同步复制的含义是一条数据同时被主备确认才返回用户操作成功，可以保证主宕机后消息还在备中，适合可靠性要求较高的场景，同步复制还可以限制未同步的数据量以减少 ha 链路的内存压力，缺点则是副本组中的某一个备出现假死就会影响写入。异步复制无需等待备确认，性能高于同步复制，切换时未提交的消息可能会丢失（参考前文的日志分叉）。在三副本甚至五副本且对可靠性要求高的场景中无法采用异步复制，采用同步复制需要每一个副本确认后才会返回，在副本数多的情况下严重影响效率。关于一条消息需要被多少副本确认这个问题，RocketMQ 服务端会有一些数量上的配置来进行灵活调整： TotalReplicas：全部副本数 InSyncReplicas：每条消息至少要被这个数量的 Broker 确认（如果主为 ASYNC_MASTER 或者 AllAck 模式则该参数不生效） MinInSyncReplicas：最小的同步副本数，如果 InSyncReplicas 。对于正常情况下，两个副本会处于同步复制，当备下线或假死时，会进行自适应降级，保证主节点还能正常收发消息，这个功能为用户提供了一个可用性优先的选择。 d. 轻量级心跳与快速隔离 在 RocketMQ v4.x 版本的实现中，Broker 周期性的（间隔 30 秒）将自身的所有 Topic 序列化并传输到 NameServer 注册进行保活。由于 Broker 上 Topic 的元数据规模较大，带来了较大的网络流量开销，Broker 的注册间隔不能设置的太短。同时 NameServer 对 Broker 是采取延迟隔离机制，防止 NameServer 网络抖动时可能瞬间移除所有 Broker 的注册信息，引发服务的雪崩。默认情况下异常主宕机时超过 2 分钟，或者备切换为主重新注册后才会替换。容错设计的同时导致 Broker 故障转移缓慢，RocketMQ v5.0 版本引入轻量级心跳（参数liteHeartBeat），将 Broker 的注册行为与 NameServer 的心跳进行了逻辑拆分，将心跳间隔减小到 1 秒。当 NameServer 间隔 5 秒（可配置）没有收到来自 Broker 的心跳请求就对 Broker 进行移除，使异常场景下自愈的时间从分钟级缩短到了秒级。 RocketMQ 高可用架构演进路线 无切换架构的演进 最早的时候，RocketMQ 基于 MasterSlave 模式提供了主备部署的架构，这种模式提供了一定的高可用能力，在 Master 节点负载较高情况下，读流量可以被重定向到备机。由于没有选主机制，在 Master 节点不可用时，这个副本组的消息发送将会完全中断，还会出现延迟消息、事务消息、Pop 消息等二级消息无法消费或者延迟。此外，备机在正常工作场景下资源使用率较低，造成一定的资源浪费。为了解决这些问题，社区提出了在一个 Broker 进程内运行多个 BrokerContainer，这个设计类似于 Flink 的 slot，让一个 Broker 进程上可以以 Container 的形式运行多个节点，复用传输层的连接，业务线程池等资源，通过单节点主备交叉部署来同时承担多份流量，无外部依赖，自愈能力强。这种方式下隔离性弱于使用原生容器方式进行隔离，同时由于架构的复杂度增加导致了自愈流程较为复杂。 切换架构的演进 另一条演进路线则是基于可切换的，RocketMQ 也尝试过依托于 Zookeeper 的分布式锁和通知机制进行 HA 状态的管理。引入外部依赖的同时给架构带来了复杂性，不容易做小型化部署，部署运维和诊断的成本较高。另一种方式就是基于 Raft 在集群内自动选主，Raft 中的副本身份被透出和复用到 Broker Role 层面去除外部依赖，然而强一致的 Raft 版本并未支持灵活的降级策略，无法在 C 和 A 之间灵活调整。两种切换方案都是 CP 设计，牺牲高可用优先保证一致性。主副本下线时选主和路由定时更新策略导致整个故障转移时间依然较长，Raft 本身对三副本的要求也会面临较大的成本压力，RocketMQ 原生的 TransientPool，零拷贝等一些用来避免减少 IO 压力的方案在 Raft 下无法有效使用。 RocketMQ DLedger 融合模式 RocketMQ DLedger 融合模式是 RocketMQ 5.0 演进中结合上述两条路线后的一个系统的解决方案。核心的特性有以下几点： 1. 利用可内嵌于 NameServer 的 Controller 进行选主，无外部依赖，对两副本支持友好。 2. 引入 EpochStartOffset 机制来计算日志分叉位点。 3. 消息在进行写入时，提供了灵活的配置来协调系统对于可用性还是一致性优先的诉求。 4. 简化日志复制协议使得日志复制为高效。 几种实现对比表如下： 与其他消息系统的对比 控制节点 1. 是否强制要求选主 Kafka 的 Controller 是 Broker 选举产生，这需要有一个存储节点间的服务发现机制。RocketMQ 的 Controller 可以作为管控节点单独存在。对 Kafka，Pulsar 而言必须选择主副本进行写入，随着时间的运行节点之间负载需要通过复杂的方案进行再均衡。对 RocketMQ 的融合架构而言，由于选主是可选的，静态布局的方案（例如无需依赖复杂的动态调度就可以较为均衡的实现跨机架跨可用区），并且无切换与切换架构可以相互转换。 2. Controller 的逻辑复杂度 RocketMQ Controller 相比 Kafka Controller 更加轻量，Kafka 的 Controller 承担 Partition 粒度的 ISR 维护和选举等功能，而RocketMQ 的 Controller 维护的数据是副本组粒度的，对于元数据只维护节点身份，更加简单。RocketMQ Controller 可以独立部署，也可以内嵌 NameServer 运行。 3. Controller 依赖程度 RocketMQ Broker 的同步副本集维护是 Master Broker 节点上报，由于不强依赖中心节点来提供租约，controller 宕机时虽然无法为同时有主故障的副本组选举，但不影响绝大部分副本组可用性。Pulsar 中通过 fencing 机制防止有多个 writer（pulsar 中的计算节点称为 broker）同时写同一个 partition，是对外部有依赖的。 数据节点 1. 副本存储结构的抽象与最小粒度不同，在这一点上其实三者的设计各有优势 Kafka 的存储抽象粒度是 Partition，对每个分区进行维护多副本，扩容需要进行数据复制，对于冷读支持更好。 RocketMQ 的日志流是 Broker 粒度的，顺序写盘效率更高，在磁盘空间不足时一般选择水平扩容，只需复制元数据。 Pulsar 其实抽象了一个分布式日志流 Journal，分区被进一步分成分片，根据配置的时间或大小进行滚动，扩容只需复制元数据。 2. 复杂的参数配置被收敛至服务端 Kafka 和 RocketMQ 都支持灵活的配置单条消息的 ack 数，即权衡数据写入灵活性与可靠性。RocketMQ 在向云原生演进的过程希望简化客户端 API 与配置，让业务方只需关心消息本身，选择在服务端配置统一配置这个值。 3. 副本数据的同步方式不同 Pulsar 采用星型写：数据直接从 writer 写到多个 bookeeper。适合客户端与存储节点混部场景。数据路径只需要 1 跳，延迟更低。缺点是当存储计算分离时，星型写需要更多的存储集群和计算集群间网络带宽。 RocketMQ 和 Kafka 采用 Y 型写：client 先写到一个主副本，由其再转发给另外 Broker 副本。虽然服务端内部带宽充裕，但需要 2 跳网络，会增加延迟。Y 型写利于解决文件多客户端写的问题，也更容易利用 23 写克服毛刺，提供更好的延迟稳定性。 高可用架构的未来 仔细阅读 RocketMQ 的源码，其实大家也会发现 RocketMQ 在各种边缘问题处理上细节满满，节点失效，网络抖动，副本一致性，持久化，可用性与延迟之间存在各种细微的权衡，这也是 RocketMQ 多年来在生产环境下所积累的核心竞争力之一。随着分布式技术的进一步发展，更多更有意思的技术，如基于 RDMA 网络的复制协议也呼之欲出。RocketMQ 将与社区协同进步，发展为 “消息，事件，流” 一体化的融合平台。 参考文档： 1. Paxos design： 2. SOFAJRaft： 3. Pulsar Geo Replication： 4. Pulsar Metadata： 5. Kafka Persistence： 6. Kafka Balancing leadership： 7. Windows Azure Storage: A Highly Available Cloud Storage Service with Strong Consistency： 8. PolarDB Serverless: A Cloud Native Database for Disaggregated Data Centers：

前言 回顾 RocketMQ 的发展历程，至今已十年有余。2022 年 RocketMQ 5.0 正式发布，全面迈进云原生时代。 11 月 5 日，2022 杭州 · 云栖大会上，阿里云智能高级产品专家杨秋弟在云原生峰会上发表主题演讲，发布消息队列 RocketMQ 5.0：从消息服务到云原生事件流处理平台。 阿里云智能高级产品专家&Apache RocketMQ 联合创始人 杨秋弟 Apache RocketMQ 发展史 回顾 Apache RocketMQ 过去十年的发展历程，可分为“诞生于互联网”与“成长于云计算”两大阶段。 第一个阶段是 RocketMQ 的从 0 到 1，在阿里内部规模化落地。2012 年，为了支撑超大规模电商互联网架构，阿里中间件研发了 RocketMQ，并在产品诞生初期开源，2017 年 RocketMQ 统一了阿里消息技术体系。 第二个阶段是云计算。2015 年 RocketMQ 上云，这也是业界首个提供公共云 SaaS 形态的开源消息队列；2016 年，阿里把 RocketMQ 捐赠给 Apache，2017 年孵化毕业，成为国内首个 TLP 的互联网中间件。 十年磨一剑，出鞘必锋芒。在这十年的过程中，通过集团打磨稳定性，依托云计算孵化创新，开源共建加速标准化建立与生态连接，RocketMQ 始终坚持开源、集团、商业三位一体的发展思路，内核演进和产品迭代齐头并进。2022 年 RocketMQ 5.0 正式发布宣告着全面迈进云原生时代。 RocketMQ 5.0：从消息服务到云原生事件流平台 回顾过去这十年，RocketMQ 服务于集团几乎所有的业务，在阿里云上更是累计服务了 10 万余家企业客户，覆盖互联网、零售、金融、汽车等 20 多个行业，大规模的生产实践持续累积产品的核心优势。 多样性。企业级应用复杂的业务诉求，催生 RocketMQ 提供丰富的消息类型，比如定时消息、事务消息、顺序消息等等。此外，也提供了像消息轨迹、消息回溯等一系列的消息治理能力。 一致性。无论是淘宝交易还是蚂蚁支付都天然对数据一致性有着极高的要求，RocketMQ 提供的分布式事务消息是业内第一个提供该特性的消息产品，将异步解耦与数据一致性完美融合，是金融客户中不可或缺的产品能力。 稳定性。稳定性是产品的根基，更是一个系统工程，RocketMQ 长期在电商和金融领域中打磨，不仅提供最高达 99.99% SLA，更是帮助客户提供全方位的健康巡检与故障排查能力，如消息轨迹、消息回溯、死信机制等等，提供多样化的稳定性兜底手段。 高性能。在双十一的极限流量下，RocketMQ 具备无限扩展能力，支持千万级并发与万亿级消息洪峰，P9999 写延迟在 1ms 内，100%在 100ms 内。 可以说，在消息领域，尤其在业务消息领域，RocketMQ 在国内已经做到顶尖，成为企业客户的首选。 而随着云原生以及数字化时代的到来，RocketMQ 也在快速的发生着变化，那么变化主要体现在哪些方面呢？ 首先，全面拥抱云原生。向下，消息系统自身实现云原生架构的演进，充分释放云基础设施的池化能力，全方位提高消息的核心技术指标。向上，消息产品形态持续演进，成为云原生应用架构的核心引擎。比如微服务、事件驱动、Serverless 等现代化应用架构。 其次，全面拥抱实时数据。企业的数字化转型从原来的业务数字化迈向了数字业务化。对业务数据的实时洞察、实时决策成为指导业务成功的关键要素。消息队列也将从在线业务架构的基础设施，延伸到实时数据架构的基础设施，从而实现事务分析一体化。 随着 5.0 的发布，RocketMQ也正式从消息服务升级到云原生事件流处理平台。 RocketMQ 5.0：云原生架构升级 首先来看 RocketMQ 自身的云原生架构演进。从下面的全景图可以看出，RocketMQ 从客户端到服务端都进行了全方位的改造，具体体现在以下几个方面： 轻量化。RocketMQ 4.0 诞生于 2012 年的淘宝电商，当时大部分的业务还跑在物理机上，单节点计算能力强，客户端节点数少且较为稳定，因此，富客户端的接入方式不仅更加高效，更可以提供诸如客户端侧负载均衡、消息缓存、故障转移等一系列企业级特性。但这种模式在云原生时代发生了改变，轻量客户端更容易被云原生技术栈所集成。因此，RocketMQ 5.0 客户端采用轻量 SDK 设计理念，将原来富客户端的逻辑下沉到服务端，满足现代化应用轻量化、Serverless 化以及 Mesh 化的趋势，更容易被集成；同时也正是因为轻量化，使得 SDK 多语言开发成本低了很多，快速覆盖当下主流的多语言版本。 弹性。存算分离架构让无状态计算节点可以快速伸缩，而分级存储以及冷热分离架构更是让消息存储具备更强的弹性能力。 高可用。基于全新的 Leaderless 架构，去 ZK 依赖的同时，可以做到副本数灵活选择，同步异步自动升降级，实现秒级故障转移；面向云的多可用区、多地域组建全局高可用能力。 基础设施云原生化。RocketMQ 整体架构走向 Kubernetes 化，拥抱 OpenTelemetry，依托于阿里云提供的 ARMS、Prometheus 以及Grafana 实现可观测能力的云原生化。 而 RocketMQ 5.0 本次的升级，除了在技术架构云原生化之外，在产品能力以及成本优化方面都有着重大的提升，我们来逐一分解。 轻量无状态消费模型 RocketMQ 4.0 采用按队列消费模型，消费者完全按照队列负载均衡，非常适合批量拉取快速消费，对单一消息状态不敏感的场景，比如流计算。然而在业务消息领域，尤其是金融场景以及事件驱动架构之下，每一条消息状态都是极为重要的。再加上不同业务类型的消息处理耗时也是各不相同，从毫秒级、到秒级甚至到分钟级，队列的负载不均衡或者短暂的 Block 都可能会引发消息的局部堆积，从而影响到最终用户的体验。因此，RocketMQ 5.0 全新推出按消息负载的轻量无状态消费模型，通过 PoP 机制巧妙地在队列模型之上构建了消息模型，业务只需要关心消息而无需关心队列，所有 API 都能够支持单条消息级别控制，如消息的消费、重试、删除等。而基于消息消费模型，客户端、连接和消费都是无状态的，可在任意 Proxy 节点上飘移，真正做到轻量化。 RocketMQ 5.0 提供按队列消费模型以及按消息消费模型，更好的满足事件与流的业务场景，正可谓鱼与熊掌兼得。 海量消息分级存储 RocketMQ 5.0 的另一个重大升级则是海量消息的分级存储。对消息队列了解的同学都知道，消息通常都是流动的短时间的存储，业内大部分消息产品对消息的保留时间都比较优先，3 天，7 天，最长 15 天不等。有限的存储空间使不仅限制了消息的保留时长，更在某些场景下可能会导致业务资损，比如在消息还没有被消费的时候，因为磁盘空间不足或者消息过期而被清除，这在金融等领域都是不可接受的。所以，RocketMQ 一直想要解决这样的问题，让存储变得更有弹性。 RocketMQ 5.0 基于ESSD、对象存储打造冷热分离以及分级存储架构，提供低成本的无限存储能力，确保消息不会因为本地磁盘空间不足而提前被清除，造成业务资损。我们提供消息存储的 Serverless，客户只需按实际存储使用量付费，而无需预购存储空间。 此外，流量削峰是消息队列极为常见的场景，而由此带来的海量消息堆积能力以及在堆积情况下的性能稳定性成为衡量产品性能的核心指标。RocketMQ 5.0 基于冷热数据分离架构进一步做到读写隔离，避免在堆积的场景下影响热数据的写入性能。分级存储的冷数据碎片规整能力更是提高了冷数据的读取性能，为用户提供一致的冷读 SLA。 售卖系列全线升级，最高降本 50% 从前面的介绍我们已经了解到，RocketMQ 5.0 在技术架构以及产品能力上都有着明显提升。 而 5.0 推出全新的售卖形态与计费策略相比 4.0 更简单、更灵活也更为普惠。实例的综合成本最高可降低 50%。接入门槛最低可降至 390 元/月，远低于自建成本。消息存储支持 Serverless 弹性能力，按需付费可大幅降低闲置成本。结合冷热分离的多级存储能力，相比开源自建可降低 67%，大幅降低消息队列的使用成本。 EventBridge：云上事件枢纽 事件驱动是一个起源很早的概念，早在几十年前，无论是操作系统内核的设计、还是客户端编程框架都大量采用了事件驱动技术。RocketMQ PushConsumer 提供的 API 其实就是一种事件驱动的编程范式，但在微服务应用架构当中，集成与通信才是刚需，事件驱动的价值并没有那么明显的体现。 而随着云原生时代的到来，计算力的构成越来越多样化。作为云原生的代表技术，Serverless 架构范式也是事件驱动的。无论是阿里云的函数计算、还是 AWS 的 Lambda，它们的主要触发源都是各种形态的事件，云产品事件，如 OSS 文件上传触发用户基于函数进行文件加工处理；用户业务事件，如 RocketMQ 触发函数运行消费逻辑处理等等。 以事件驱动为核心理念，阿里云推出了 EventBridge 产品，其使命就是打造云上的事件枢纽。通过EventBridge 可以兑现四大业务价值： 统一事件枢纽。阿里云从 IaaS、PaaS到第三方的 SaaS，每天都有数以亿计的事件产生，但却没有一种简单和统一的方式来触达这些事件；这些事件散落在各个地方形成『事件孤岛』，很难挖掘出有用的业务价值。只有充分发挥数据的规模效应，建立起数据之间的血缘关系，我们才能更好的发掘出数据的价值；所以 EventBridge 首要任务便是统一阿里云上的事件规范，拥抱CloudEvents 事件标准，打造阿里云统一的事件枢纽。 事件驱动引擎。当 EventBridge 连接了海量的事件源后，基于 RocketMQ 毫秒级的事件触发能力，必将加速企业 EDA/Serverless 的架构升级。 开放与集成。EventBridge 提供丰富的跨云、跨平台、跨产品、跨地域以及跨账号的连接能力，能够促进云产品、应用程序、SaaS 服务的相互集成。 低代码。EventBridge 借助Serverless 的应用中心，通过简单的规则匹配以及丰富的模板，即可实现事件的分发、过滤、转换等处理，进一步提升事件驱动的效率。 让消息无处不在，让事件无所不及 依托于 EventBridge、RocketMQ 以及函数计算 FC 的强强联合，目前 EventBridge 的事件生态已初具规模。 在云产品事件集成方面，目前已经集成 200+云产品事件源，3000 多种事件类型。 在数据集成与处理方面，EventBridge 与微服务应用、大数据、IoT 等场景深度集成。比如与消息生态的融合，阿里云 6 款消息队列产品通过 EventBridge 实现消息数据的互联互通，并且通过 EventBridge 的全球事件网络赋予消息全球消息路由的能力，同时也可以通过EventBridge 提供的丰富的模板，实现 ETL 数据处理能力。 在 SaaS 应用集成方面，包括钉钉、聚石塔以及云上 50 多个 SaaS 服务都可以通过 EventBridgehook 方式连接到 EventBridge。 在事件触发方面，EventBridge 目前已经触达 10 多个事件目标，海量的事件源都可以通过 EventBridge 触发包括 FC/SAE 等在内的 10 多款事件目标云产品。除此之外，目前 EventBridge 已经对接了阿里云全量的云产品 API，任何一个事件都可以通过云产品 API 的方式进行触达。 未来还有会更多的事件源以及事件目标接入到 EventBridge 上来。 RocketMQ Streams：轻量级计算的新选择 正如开篇所提到的，基于云原生架构的全面升级，RocketMQ 5.0 也将从在线业务架构的基础设施，延伸到实时数据架构的基础设施，实现事务分析一体化。将 RocketMQ Streams 打造成为轻量级计算的新选择。 业内最常见如 Flink、Spark 等大数据框架大多采用中心化的 MasterSlave 架构，依赖和部署比较重，每个任务的执行都需要很大的开销，有较高的使用成本。而与之不同的是，RocketMQ Streams 着重打造轻资源，高性能的轻量计算引擎，无额外依赖，最低1core，1g 即可完成部署，适用于大数据量、高过滤、轻窗口计算的场景，在资源敏感型场景，如消息队列流计算、安全风控，边缘计算等，RocketMQ Streams 具有有很大优势。阿里云消息团队与安全团队合作，通过对过滤场景做大量优化，性能提升 35 倍，资源节省 50%80%。 目前，RocketMQ Streams 已经在开源社区发布，未来计划在 2023 年 4 月在阿里云完成商业化。 RocketMQ 这十年，我们一同向前 RocketMQ历经十余年的打磨，已经取得了众多成果。全球拥有 700+的贡献者，1.8w Star 数，超过 80%的主流云厂商提供了 RocketMQ 的商业托管服务，Apache RocketMQ 社区始终保持着极高的活跃度，因此，也荣获了科创中国“开源创新榜”，中日韩开源软件优秀技术奖等十多个国内外开源奖项。 而阿里云作为 RocketMQ 的起源和核心贡献者，不仅 100%覆盖全集团业务，承载千万级并发万亿级消息洪峰。十多年以来更是累计服务 10w+万企业客户，覆盖互联网、零售、汽车等 20 多个行业，超过 75%的头部企业选择使用 RocketMQ。期望阿里云的消息队列 RocketMQ 可以成为广大企业客户的心之所选。也诚邀更广大的开发者来积极参与RocketMQ 的开源社区建设，一起将 RocketMQ 打造为消息领域的引领者。

本文目录 背景知识 大数据时代的构架演进 RocketMQ Connector&Stream Apache Hudi 构建Lakehouse实操 本文标题包含三个关键词：Lakehouse、RocketMQ、Hudi。我们先从整体Lakehouse架构入手，随后逐步分析架构产生的原因、架构组件特点以及构建Lakehouse架构的实操部分。 背景知识 1、Lakehouse架构 Lakehouse最初由Databrick提出，并对Lakehouse架构特征有如下要求： （1）事务支持 企业内部许多数据管道通常会并发读写数据。对ACID事务的支持确保了多方并发读写数据时的一致性问题； （2）Schema enforcement and governance Lakehouse应该有一种方式可以支持模式执行和演进、支持DW schema的范式（如星星或雪花模型），能够对数据完整性进行推理，并且具有健壮的治理和审计机制； （3）开放性 使用的存储格式是开放式和标准化的（如parquet），并且为各类工具和引擎，包括机器学习和Python/R库，提供API，以便它们可以直接有效地访问数据； （4）BI支持 Lakehouse可以直接在源数据上使用BI工具。这样可以提高数据新鲜度、减少延迟，并且降低了在数据池和数据仓库中操作两个数据副本的成本； （5）存储与计算分离 在实践中，这意味着存储和计算使用单独的集群，因此这些系统能够扩展到支持更大的用户并发和数据量。一些现代数仓也具有此属性； （6）支持从非结构化数据到结构化数据的多种数据类型 Lakehouse可用于存储、优化、分析和访问许多数据应用所需的包括image、video、audio、text以及半结构化数据； （7）支持各种工作负载 包括数据科学、机器学习以及SQL和分析。可能需要多种工具来支持这些工作负载，但它们底层都依赖同一数据存储库； （8）端到端流 实时报表是许多企业中的标准应用。对流的支持消除了需要构建单独系统来专门用于服务实时数据应用的需求。 从上述对Lakehouse架构的特点描述我们可以看出，针对单一功能，我们可以利用某些开源产品组合构建出一套解决方案。但对于全部功能的支持，目前好像没有一个通用的解决方案。接下来，我们先了解大数据时代主流的数据处理架构是怎样的。 大数据时代的架构演进 1、大数据时代的开源产品 大数据时代的开源产品种类繁多，消息领域的RocketMQ、Kafka；计算领域的flink、spark、storm；存储领域的HDFS、Hbase、Redis、ElasticSearch、Hudi、DeltaLake等等。 为什么会产生这么多开源产品呢？首先在大数据时代数据量越来越大，而且每个业务的需求也各不相同，因此就产生出各种类型的产品供架构师选择，用于支持各类场景。然而众多的品类产品也给架构师们带来一些困扰，比如选型困难、试错成本高、学习成本高、架构复杂等等。 2、当前主流的多层架构 大数据领域的处理处理场景包含数据分析、BI、科学计算、机器学习、指标监控等场景，针对不同场景，业务方会根据业务特点选择不同的计算引擎和存储引擎；例如交易指标可以采用binlog + CDC+ RocketMQ + Flink + Hbase + ELK组合，用于BI和Metric可视化。 （1）多层架构的优点：支持广泛的业务场景； （2）多层架构的缺点： 处理链路长，延迟高； 数据副本多，成本翻倍； 学习成本高； 造成多层架构缺点主要原因是存储链路和计算链路太长。 我们真的需要如此多的解决方案来支持广泛的业务场景吗？Lakehouse架构是否可以统一解决方案？ 多层架构的存储层是否可以合并？Hudi产品是否能够支持多种存储需求？ 多层架构的计算层是否可以合并？RocketMQ stream是否能够融合消息层和计算层？ 当前主流的多层架构 3、Lakehouse架构产生 Lakehouse架构是多层架构的升级版本，将存储层复杂度继续降低到一层。再进一步压缩计算层，将消息层和计算层融合，RocketMQ stream充当计算的角色。我们得到如下图所示的新架构。新架构中，消息出入口通过RocketMQ connector实现，消息计算层由RocketMQ stream实现，在RocketMQ内部完成消息计算中间态的流转；计算结果通过RocketMQHudiconnector收口落库Hudi，Hudi支持多种索引，并提供统一的API输出给不同产品。 Lakehouse架构 下面我们分析下该架构的特点。 （1）Lakehouse架构的优点： 链路更短，更适合实时场景，数据新鲜感高； 成本可控，降低了存储成本； 学习成本低，对程序员友好； 运维复杂度大幅降低； （2）Lakehouse架构的缺点 对消息产品和数据湖产品的稳定性、易用性等要求高，同时消息产品需要支持计算场景，数据湖产品需要提供强大的索引功能。 （3）选择 在Lakehouse架构中我们选择消息产品RocketMQ和数据湖产品Hudi。 同时，可以利用RocketMQ stream在RocketMQ集群上将计算层放在其中集成，这样就将计算层降低到一层，能够满足绝大部分中小型大数据处理场景。 接下来我们逐步分析RocketMQ和Hudi两款产品的特点。 RocketMQ Connector & Stream RocketMQ 发展历程图 RocketMQ从2017年开始进入Apache孵化，2018年RocketMQ 4.0发布完成云原生化，2021年RocketMQ 5.0发布全面融合消息、事件、流。 1、业务消息领域首选 RocketMQ作为一款“让人睡得着觉的消息产品”成为业务消息领域的首选，这主要源于产品的以下特点： （1）金融级高可靠 经历了阿里巴巴双十一的洪峰检验； （2）极简架构 如下图所示， RocketMQ的架构主要包含两部分包括：源数据集群NameServer Cluster和计算存储集群Broker Cluster。 RocketMQ 构架图 NameServer节点无状态，可以非常简单的进行横向扩容。Broker节点采用主备方式保证数据高可靠性，支持一主多备的场景，配置灵活。 搭建方式：只需要简单的代码就可以搭建RocketMQ集群： Jar： nohup sh bin/mqnamesrv & nohup sh bin/mqbroker n localhost:9876 & On K8S： kubectl apply f example/rocketmq_cluster.yaml （3）极低运维成本 RocketMQ的运维成本很低，提供了很好的CLI工具MQAdmin，MQAdmin提供了丰富的命令支持，覆盖集群健康状态检查、集群进出流量管控等多个方面。例如，mqadmin clusterList一条命令可以获取到当前集群全部节点状态（生产消费流量、延迟、排队长度、磁盘水位等）；mqadmin updateBrokerConfig命令可以实时设置broker节点或topic的可读可写状态，从而可以动态摘除临时不可用节点，达到生产消费的流量迁移效果。 （4）丰富的消息类型 RocketMQ支持的消息类型包括：普通消息、事务消息、延迟消息、定时消息、顺序消息等。能够轻松支持大数据场景和业务场景。 （5）高吞吐、低延迟 压测场景主备同步复制模式，每台Broker节点都可以将磁盘利用率打满，同时可以将p99延迟控制在毫秒级别。 2、RocketMQ 5.0概况 RocketMQ 5.0是生于云、长于云的云原生消息、事件、流超融合平台，它具有以下特点： （1）轻量级SDK 全面支持云原生通信标准 gRPC 协议； 无状态 Pop 消费模式，多语言友好，易集成； （2）极简架构 无外部依赖，降低运维负担； 节点间松散耦合，任意服务节点可随时迁移； （3）可分可合的存储计算分离 Broker 升级为真正的无状态服务节点，无 binding； Broker 和 Store节点分离部署、独立扩缩； 多协议标准支持，无厂商锁定； 可分可合，适应多种业务场景，降低运维负担； 如下图所示，计算集群（Broker）主要包括抽象模型和相对应的协议适配，以及消费能力和治理能力。存储集群（Store）主要分为消息存储CommitLog（多类型消息存储、多模态存储）和索引存储Index（多元索引）两部分，如果可以充分发挥云上存储的能力，将CommitLog和Index配置在云端的文件系统就可以天然的实现存储和计算分离。 （4）多模存储支持 满足不同基础场景下的高可用诉求； 充分利用云上基础设施，降低成本； （5）云原生基础设施： 可观测性能力云原生化，OpenTelemetry 标准化； Kubernetes 一键式部署扩容交付。 RocketMQ 5.02021年度大事件及未来规划 3、RocketMQConnector a、传统数据流 （1）传统数据流的弊端 生产者消费者代码需要自己实现，成本高； 数据同步的任务没有统一管理； 重复开发，代码质量参差不齐； （2）解决方案：RocketMQ Connector 合作共建，复用数据同步任务代码； 统一的管理调度，提高资源利用率； b、RocketMQ Connector数据同步流程 相比传统数据流，RocketMQ connector数据流的不同在于将 source 和 sink 进行统一管理，同时它开放源码，社区也很活跃。 4、RocketMQ Connector架构 如上图所示，RocketMQ Connector架构主要包含Runtime和Worker两部分，另外还有生态Source&Sink。 （1）标准：OpenMessaging （2）生态：支持ActiveMQ、Cassandra、ES、JDBC、JMS、MongoDB、Kafka、RabbitMQ、Mysql、Flume、Hbase、Redis等大数据领域的大部分产品； （3）组件：Manager统一管理调度，如果有多个任务可以将所有任务统一进行负载均衡，均匀的分配到不同Worker上，同时Worker可以进行横向扩容。 5、RocketMQ Stream RocketMQ Stream是一款将计算层压缩到一层的产品。它支持一些常见的算子如window、join、维表，兼容Flink SQL、UDF/UDAF/UDTF。 Apache Hudi Hudi 是一个流式数据湖平台，支持对海量数据快速更新。内置表格式，支持事务的存储层、一系列表服务、数据服务（开箱即用的摄取工具）以及完善的运维监控工具。Hudi 可以将存储卸载到阿里云上的 OSS、AWS 的S3这些存储上。 Hudi的特性包括： 事务性写入，MVCC/OCC并发控制； 对记录级别的更新、删除的原生支持； 面向查询优化：小文件自动管理，针对增量拉取优化的设计，自动压缩、聚类以优化文件布局； Apache Hudi是一套完整的数据湖平台。它的特点有： 各模块紧密集成，自我管理； 使用 Spark、Flink、Java 写入； 使用 Spark、Flink、Hive、Presto、Trino、Impala、 AWS Athena/Redshift等进行查询； 进行数据操作的开箱即用工具/服务。 Apache Hudi主要针对以下三类场景进行优化： 1、流式处理栈 (1) 增量处理； (2) 快速、高效； (3) 面向行； (4) 未优化扫描； 2、批处理栈 (1) 批量处理； (2) 低效； (3) 扫描、列存格式； 3、增量处理栈 (1) 增量处理； (2) 快速、高效； (3) 扫描、列存格式。 构建 Lakehouse 实操 该部分只介绍主流程和实操配置项，本机搭建的实操细节可以参考附录部分。 1、准备工作 RocketMQ version：4.9.0 rocketmqconnecthudi version：0.0.1SNAPSHOT Hudi version：0.8.0 2、构建RocketMQHudiconnector (1) 下载： _  git clone _ (2) 配置： /data/lakehouse/rocketmqexternals/rocketmqconnect/rocketmqconnectruntime/target/distribution/conf/connect.conf 中connectorplugin 路径 (3) 编译： cd rocketmqexternals/rocketmqconnecthudi mvn clean install DskipTest U rocketmqconnecthudi0.0.1SNAPSHOTjarwithdependencies.jar就是我们需要使用的rocketmqhudiconnector 3、运行 (1) 启动或使用现有的RocketMQ集群，并初始化元数据Topic： connectorclustertopic （集群信息） connectorconfigtopic （配置信息） connectoroffsettopic （sink消费进度） connectorpositiontopic （source数据处理进度 并且为了保证消息有序，每个topic可以只建一个queue） (2) 启动RocketMQ connector运行时 cd /data/lakehouse/rocketmqexternals/rocketmqconnect/rocketmqconnectruntime sh ./run_worker.sh Worker可以启动多个 (3) 配置并启动RocketMQhudiconnector任务 请求RocketMQ connector runtime创建任务 curl http://{runtimeip}:{runtimeport}/connectors/{rocketmqhudisinkconnectorname} ?config='{"connectorclass":"org.apache.rocketmq.connect.hudi.connector.HudiSinkConnector","topicNames":"topicc","tablePath":"file:///tmp/hudi_connector_test","tableName":"hudi_connector_test_table","insertShuffleParallelism":"2","upsertShuffleParallelism":"2","deleteParallelism":"2","sourcerecordconverter":"org.apache.rocketmq.connect.runtime.converter.RocketMQConverter","sourcerocketmq":"127.0.0.1:9876","srccluster":"DefaultCluster","refreshinterval":"10000","schemaPath":"/data/lakehouse/config/user.avsc"\}’ 启动成功会打印如下日志： 20210906 16:23:14 INFO pool2thread1 Open HoodieJavaWriteClient successfully (4) 此时向source topic生产的数据会自动写入到1Hudi对应的table中，可以通过Hudi的api进行查询。 4、配置解析 (1) RocketMQ connector需要配置RocketMQ集群信息和connector插件位置，包含：connect工作节点id标识workerid、connect服务命令接收端口httpPort、rocketmq集群namesrvAddr、connect本地配置储存目录storePathRootDir、connector插件目录pluginPaths 。 RocketMQ connector配置表 (2) Hudi任务需要配置Hudi表路径tablePath和表名称tableName，以及Hudi使用的Schema文件。 Hudi任务配置表 _点击__即可查看Lakehouse构建实操视频_ 附录：在本地Mac系统构建Lakehouse demo 涉及到的组件：rocketmq、rocketmqconnectorruntime、rocketmqconnecthudi、hudi、hdfs、avro、sparkshell0、启动hdfs 下载hadoop包 cd /Users/osgoo/Documents/hadoop2.10.1 vi coresite.xml fs.defaultFS hdfs://localhost:9000 vi hdfssite.xml dfs.replication 1 ./bin/hdfs namenode format ./sbin/startdfs.sh jps 看下namenode,datanode lsof i:9000 ./bin/hdfs dfs mkdir p /Users/osgoo/Downloads 1、启动rocketmq集群，创建rocketmqconnector内置topic QickStart：https://rocketmq.apache.org/docs/quickstart/ sh mqadmin updatetopic t connectorclustertopic n localhost:9876 c DefaultCluster sh mqadmin updatetopic t connectorconfigtopic n localhost:9876 c DefaultCluster sh mqadmin updatetopic t connectoroffsettopic n localhost:9876 c DefaultCluster sh mqadmin updatetopic t connectorpositiontopic n localhost:9876 c DefaultCluster 2、创建数据入湖的源端topic，testhudi1 sh mqadmin updatetopic t testhudi1 n localhost:9876 c DefaultCluster 3、编译rocketmqconnecthudi0.0.1SNAPSHOTjarwithdependencies.jar cd rocketmqconnecthudi mvn clean install DskipTest U 4、启动rocketmqconnector runtime 配置connect.conf workerId=DEFAULT_WORKER_1 storePathRootDir=/Users/osgoo/Downloads/storeRoot Http port for user to access REST API httpPort=8082 Rocketmq namesrvAddr namesrvAddr=localhost:9876 Source or sink connector jar file dir,The default value is rocketmqconnectsample pluginPaths=/Users/osgoo/Downloads/connectorplugins 拷贝 rocketmqhudiconnector.jar 到 pluginPaths=/Users/osgoo/Downloads/connectorplugins sh run_worker.sh 5、配置入湖config curl http://localhost:8082/connectors/rocketmqconnecthudi?config='\{"connectorclass":"org.apache.rocketmq.connect.hudi.connector.HudiSinkConnector","topicNames":"testhudi1","tablePath":"hdfs://localhost:9000/Users/osgoo/Documents/basepath7","tableName":"t7","insertShuffleParallelism":"2","upsertShuffleParallelism":"2","deleteParallelism":"2","sourcerecordconverter":"org.apache.rocketmq.connect.runtime.converter.RocketMQConverter","sourcerocketmq":"127.0.0.1:9876","sourcecluster":"DefaultCluster","refreshinterval":"10000","schemaPath":"/Users/osgoo/Downloads/user.avsc"\}' 6、发送消息到testhudi1 7、 利用spark读取 cd /Users/osgoo/Downloads/spark3.1.2binhadoop3.2/bin ./sparkshell \ packages org.apache.hudi:hudispark3bundle_2.12:0.9.0,org.apache.spark:sparkavro_2.12:3.0.1 \ conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' import org.apache.hudi.QuickstartUtils._ import scala.collection.JavaConversions._ import org.apache.spark.sql.SaveMode._ import org.apache.hudi.DataSourceReadOptions._ import org.apache.hudi.DataSourceWriteOptions._ import org.apache.hudi.config.HoodieWriteConfig._ val tableName = "t7" val basePath = "hdfs://localhost:9000/Users/osgoo/Documents/basepath7" val tripsSnapshotDF = spark. read. format("hudi"). load(basePath + "/") tripsSnapshotDF.createOrReplaceTempView("hudi_trips_snapshot") spark.sql("select from hudi_trips_snapshot").show()

在 Spring 生态中玩转 RocketMQ 系列教程现已登陆知行动手实验室，立即体验！ 移动端同学，需要在PC端登录 start.aliyun.com 进行体验。 RocketMQ 作为业务消息的首选，在消息和流处理领域被广泛应用。而微服务生态 Spring 框架也是业务开发中最受欢迎的框架，两者的完美契合使得 RocketMQ 成为 Spring Messaging 实现中最受欢迎的消息实现。本文展示了 5 种在 Spring 生态中文玩转 RocketMQ 的方式，并描述了每个项目的特点和使用场景。 一、前言 上世纪 90 年代末，随着 Java EE（Enterprise Edition）的出现，特别是 Enterprise Java Beans 的使用需要复杂的描述符配置和死板复杂的代码实现，增加了广大开发者的学习曲线和开发成本，由此基于简单的 XML 配置和普通 Java 对象（Plain Old Java Objects）的 Spring 技术应运而生，依赖注入（Dependency Injection），控制反转（Inversion of Control）和面向切面编程（AOP）的技术更加敏捷地解决了传统 Java 企业及版本的不足。随着 Spring 的持续演进，基于注解（Annotation）的配置逐渐取代了 XML 文件配置。除了依赖注入、控制翻转、AOP 这些技术，Spring 后续衍生出 AMQP、Transactional、Security、Batch、Data Access 等模块，涉及开发的各个领域。 2014 年 4 月 1 日，Spring Boot 1.0.0 正式发布。它基于“约定大于配置”（Convention over configuration)这一理念来快速地开发，测试，运行和部署 Spring 应用，并能通过简单地与各种启动器(如springbootwebstarter)结合，让应用直接以命令行的方式运行，不需再部署到独立容器中。Spring Boot 的出现可以说是 Spring 框架的第二春，它不但简化了开发的流程，目前更是事实标准。下面这幅图可以看出相同功能的 Spring 和 Spring Boot 的代码实现对比。 Apache RocketMQ 是一款是业界知名的分布式消息和流处理中间件，它主要功能是消息分发、异步解耦、削峰填谷等。RocketMQ 是一款金融级消息及流数据平台，RocketMQ 在交易、支付链路上用的很多，主要是对消息链路质量要求非常高的场景，能够支持万亿级消息洪峰。RocketMQ 在业务消息中被广泛应用，并衍生出顺序消息、事务消息、延迟消息等匹配各类业务场景的特殊消息。 本文的主角就是 Spring 和 RocketMQ，那几乎每个 Java 程序员都会使用 Spring 框架与支持丰富业务场景的 RocketMQ 会碰撞出怎么样的火花？ 二、RocketMQ 与 Spring 的碰撞 在介绍 RocketMQ 与 Spring 故事之前，不得不提到 Spring 中的两个关于消息的框架，Spring Messaging 和 Spring Cloud Stream。它们都能够与 Spring Boot 整合并提供了一些参考的实现。和所有的实现框架一样，消息框架的目的是实现轻量级的消息驱动的微服务，可以有效地简化开发人员对消息中间件的使用复杂度，让系统开发人员可以有更多的精力关注于核心业务逻辑的处理。 1. Spring Messaging Spring Messaging 是 Spring Framework 4 中添加的模块，是 Spring 与消息系统集成的一个扩展性的支持。它实现了从基于 JmsTemplate 的简单的使用 JMS 接口到异步接收消息的一整套完整的基础架构，Spring AMQP 提供了该协议所要求的类似的功能集。在与 Spring Boot 的集成后，它拥有了自动配置能力，能够在测试和运行时与相应的消息传递系统进行集成。单纯对于客户端而言，Spring Messaging 提供了一套抽象的 API 或者说是约定的标准，对消息发送端和消息接收端的模式进行规定，比如消息 Messaging 对应的模型就包括一个消息体 Payload 和消息头 Header。不同的消息中间件提供商可以在这个模式下提供自己的 Spring 实现：在消息发送端需要实现的是一个 XXXTemplate 形式的 Java Bean，结合 Spring Boot 的自动化配置选项提供多个不同的发送消息方法；在消息的消费端是一个 XXXMessageListener 接口（实现方式通常会使用一个注解来声明一个消息驱动的 POJO），提供回调方法来监听和消费消息，这个接口同样可以使用 Spring Boot 的自动化选项和一些定制化的属性。 在 Apache RocketMQ 生态中，RocketMQSpringBootStarter（下文简称 RocketMQSpring）就是一个支持 Spring Messaging API 标准的项目。该项目把 RocketMQ 的客户端使用 Spring Boot 的方式进行了封装，可以让用户通过简单的 annotation 和标准的 Spring Messaging API 编写代码来进行消息的发送和消费，也支持扩展出 RocketMQ 原生 API 来支持更加丰富的消息类型。在 RocketMQSpring 毕业初期，RocketMQ 社区同学请 Spring 社区的同学对 RocketMQSpring 代码进行 review，引出一段罗美琪（RocketMQ）和春波特（Spring Boot）故事的佳话[1]，著名 Spring 布道师 Josh Long 向国外同学介绍如何使用 RocketMQSpring 收发消息[2]。RocketMQSpring 也在短短两年时间超越 SpringKafka 和 SpringAMQP（注:两者均由 Spring 社区维护），成为 Spring Messaging 生态中最活跃的消息项目。 2. Spring Cloud Stream Spring Cloud Stream 结合了 Spring Integration 的注解和功能，它的应用模型如下： Spring Cloud Stream 框架中提供一个独立的应用内核，它通过输入(@Input)和输出(@Output)通道与外部世界进行通信，消息源端(Source)通过输入通道发送消息，消费目标端(Sink)通过监听输出通道来获取消费的消息。这些通道通过专用的 Binder 实现与外部代理连接。开发人员的代码只需要针对应用内核提供的固定的接口和注解方式进行编程，而不需要关心运行时具体的 Binder 绑定的消息中间件。 在运行时，Spring Cloud Stream 能够自动探测并使用在 classpath 下找到的 Binder。这样开发人员可以轻松地在相同的代码中使用不同类型的中间件：仅仅需要在构建时包含进不同的 Binder。在更加复杂的使用场景中，也可以在应用中打包多个 Binder 并让它自己选择 Binder，甚至在运行时为不同的通道使用不同的 Binder。 Binder 抽象使得 Spring Cloud Stream 应用可以灵活的连接到中间件，加之 Spring Cloud Stream 使用利用了 Spring Boot 的灵活配置配置能力，这样的配置可以通过外部配置的属性和 Spring Boot 支持的任何形式来提供（包括应用启动参数、环境变量和 application.yml 或者 application.properties 文件），部署人员可以在运行时动态选择通道连接 destination（例如，RocketMQ 的 topic 或者 RabbitMQ 的 exchange）。 Spring Cloud Stream 屏蔽了底层消息中间件的实现细节，希望以统一的一套 API 来进行消息的发送/消费，底层消息中间件的实现细节由各消息中间件的 Binder 完成。Spring 官方实现了 Rabbit binder 和 Kafka Binder。Spring Cloud Alibaba 实现了 RocketMQ Binder[3]，其主要实现原理是把发送消息最终代理给了 RocketMQSpring 的 RocketMQTemplate，在消费端则内部会启动 RocketMQSpring Consumer Container 来接收消息。以此为基础，Spring Cloud Alibaba 还实现了 Spring Cloud Bus RocketMQ， 用户可以使用 RocketMQ 作为 Spring Cloud 体系内的消息总线，来连接分布式系统的所有节点。通过 Spring Cloud Stream RocketMQ Binder，RocketMQ 可以与 Spring Cloud 生态更好的结合。比如与 Spring Cloud Data Flow、Spring Cloud Funtion 结合，让 RocketMQ 可以在 Spring 流计算生态、Serverless(FaaS) 项目中被使用。 如今 Spring Cloud Stream RocketMQ Binder 和 Spring Cloud Bus RocketMQ 做为 Spring Cloud Alibaba 的实现已登陆 Spring 的官网[4]，Spring Cloud Alibaba 也成为 Spring Cloud 最活跃的实现。 三、如何在 Spring 生态中选择 RocketMQ 实现？ 通过介绍 Spring 中的消息框架，介绍了以 RocketMQ 为基础与 Spring 消息框架结合的几个项目，主要是 RocketMQSpring、Spring Cloud Stream RocketMQ Binder、Spring Cloud Bus RocketMQ、Spring Data Flow 和 Spring Cloud Function。它们之间的关系可以如下图表示。 如何在实际业务开发中选择相应项目进行使用？下面分别列出每个项目的特点和使用场景。 1. RocketMQSpring 特点： 作为起步依赖，简单引入一个包就能在 Spring 生态用到 RocketMQ 客户端的所有功能。 利用了大量自动配置和注解简化了编程模型，并且支持 Spring Messaging API。 与 RocketMQ 原生 Java SDK 的功能完全对齐。 使用场景： 适合在 Spring Boot 中使用 RocketMQ 的用户，希望能用到 RocketMQ 原生 java 客户端的所有功能，并通过 Spring 注解和自动配置简化编程模型。 2. Spring Cloud Stream RocketMQ Binder 特点： 屏蔽底层 MQ 实现细节，上层 Spring Cloud Stream 的 API 是统一的。如果想从 Kafka 切到 RocketMQ，直接改个配置即可。 与 Spring Cloud 生态整合更加方便。比如 Spring Cloud Data Flow，这上面的流计算都是基于 Spring Cloud Stream；Spring Cloud Bus 消息总线内部也是用的 Spring Cloud Stream。 Spring Cloud Stream 提供的注解，编程体验都是非常棒。 使用场景： 在代码层面能完全屏蔽底层消息中间件的用户，并且希望能项目能更好的接入 Spring Cloud 生态（Spring Cloud Data Flow、Spring Cloud Funtcion 等）。 3. Spring Cloud Bus RocketMQ 特点： 将 RocketMQ 作为事件的“传输器”，通过发送事件（消息）到消息队列上，从而广播到订阅该事件（消息）的所有节点上，完成事件的分发和通知。 使用场景： 在 Spring 生态中希望用 RocketMQ 做消息总线的用户，可以用在应用间事件的通信，配置中心客户端刷新等场景。 4. Spring Cloud Data Flow 特点： 以 Source/Processor/Sink 组件进行流式任务处理。RocketMQ 作为流处理过程中的中间存储组件。 使用场景： 流处理，大数据处理场景。 5. Spring Cloud Function 特点： 消息的消费/生产/处理都是一次函数调用，融合 Java 生态的 Function 模型。 使用场景： Serverless 场景。 本文整体介绍了在 Spring 生态中接入 RockeMQ 的 5 种方法，让各位开发者对几种经典场景有宏观的了解。后续会有专栏详细介绍上述各个项目的具体使用方法和应用场景，真正地在 Spring 生态中玩转 RocketMQ！ 在 Spring 生态中玩转 RocketMQ 系列教程现已登陆知行动手实验室，立即体验！ 移动端同学，需要在PC端登录 start.aliyun.com 进行体验。 相关链接：