Producer 启动流程与路由获取

我们先从 Producer 的启动开始。一个 Producer 启动的时候它可不是傻乎乎地直接就开始发消息了——它得先弄清楚“把消息发到哪里去”。Producer 启动的核心步骤创建 Producer 实例设置 ProducerGroup 名称同一个业务使用同一个 Group设置 NameServer 地址列表指定集群注册中心的地址配置其他参数发送超时时间、重试次数等调用 start 方法启动 Producer初始化 MQClientInstance客户端核心实例负责网络通信和路由管理启动 Netty 通信客户端建立与 NameServer 的连接从 NameServer 拉取所有 Topic 的路由信息启动定时任务每 30 秒更新一次路由信息Producer 就绪等待发送消息关键点解析ProducerGroup一个业务标识同一个业务里的 Producer 实例归属于同一个 Group。在事务消息中同一个 Group 的 Producer 可以互相回查事务状态。MQClientInstance这是客户端最核心的实例所有 Producer 和 Consumer 共用一个 MQClientInstance同一个 JVM 里。它负责管理与 NameServer 的连接管理与 Broker 的连接维护本地路由缓存统一的心跳和网络通信路由获取Producer 启动时会从 NameServer 拉取所有 Topic 的路由信息不仅仅是某个特定 Topic这样后续发消息时就不需要再等路由查询了。定时更新路由信息在本地缓存后会有一个定时任务每隔 30 秒 从 NameServer 拉取最新路由保证路由信息的时效性。消息发送的三种方式同步、异步、单向在《入门认知篇》里我们简单提过三种发送方式现在我们从底层实现的角度再来看一遍。同步发送Sync这是最常用、最简单的方式。发送消息后线程阻塞等待 Broker 返回响应收到响应后才继续执行。SendResult sendResult producer.send(msg);// 阻塞等待直到收到 Broker 的响应BrokerProducerBrokerProducer线程阻塞等待线程恢复⏱️ 整个过程中线程处于阻塞状态构造消息发送请求同步返回响应成功/失败根据结果处理业务逻辑适用场景关键业务——比如下单成功后的订单消息必须确认 Broker 收到了才能继续。异步发送Async发送消息后线程不阻塞立即返回。等 Broker 响应回来后通过回调函数来处理结果。producer.sendAsync(msg, new SendCallback() {Overridepublic void onSuccess(SendResult sendResult) {// 处理成功逻辑}Overridepublic void onException(Throwable e) {// 处理失败逻辑}});// 这里立即返回不阻塞BrokerProducer 异步线程Producer 主线程BrokerProducer 异步线程Producer 主线程主线程立即返回可以继续其他工作✅ 主线程未被阻塞⏱️ 响应处理在异步线程中构造消息提交异步发送任务发送请求返回响应回调 SendCallback适用场景对延迟敏感但需要知道结果——比如前端请求发消息不能阻塞用户操作但失败时需要通知用户。单向发送Oneway只发不管连响应都不等。最轻量、最快但最不可靠。producer.sendOneway(msg);// 不管结果直接继续BrokerProducerBrokerProducer发完就走不等待任何响应 最快但没有任何可靠性保证构造消息发送请求单向继续执行后续代码适用场景日志上报、监控数据等——丢了就丢了业务上能接受。 小贴士同步发送和异步发送虽然看起来差异很大但在底层它们都使用了异步网络通信Netty。同步发送只是在异步通信的基础上用 CountDownLatch 等同步工具做了一个“阻塞等待”的封装。本质上RocketMQ 的网络通信模型是全异步的。消息发送的负载均衡策略当一个 Topic 有多个 MessageQueue分布在不同的 Broker 上时Producer 如何选择将消息发到哪个 Queue这就涉及负载均衡策略。RocketMQ 默认提供了两种内置策略你也可以自定义实现策略一轮询Round Robin—— 默认策略依次轮流选择 Queue确保消息均匀分布。在顺序消息场景下会结合消息 Key 做哈希保证同一个 Key 的消息落到同一个 Queue。MessageQueue消息序列第 1 条第 2 条第 3 条第 4 条第 5 条第 6 条消息 1消息 2消息 3消息 4消息 5消息 6Queue 0Queue 1Queue 2按顺序轮流分配1→Q1, 2→Q2, 3→Q3, 4→Q1…策略二一致性哈希Consistent Hash根据消息 Key 的哈希值通过哈希环来决定消息发到哪个 Queue。保证同一个 Key 的消息始终落在同一个 Queue 上对于顺序消息特别重要。消息 Key哈希值映射到环上哈希值映射到环上哈希值映射到环上顺时针查找第一个节点顺时针查找第一个节点顺时针查找第一个节点一致性哈希环节点 1Queue 0节点 2Queue 1节点 3Queue 2节点 4Queue 3节点 5Queue 0节点 6Queue 1order_123order_456order_789位置 A位置 B位置 C同一 Key 始终落到同一 Queue新增节点时只影响局部 Key 的重新分配自定义策略实现 MessageQueueSelector 接口根据自己的业务逻辑选择 Queue比如按订单 ID 取模。消息发送的重试机制与故障规避消息发送不是总能一次成功的。网络抖动、Broker 故障、磁盘满了……各种原因都可能导致发送失败。RocketMQ 有一套完善的重试和故障规避机制。重试机制是否否是是否是否发送消息第 1 次尝试成功返回成功是否为可重试的异常直接返回失败不重试第 2 次尝试等待 50ms成功… 最多重试 retryTimesWhenSendFailed 次最后一次成功