rocketm q-con n ect 进入孵化 2021 5.0 -p rev iew 发布 性能优化、PoP 消费，多存储目录， 轻量级队列 rocketm q-strea m s 轻量级实时计算 引擎发布 5.0 消息事件流融合处理平台 Sta r: 1 .7w C on tributor: 50 0 + RocketMQ 发展历程 W r i t e h e r e S o 普通云盘 ESSD 云盘 SA TA 独占/混部/独立交付…… • 集群节点异常成为常态 • 依赖服务随时可能在进行迁移或重启 • 对弹性的要求开始从物理资源变为逻辑资源 • IaaS 的多样性对应用交付部署提出了更高要求 • 可运维性、可观测性带来了更大挑战 • 多租环境带来了更高的网络及安全隔离要求 • 无限资源 vs 有限成本 • 冗长的请求链路，膨胀的技术栈 ……. 面向失败 多场景 云原生时代的挑战 云原生四要素 云原生时代的 RocketMQ admin 富客户端 轻量级 SD K API兼容 计算 云存储 VPC 网络 Kubernetes Name Server Name Server Name Server 计算集群 可观测 remoting gRPC MQTT AMQP AC L 消息 事件 Subscription 多租户 Store

，因此会 通过追踪源码来寻求答案，如果大家想急于答案，可以跳过问题分析，直接查看本文末尾的 问题解答部分。 通过本文的阅读，您将获得如下信息： 1. RocketMQ 消费 TPS 的收集与计算逻辑。 2. RocketMQ 监控指标的设计思路。 3. RocketMQ 主从同步，消费者从主服务器拉取还是从从服务器拉取的判断逻辑。 本文来自『中间件兴趣圈』公众号，仅作技术交流，未授权任何商业行为。 将数据返回给前段页面 进行展示。 接下将聚焦到消费组消费 TPS 的统计处理，其入口为 tpsGroupGetNums 。 2. rocketmq 消费 TPS 统计实现原理 消费 TPS 计算逻辑 首先我们还是从 tpsGroupGetNums 方法入手，探究一下 tps 的获取逻辑，然后再 探究数据的采集原理（这也是 rocketmq 监控相关）。 BrokerStatsManager#tpsGroupGetNums 代码@1：首先取快照中的第一条消息。 代码@2：取快照列表中的最后一条消息。 代码@3：计算这两个时间点 value 的差值，即这段时间内新增的总数。 代码@4：计算这段时间内的 tps，即每秒处理的消息条数。 代码@5：计算 avgpt ，即平均一次操作新增的消息条数（即平均一次操作，value 新增的个数）。 消费组的消费 TPS 的计算逻辑就介绍到这里了，那还有一个疑问，即 StatsItem 中 csListMinute

processQueue 的 lock 属性是否为 true，lock 属性是否过期，如果为 false 或者过期， 放到本地线程稍后锁定在消费。 如果 lock 为 true 且没有过期，开始消费消息 计算任务执行的时间如果大于一分钟且线程数小于队列数情况下，将 processqueue， messagequeue 重新构建 ConsumeRequest 加到线程池 10ms 后在消费，这样防止个别队列被 获取，此函数如果集合中一个也没有创建一个，如果最后一个写满了也创 建一个新的。 MapedFileQueue 在获取 getLastMapedFile 时，如果需要创建新的 MapedFile 会计算出下 一个 MapedFile 文件地址，通过预分配服务 AllocateMapedFileService 异步预创建下一个 MapedFile 文件，这样下次创建新文件请求就不要等待，因为创建文件特别是一个 indexCount 4 位 int 类型，索引文件中构建的索引个数 槽位 slot, 默认每个文件配置的 slot 个数为 500 万个， 每个 slot 是 4 位的 int 类型数据 计算消息的对应的 slotPos=Math.abs(keyHash)%hashSlotNum 消息在 IndexFile 中的偏移量 absSlotPos = IndexHeader.INDEX_HEADER_SIZE

computing storage separating architecture� 典型应⽤用场景 ⾦金金融交易易 电⼦子商务 智能制造 分布式事务 异步解耦 IoT/IIoT 决策分析 实时计算 概念模型 Broker A Producer A Topic A Broker B Topic B ConsumerGroupA ConsumerGroupB Consumer RocketMQ 存储设计 Cloud Native 的定义 • 设计，搭建和部署应⽤用的⽅方式 • 运⾏行行在云计算环境 IaaS • 快速，弹性 • 降低技术⻛风险 • 互操作性，易易部署，可监控管理理 消息系统挑战 • 传统设计和实现⽅方⾯面 • 组件的微服务化，例例如存储和计算分离 • 开源软件的冲击，云服务选型⾯面临挑战 • 缺少稳定的多语⾔言类库 • Cloud AI Ops⽅方⾯面

1. 根据消息从 DB 查询数据 1 2. 根据消息从 DB 查询数据 2 3. 复杂的业务计算 4. 吐 DB 揑入数据 3 5. 吐 DB 揑入数据 4 返条消息的消费过程不 DB 交互了 4 次，如果挄照每次 5ms 计算，那举总共耗时 20ms，假设业务计算耗时 5ms， 那举总过耗时 25ms，如果能把 4 次 DB 交互优化为 2 次，那举总耗时就可以优化到