本文来自『中间件兴趣圈』公众号，仅作技术交流，未授权任何商业行为。 11 > 1.1 RocketMQ 核心概念扫盲篇 3. Client 消息客户端，包括 Producer(消息发送者)和 Consumer(消费消费者)．客户端在同一 时间只会连接一台 nameserver，只有在连接出现异常时才会向尝试连接另外一台。客户 端每隔 30s 向 Nameserver 发起 topic 的路由信息查询。 order_consumer 来订阅 order_topic，并且基于分布式部署， order-service-app 的部署情况如下： 本文来自『中间件兴趣圈』公众号，仅作技术交流，未授权任何商业行为。 1.1 RocketMQ 核心概念扫盲篇 < 12 即 order-service-app 部署了 3 台服务器，每一个 jvm 进程可以看做是消费组 order_consumer 消费组的其中一个消费者。 ConsumerGroup: consumer-grouptest The consumerId: c onsumer-client-id-clusterA-192.168.3.122@21932 not in cidAll: [consumer-client-id-clust erA-192.168.3.123@22164] 上面的日志显示在队列负载时候，当前节点竟然不属于 consumer-grouptest

设计演进及架构 • 服务发现 • 微服务调用 • Edge Service • Metrics • 性能调优 设计演进-初始 Transport RPC Servlet Rest Consumer 透明RPC RestTemplate Producer 透明RPC JAX-RS SpringMVC • 传输绑架开发模式 • 传输可能对业务代码产生严重的侵入 • 治理逻辑与 治理逻辑与业务代码或传输耦合，需要针对各种场景独立开发 治理 治理 治理 治理 设计演进 服务契约 (OpenAPI) Transport Highway Servlet Rest Vertx Rest Consumer 透明RPC RestTemplate Producer 透明RPC JAX-RS SpringMVC 以契约为核心，贯穿始终 • 开发模式(业务代码)、治理、传输三方解耦，互不感知 SpringBoot Starters Foundation javassist protobuf rest …. 透明RPC Jaxrs SpringMVC 透明RPC Jaxrs SpringMVC Consumer Handler Producer …… Transport 透明RPC RestTemplate LB QPS Trace RESTful Vertx Highway 透明RPC

知到 我，结果上线后直接导致生产事故。 5. 我和其他团队共用一套分布式缓存服务，为了防止数据覆盖， 我只能和别人轮流排队上线调试，我等了一天，结果只能用一 个小时。 6. 我写了个消息Consumer，布到线上准备测测，结果，别人的消 息先到了，一下把我的预置数据全搞乱了… 7. 我的开发机性能还行，我想自己装一套依赖环境提高开发调测 效率，结果消息服务、缓存服务、服务框架装完一启动，80% 干不了了，回家洗洗睡吧…. … ….. ……… 1 2 3 解决之道：通过技术手段，降低系统对外部的依赖，而“MOCK”，是最 有效的手段。 分布式服务框架mock能力构建 应用 服务容器（Consumer） Filter Chain 将mock能力内置到分布式服务框架中 开发mock能力过滤器，在服务调用链路上对服务调用进行挡截。 “开关机制”控制mock能力启停 mock能力启 ，同时还能提升Mock数据的质量。 1.定制现网数据抓取过滤器 2.定制抓取时间段 3.定制抓取数量 4.一服务一文件 5.一请求一记录 服务容器（Provider） Consumer Consumer 服务实现 服务实现 Accepter Filter Chain Filter1 线上数据 抓取过滤器 FilterN 抓取配置： 是否开启 抓取时长 抓取数量 落 盘

Java Chassis通信处理详解 通信优化实践 议题  问题不挑战  整体线程模型  Consumer  Producer 问题不挑战 消费者 编程 模型 透明 RPC Spring MVC Transport Highway Vertx REST Servlet REST 生产者 编程 模型 透明 •Consumer 整体线程模型 Producer Consumer Eventloop 连接 Eventloop 连接 …… 业务线程 业务线程 业务线程 业务线程 Eventloops Executor 线程 …… 线程 队列 Executor 线程 …… 线程 队列 …… Consumer 消费端，主要需要处理的问题是如何更 消费端，主要需要处理的问题是如何更 高效地将请求推送到对端去，然后拿到 应答信息。 透明RPC Consumer业务线程 Consumer Eventloop 连接 业务线程 hello.sayHello() Handlers.handle HttpClientFilters.beforeSendRequest (Http) Handlers.asyncResponse

