RocketMQ 消息发送流程与 Broker 故障延迟机制解析
消息发送的容错思路
Producer 向 Topic 发送消息时，RocketMQ 客户端需要在多个 Broker 的队列之间进行选择。如果某个 Broker 因为网络抖动、GC 暂停或自身负载过高而响应变慢，反复向其发送消息会加剧问题。为此，RocketMQ 在消息发送链路中提供了两种主要容错手段：
- 发送失败后的重试机制；
- Broker 故障延迟机制，即把近期表现不佳的 Broker 暂时从候选队列中剔除，等待一段"隔离时间"后再重新尝试。
隔离时间的长度与发送耗时正相关：发送越慢，被"冷落"的时间越长。下面重点分析故障延迟机制的实现以及消息从客户端到 Broker 的完整发送链路。
故障延迟机制的核心实现
故障延迟机制由 LatencyFaultTolerance 接口及其实现类负责。每个 Broker 对应一条故障记录，保存当前延迟和恢复可用的时间戳。
class BrokerFaultRecord implements Comparable<BrokerFaultRecord> {
private final String brokerName;
private long currentLatency;
private long availableTime;
public BrokerFaultRecord(String brokerName) {
this.brokerName = brokerName;
}
public boolean isAvailable() {
return System.currentTimeMillis() >= this.availableTime;
}
@Override
public int compareTo(BrokerFaultRecord other) {
return Long.compare(this.availableTime, other.availableTime);
}
// getter / setter ...
}
管理类负责更新记录、判断可用性以及在多条故障记录中挑选相对"较轻"的 Broker。
public class FaultToleranceManager implements LatencyFaultTolerance<String> {
private final ConcurrentHashMap<String, BrokerFaultRecord> records =
new ConcurrentHashMap<>();
private final AtomicInteger worstSelector = new AtomicInteger(0);
@Override
public void updateFaultItem(String brokerName, long currentLatency,
long notAvailableDuration) {
long now = System.currentTimeMillis();
BrokerFaultRecord old = records.get(brokerName);
if (old == null) {
BrokerFaultRecord record = new BrokerFaultRecord(brokerName);
record.setCurrentLatency(currentLatency);
record.setAvailableTime(now + notAvailableDuration);
old = records.putIfAbsent(brokerName, record);
if (old != null) {
old.setCurrentLatency(currentLatency);
old.setAvailableTime(now + notAvailableDuration);
}
} else {
old.setCurrentLatency(currentLatency);
old.setAvailableTime(now + notAvailableDuration);
}
}
@Override
public boolean isAvailable(String brokerName) {
BrokerFaultRecord record = records.get(brokerName);
return record == null || record.isAvailable();
}
@Override
public String pickOneAtLeast() {
List<BrokerFaultRecord> list = new ArrayList<>(records.values());
if (list.isEmpty()) {
return null;
}
Collections.shuffle(list);
Collections.sort(list);
int half = list.size() / 2;
if (half <= 0) {
return list.get(0).getBrokerName();
}
int index = worstSelector.getAndIncrement() % half;
return list.get(index).getBrokerName();
}
@Override
public void remove(String brokerName) {
records.remove(brokerName);
}
}
pickOneAtLeast 先对记录做一次随机化再排序，随后在前一半恢复时间较短的 Broker 中轮询选择。这样既优先使用"问题较小"的 Broker，又通过轮询避免多个 Producer 线程扎堆。
队列选择逻辑
Producer 真正发送前，会先调用队列选择策略。下面是一段重构后的队列选择代码，变量名和结构均做了调整，但逻辑保持不变。
public MessageQueue selectOneMessageQueue(TopicPublishInfo topicInfo, String lastBroker) {
if (!this.sendLatencyFaultEnable) {
return topicInfo.selectOneMessageQueue(lastBroker);
}
int startIndex = topicInfo.getSendWhichQueue().getAndIncrement();
List<MessageQueue> queueList = topicInfo.getMessageQueueList();
for (int i = 0; i < queueList.size(); i++) {
int pos = Math.abs(startIndex + i) % queueList.size();
if (pos < 0) {
pos = 0;
}
MessageQueue candidate = queueList.get(pos);
if (!this.faultTolerance.isAvailable(candidate.getBrokerName())) {
continue;
}
if (lastBroker == null || lastBroker.equals(candidate.getBrokerName())) {
return candidate;
}
}
String fallbackBroker = this.faultTolerance.pickOneAtLeast();
int writeQueueNums = topicInfo.getQueueIdByBroker(fallbackBroker);
if (writeQueueNums > 0) {
MessageQueue fallback = topicInfo.selectOneMessageQueue();
fallback.setBrokerName(fallbackBroker);
fallback.setQueueId(
topicInfo.getSendWhichQueue().getAndIncrement() % writeQueueNums
);
return fallback;
}
if (fallbackBroker != null) {
this.faultTolerance.remove(fallbackBroker);
}
return topicInfo.selectOneMessageQueue();
}
这段逻辑的含义是：先轮询所有队列，跳过被标记为不可用的 Broker；如果所有队列都不合适，则从故障记录中挑选一个相对可靠的 Broker，临时构造一个目标队列；如果连回退对象也无法使用，则清空该记录并回到默认选择策略。
发送延迟的计算
发送结束后，Producer 会根据本次发送耗时更新故障记录。下面给出经过重构的延迟计算与更新逻辑。
public class SendFaultStrategy {
private final LatencyFaultTolerance<String> faultTolerance;
private final long[] latencyThresholds =
{50L, 100L, 550L, 1000L, 2000L, 3000L, 15000L};
private final long[] isolateDurations =
{0L, 0L, 30000L, 60000L, 120000L, 180000L, 600000L};
public void updateFaultItem(String brokerName, long sendCost, boolean isolation) {
if (!this.sendLatencyFaultEnable) {
return;
}
long actualLatency = isolation ? 30000 : sendCost;
long duration = computeNotAvailableDuration(actualLatency);
this.faultTolerance.updateFaultItem(brokerName, actualLatency, duration);
}
private long computeNotAvailableDuration(long latency) {
for (int i = latencyThresholds.length - 1; i >= 0; i--) {
if (latency >= latencyThresholds[i]) {
return isolateDurations[i];
}
}
return 0L;
}
}
当发送耗时达到 550 ms 时，Broker 会被隔离 30 秒；耗时越高，隔离时间越长。若 isolation 为 true，则按 30 秒计算延迟，用于明确判定 Broker 不可用的场景。
消息发送的核心流程
以异步发送为例，客户端代码通常如下：
DefaultMQProducer producer = new DefaultMQProducer("AsyncProducerGroup");
producer.setRetryTimesWhenSendAsyncFailed(0);
producer.start();
for (int i = 0; i < messageCount; i++) {
byte[] body = ("hello-" + i).getBytes(StandardCharsets.UTF_8);
Message msg = new Message("TopicDemo", "TagA", "Key" + i, body);
final int seq = i;
producer.send(msg, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
System.out.printf("%d OK %s%n", seq, sendResult.getMsgId());
}
@Override
public void onException(Throwable e) {
System.out.printf("%d Exception %s%n", seq, e.getMessage());
}
});
}
最终调用链会进入 DefaultMQProducerImpl#sendKernelImpl。该方法负责定位 Broker 地址、生成消息唯一 ID、压缩消息体、执行发送钩子，并构造 SendMessageRequestHeader，随后根据同步/异步/单向模式交给 MQClientAPIImpl 发送。下面是一段高度抽象后的核心流程代码：
private SendResult sendKernelImpl(Message msg, MessageQueue mq,
CommunicationMode mode, SendCallback callback,
TopicPublishInfo topicInfo, long timeout) {
String brokerAddr = locateBrokerAddress(mq.getBrokerName());
brokerAddr = MixAll.brokerVIPChannel(
this.defaultMQProducer.isSendMessageWithVIPChannel(), brokerAddr);
prepareMessage(msg);
SendMessageRequestHeader header = buildRequestHeader(msg, mq);
switch (mode) {
case ASYNC:
return remotingClient.sendMessage(
brokerAddr, mq.getBrokerName(), msg, header,
timeout, mode, callback, topicInfo, ...);
case ONEWAY:
case SYNC:
return remotingClient.sendMessage(
brokerAddr, mq.getBrokerName(), msg, header,
timeout, mode, ...);
default:
throw new UnsupportedOperationException();
}
}
private String locateBrokerAddress(String brokerName) {
String addr = this.mQClientFactory.findBrokerAddressInPublish(brokerName);
if (addr == null) {
tryToFindTopicPublishInfo(brokerName);
addr = this.mQClientFactory.findBrokerAddressInPublish(brokerName);
}
return addr;
}
private SendMessageRequestHeader buildRequestHeader(Message msg, MessageQueue mq) {
SendMessageRequestHeader header = new SendMessageRequestHeader();
header.setTopic(msg.getTopic());
header.setQueueId(mq.getQueueId());
header.setProducerGroup(this.defaultMQProducer.getProducerGroup());
header.setBornTimestamp(System.currentTimeMillis());
header.setProperties(MessageDecoder.messageProperties2String(msg.getProperties()));
// ...
return header;
}
这里的 remotingClient 实际为 NettyRemotingClient，它把请求封装为 RequestCode.SEND_MESSAGE 命令发送到 Broker。
Broker 接收与存储
Broker 端由 SendMessageProcessor 处理 SEND_MESSAGE 请求。其主要流程是先做权限与 Topic 校验，再构造内部消息对象，最后调用存储引擎写入。
public class SendMessageProcessor {
private RemotingCommand handleSendMessage(ChannelHandlerContext ctx,
RemotingCommand request,
SendMessageRequestHeader header) {
RemotingCommand response = createResponse(request);
preCheck(ctx, header, response);
if (response.getCode() != ResponseCode.SUCCESS) {
return response;
}
TopicConfig topicConfig =
this.brokerController.getTopicConfigManager()
.selectTopicConfig(header.getTopic());
int queueId = resolveQueueId(header.getQueueId(), topicConfig);
MessageExtBrokerInner inner = buildInnerMessage(ctx, request, header, queueId);
if (this.brokerController.getBrokerConfig().isRejectTransactionMessage()
&& isTransactionMessage(inner)) {
response.setCode(ResponseCode.NO_PERMISSION);
return response;
}
PutMessageResult putResult =
this.brokerController.getMessageStore().putMessage(inner);
return assembleSendResult(putResult, response, inner, queueId);
}
}
其中 preCheck 包括：
- 检查 Broker 写权限；
- 检查 Topic 是否允许发送、是否存在，不存在时自动创建；
- 检查队列编号是否合法。
校验通过后，消息被转换为 MessageExtBrokerInner，随后进入 DefaultMessageStore#putMessage 持久化。写入成功后，Broker 返回 SendResult，Producer 根据结果决定是否触发重试或更新故障延迟记录。
完整链路
将以上步骤串联起来，一条消息从客户端到 Broker 的完整流程如下：
- 调用
DefaultMQProducer#send； DefaultMQProducerImpl选择目标队列，期间利用故障延迟机制剔除不可用 Broker；- 调用
sendKernelImpl组装请求头，通过NettyRemotingClient发送命令； - Broker 端的
SendMessageProcessor校验请求并调用存储引擎写入消息； - Producer 收到结果后，根据发送耗时更新故障延迟记录。
