RocketMQ-Producer

rocketmq 发送端的设计
1、需要考虑什么？

容错设计

• 如果投递到某个服务端失败了，在多主的情况下可以选择可用的服务端
• 服务端不可用后要进入熔断状态，服务不可用还继续提供服务会带来一系列的问题
• 服务端需要可以自动在熔断状态下恢复，服务可用但是长时间不能自动恢复

重试

• 可配置的重试策略（重试策略，异步消息不存在重试的语义）
• 重试支持选择新队列（在一个不可服务的队列上重试是一种资源的浪费，同时也是无效的重试）

2、消息投递需要的参数

• topic

原则是：根据topic选择相应的逻辑队列，逻辑队列中相应的指定了broker, topic

3、rocketmq究竟怎么做？

3.1、TopicPublishInfo
这个数据结构保存的是该topic下的所有的逻辑队列List<MessageQueue>，以及本次发起请求时选择队列的下标值ThreadLocalIndex（index & messagequeues.size 才是真实的下标值）。
PS：这里有点绕，为何是保存所有的topic下的逻辑队列？而不是按照broker先分组，每个broker下再分逻辑队列呢？

3.2、FaultItem
这个数据结构保存着broker name， 以及对应的隔离时间，具体要看isAvailable()。该数据接口维系在faultItemTable这个Map中（ConncurrentHashMap，保证并发安全）

3.3、失败之后做了什么？

this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, true); 

参数1是名称，参数2一次调用耗费的时间，参数2是是否隔离（隔离意味着熔断）

如果开启了容错保护机制，那么会根据是否隔离来计算broker的隔离时间。currentLatency + 30000 -> 并在notAvailableDuration = {0L, 0L, 30000L, 60000L, 120000L, 180000L, 600000L}这个数组中找到第一个大于该结果的值作为隔离的持续时间。

比如：currentLatency 0:30000，那么结果无疑就是60000，如果是30000:90000，那么结果就是120000，以此类推。所以隔离的持续时间其实是数组中的值。

PS: faultItem中记录的是隔离时间是：System.currentTimeMillis() + notAvailableDuration。而判断broker有效是：(System.currentTimeMillis() - startTimestamp) >= 0

3.4、选择队列的依据是什么？

随机下标（ThreadLocalIndex + 1）, 以messagequeues.size作为循环次数，找到可用的broker(判断可用的依据是如上)，如果找不到，则依据某个算法（随机打乱排序取二分之一位置的节点）在整个队列中选择一个队列（队列的id为随机值%broker可写队列长度）。

容错设计

1、第一次有地失败，并且该服务端被标记为隔离状态，时间参考notAvailableDuration值。
2、在这个持续时间内，该服务不会提供服务（运维这个时候需要去排查原因并重新启动）
3、超过该持续时间后，该节点会重新变为可用（但是不一定真实可用，如果不可用将会重新回到第1步）

重试

• 配置项：retryTimesWhenSendFailed（默认2次），超过该次数将不再重试
• 重试时将会按照上文说的策略进行筛选（ThreadLocalIndex对于当前线程有效地重试递增）

核心的代码包括：

DefaultMQProducerImpl#sendDefaultImpl()
DefaultMQProducerImpl#tryToFindTopicPublishInfo()
DefaultMQProducerImpl#selectOneMessageQueue()
DefaultMQProducerImpl#updateFaultItem()