RocketMq之Producer原理浅析

生产者是怎么发消息的？

MessageQueue

首先在理解生产者发消息之前，必须要明白一个概念：MessageQueue是什么？ 其实MessageQueue是RocketMq的一种数据分片+物理存储机制。

我们一般在创建 Topic 的时候会指定 MessageQueue 的数量。

如上图，一个 Topic 中有4个 MessageQueue，每个 Brokers 上有2个 MessageQueue ，生产者通过算法（默认是均匀分配）来把消息写入不同的 MessageQueue 中。MessageQueue 的数据可以持久化在磁盘上。

这样就把消息分散到了多个 Broker 上，大大提升 Broker 的抗并发能力！

Producer连接NameSever

Producer 通过 NameSever 获取指定 Topic 的 Broker 路由信息，并在本地保存一份缓存数据，比如一个Topic有哪些 MessageQueue，MessageQueue 在哪几台 Broker 上，Broker 的ip.port等等。 Producer 发送消息只发到 Master Broker上，Slave 通过 主从同步获取数据。

那么 Produce 是怎么链接 NameSever 的呢？

• 连接 单个生产者者和一台 Nameserver 保持长连接，定时查询topic配置信息，如果该nameserver挂掉，生产者会自动连接下一个nameserver，直到有可用连接为止，并能自动重连。
• 轮询时间 默认情况下，生产者每隔30秒从nameserver获取所有topic的最新队列情况，这意味着某个broker如果宕机，生产者最多要30秒才能感知，在此期间，发往该broker的消息发送失败。该时间由DefaultMQProducer的pollNameServerInteval参数决定，可手动配置。
• 心跳 与nameserver没有心跳

Producer连接Broker

• 连接 生产者 跟 Topic 涉及的所有Broker 保持长连接。
• 心跳 默认情况下，生产者每隔30秒向所有broker发送心跳。broker每隔10秒钟（此时间无法更改），扫描所有还存活的连接，若某个连接2分钟内（当前时间与最后更新时间差值超过2分钟，此时间无法更改）没有发送心跳数据，则关闭连接。

Producer是怎么发消息的

容错机制

Producer 作为发送消息的一方，有3中容错机制：

• 本地缓存 把从 NameSever 获取的信息缓存到本地，以防 NameSever 宕机
• 不可用Broker集合 Producer有一个 Broker 的容错机制，开关sendLatencyFaultEnable可以开启，RocketMq内部会维护一个故障Broker的HashMap，把一定延迟级别的Broker放入这个map，下次选择Broker的时候，就会规避不可用的Broker。
• 重试 Producer发送消息时，有一个重试机制，默认重试3次
• 死信队列 Consumer消费重试超过指定次数，进入死信队列

负载均衡

生产者通过轮询某个 Topic 下的所有 MessageQueue 的方式来实现发送方的负载均衡，如下图：

通过这种方式，可以将一个 Topic 的消息分散到多个 MessageQueue 上，进而分散到多个 Broker 上。

思考

NameSever宕机怎么办

如果与 Producer 连接的 NameSever 突然宕机，Producer 最长要30秒才能感知到，此时Producer 可以先从本地缓存读取 Topic 的路由信息。直到连接到下一个 NameSever，刷新本地缓存。

Broker宕机怎么办

如果与 Producer 连接的 Broker 突然宕机，比如 Master Broker 挂了，此时其他 Slave Broker 会选举出一台 Master Broker，但在这个过程中，Producer 发送的消息都会失败。

对于这个问题，在Producer中有一个开关sendLatencyFaultEnable，这个开启后会有一个容错机制，比如某次访问一个Broker有500ms延迟还无法访问，那么接下来就会回避访问该Broker一段时间，比如3000ms内不再访问该Broker，避免消息打到故障的Broker上。

此外， Producer 本身可以捕获发送异常，进行重试。