RocketMQ消息发送主要分为以下几步：

• 根据主题获取路由信息（主题对应的所有队列，队列所在broker，broker的IP地址和端口号）
• 选择其中一条队列
• 发送消息
• 成功返回，失败重试，同步和异步默认2次，单向发送不重试

详细流程

获取路由信息

public class TopicRouteData extends RemotingSerializable {
    private List<QueueData> queueDatas;
    private List<BrokerData> brokerDatas;
}

先从本地缓存获取，如果没有再从NameServer获取，从NameServer获取时会先尝试获取一个ReentrantLock，目的是防止并发请求，主题的路由信息更新并不频繁。另外生产者端还会每30s获取一次路由信息。

获取到路由信息后，还要跟本地缓存相比较，如果修改了，则更新发布和订阅者的路由信息。同一个进程可以启多个不同组的生产者和消费者，但客户端实例（MQClientInstance）是单例的

选择队列

首次随机选择一个队列，后面队列ID加一取模，并且不选择上一次的broker，通过ThreadLocal保证线程安全

public MessageQueue selectOneMessageQueue(final String lastBrokerName) {
    if (lastBrokerName == null) {
        return selectOneMessageQueue();
    } else {
        for (int i = 0; i < this.messageQueueList.size(); i++) {
            int index = this.sendWhichQueue.getAndIncrement();
            int pos = Math.abs(index) % this.messageQueueList.size();
            //当index=Integer.MIN_VALUE时，Math.abs是负数，仍为MIN_VALUE
            if (pos < 0)
                pos = 0;
            MessageQueue mq = this.messageQueueList.get(pos);
            if (!mq.getBrokerName().equals(lastBrokerName)) {
                return mq;
            }
        }
        return selectOneMessageQueue();
    }
}

失败延迟机制：每次发送成功或是异常，都会记录发送时间，并根据发送时间计算broker不可用时间戳，如果发送成功但是大于等于550ms，会标记该broker 30s、60s、120s、180s、600s不可用，如果发送失败，会标记该broker 15分钟不可用，这段时间内不再向该broker发送消息

public void updateFaultItem(final String brokerName, final long currentLatency, boolean isolation) {
    if (this.sendLatencyFaultEnable) {
        //如果发送成功，isolation是false，发送失败为true，会标记该broker 15分钟不可用
        long duration = computeNotAvailableDuration(isolation ? 30000 : currentLatency);
        this.latencyFaultTolerance.updateFaultItem(brokerName, currentLatency, duration);
    }
}

private long computeNotAvailableDuration(final long currentLatency) {
    for (int i = latencyMax.length - 1; i >= 0; i--) {
        //latencyMax = {50L, 100L, 550L, 1000L, 2000L, 3000L, 15000L};
        //notAvailableDuration = {0L, 0L, 30000L, 60000L, 120000L, 180000L, 600000L};
        if (currentLatency >= latencyMax[i])
            return this.notAvailableDuration[i];
    }
    return 0;
}