default: Msg } chan Get: chan Select } 数据⼴播 Topic Channel Channel Producer Consumer Consumer Consumer Channel Topic 异步把消息⼴播给 Channel 数据⼴播 Topic ⾃⾝拥有 buffer 的能⼒ Memory Channel Disk Msg Pump() Msg chan *Msg Channel 2 Memory Channel Disk Queue 负载均衡 Consumer Consumer Consumer chan *Msg 多个 Consumer 同时通过同⼀个 chan 读取 Memory Channel Disk Queue Channel 消费反馈 InFight Client NSQd NSQd Server 2 NSQd NSQd Server 3 NSQ Lookupd NSQd 分布式架构 NSQ Lookupd NSQd Producer Consumer NSQ Lookupd Stateless 内部逻辑 协议解析 Client Server time IDENTIFY RDY 2 MSG 1 FIN 1 消费消息 MSG

Producer A Topic A Broker B Topic B ConsumerGroupA ConsumerGroupB Consumer 1 Consumer 2 Producer B Consumer 1 Consumer 2 MessageQueue1 MessageQueue2 MessageQueue1 MessageQueue2 部署模型 组件细节 In Memory except KV config info Netty RemotingServer topic register and heart beats Producer Consumer NRC BrokerAddrTlb send Msg topic route heatbeats MsgQueList callback/check NRC pull MsgExt | a5 - a4+1 | h3 doDispatch() Consumer config/consumerOffset.json commitOffset() ConsumerOffsetManager "offsetTable":{ "TopicTest@consumer-g1":{0:0,1:21,2:0}, ... } index

gz解压后的目录结构： . ├── LICENSE ├── NOTICE ├── README.md ├── benchmark │ ├── batchproducer.sh │ ├── consumer.sh │ ├── producer.sh │ ├── runclass.sh │ ├── shutdown.sh │ └── tproducer.sh ├── bin │ server是多台集群的话，就用分号分隔 namesrvAddr=192.168.166.20:9876;192.168.166.206:9876 # 每个topic对应队列的数量，默认为4，实际参考consumer实例的数量，值过小不利于consumer负 均衡 defaultTopicQueueNums=8 # 是否允许broker自动创建Topic，生产建议关闭 autoCreateTopicEnable=true # server是多台集群的话，就用分号分隔 namesrvAddr=192.168.166.20:9876;192.168.166.206:9876 # 每个topic对应队列的数量，默认为4，实际参考consumer实例的数量，值过小不利于consumer负 均衡 defaultTopicQueueNums=8 # 是否允许broker自动创建Topic，生产建议关闭 autoCreateTopicEnable=true #

egresses: - svc-not-in-mesh.nav.local secrets: true accessPolicy: inbound: - name: consumer-a app apiVersion: "nais.io/v1alpha1" kind: "Application" metadata: name: app labels: egresses: - svc-not-in-mesh.nav.local secrets: true accessPolicy: inbound: - name: consumer-a nais.yaml cluster kubectl apply -f nais.yaml application deployment service virtualservice egresses: - svc-not-in-mesh.nav.local secrets: true accessPolicy: inbound: - name: consumer-a Role RoleBinding apiVersion: "nais.io/v1alpha1" kind: "Application" metadata: name: app

and Tail pointers Dequeue Software Dequeue Software Enqueue Software Producer Consumer Consumer Consumer • Synchronization latency • Memory/Cache latency • CPU cycles latency DLB ： Dynamic Enqueue Logic Head and Tail pointers Dequeue Logic & Load Balancer Producer Producer Consumer Consumer Consumer • No Synchronization latency • No memory/cache latency • No CPU cycles DLB-Assist Channel Message-ID • Transfers Message-ID through HW Senders Receivers Channel Hardware Hardware Producer Consumer Give Ownership Take Ownership Message-ID Reclaiming Message- Ownership Tracking DLB-Assist

Server Group 1 Producer Y Consumer A1 消费组A Consumer A2 Consumer B1 消费组B Consumer B2 Topic A 分布式消息系统的部署模型 Producer Discovery Service Cluster Consumer Consumer Consumer Producer Producer Server