RocketMQ源码分析之Producer发送消息（五）

本节主要讲Producer同步发送消息的流程，异步的暂时不打算讲。因为还要讲Broker接受消息，存储消息，以及RemotingCommand。

老规矩电路图送上，为什么我喜欢电路图，因为UML比时序图或其他任何图都更能反映出类之间的调用。

发送消息时序图：下图带有浅绿色背景区域的部分为循环调用。retry send，你们懂得。

从时序图中可以看到整个发送消息都是围绕DefaultMQProducerImpl类展开的。

1、DefaultMQProducerImpl.sendDefaultImpl()方法，这里只贴出来主要代码

private SendResult sendDefaultImpl(Message msg, final CommunicationMode communicationMode,
        final SendCallback sendCallback, final long timeout ){
    this.makeSureStateOK();//判断服务是否是Running状态，不是抛异常出来
    Validators.checkMessage(msg, this.defaultMQProducer);//检查消息正确性，包括topic，消息大小等

    TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());//获取topic队列配置
    for (; times < timesTotal; times++) {
        MessageQueue mqSelected = this.selectOneMessageQueue(topicPublishInfo, lastBrokerName);//选择消息队列，后面详细说
        sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout - costTime);
    }
}

 2、DefaultMQProducerImpl.makeSureStateOK()方法

 这个方法就是检查服务状态是不是running，很多地方都会调用这个方法。

private void makeSureStateOK() throws MQClientException {
    if (this.serviceState != ServiceState.RUNNING) {
        throw new MQClientException("The producer service state not OK, "
                + this.serviceState
                + FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK), null);
    }
}

3、DefaultMQProducerImpl..tryToFindTopicPublishInfo()方法

获取Topic的队列配置
该方法主要是获取主题的队列信息，当从本地内存中获取不到时，从NameServer服务器获取主题配置信息。

private TopicPublishInfo tryToFindTopicPublishInfo(final String topic) {
    TopicPublishInfo topicPublishInfo = this.topicPublishInfoTable.get(topic);
    if (null == topicPublishInfo || !topicPublishInfo.ok()) {
        this.topicPublishInfoTable.putIfAbsent(topic, new TopicPublishInfo());
　　        //当从本地内存中获取不到时,从NameServer服务器获取主题的队列配置信息
        this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic);
        topicPublishInfo = this.topicPublishInfoTable.get(topic);
    }

    if (topicPublishInfo.isHaveTopicRouterInfo() || topicPublishInfo.ok()) {
        return topicPublishInfo;
    } else {
        this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic, true, this.defaultMQProducer);
        topicPublishInfo = this.topicPublishInfoTable.get(topic);
        return topicPublishInfo;
    }
}

 4、MQFaultStrategy.selectOneMessageQueue()方法

RcoketMQ中一个topic对应多个队列，该方法从topic的队列列表中选中一个Queue返回，选择队列的算法为轮询。

public MessageQueue selectOneMessageQueue(final TopicPublishInfo tpInfo, final String lastBrokerName) {
    if (this.sendLatencyFaultEnable) {
        try {
            int index = tpInfo.getSendWhichQueue().getAndIncrement();//记录选择累计次数
            for (int i = 0; i < tpInfo.getMessageQueueList().size(); i++) {
                int pos = Math.abs(index++) % tpInfo.getMessageQueueList().size();//总次数对队列大小取余
                if (pos < 0)
                    pos = 0;
                MessageQueue mq = tpInfo.getMessageQueueList().get(pos);//根据余数取出一个队列返回，这样就保证了发送端几乎是均衡把消息发送到各个队列里面。
                if (latencyFaultTolerance.isAvailable(mq.getBrokerName())) {
                    if (null == lastBrokerName || mq.getBrokerName().equals(lastBrokerName))
                        return mq;
                }
            }

            final String notBestBroker = latencyFaultTolerance.pickOneAtLeast();
            int writeQueueNums = tpInfo.getQueueIdByBroker(notBestBroker);
            if (writeQueueNums > 0) {
                final MessageQueue mq = tpInfo.selectOneMessageQueue();
                if (notBestBroker != null) {
                    mq.setBrokerName(notBestBroker);
                    mq.setQueueId(tpInfo.getSendWhichQueue().getAndIncrement() % writeQueueNums);
                }
                return mq;
            } else {
                latencyFaultTolerance.remove(notBestBroker);
            }
        } catch (Exception e) {
            log.error("Error occurred when selecting message queue", e);
        }

        return tpInfo.selectOneMessageQueue();
    }

    return tpInfo.selectOneMessageQueue(lastBrokerName);
}

 getSendWhichQueue()说一下，看下代码：

这里用到了volatile关键字和ThreadLocal类，volatile保证了sendWhichQueue一旦变化，Group下其他Producer马上就能看到最新数据。而之所以用ThreadLocal<Integer>不用AtomicInteger类。因为这里只需要定义一个线程内的局部变量，只供本线程使用。而AtomicInteger主要是为了解决兵法问题。

public class TopicPublishInfo {
    private volatile ThreadLocalIndex sendWhichQueue = new ThreadLocalIndex();
}

public class ThreadLocalIndex {
    private final ThreadLocal<Integer> threadLocalIndex = new ThreadLocal<Integer>();
}

5、DefaultMQProducerImpl.sendKernelImpl()方法，只贴出了主要代码，具体代码大家还是去看源码吧。

sendKernelImpl方法主要作用就是封包，发送消息给Broker。

private SendResult sendKernelImpl(final Message msg,
                                  final MessageQueue mq,
                                  final CommunicationMode communicationMode,
                                  final SendCallback sendCallback,
                                  final TopicPublishInfo topicPublishInfo,
                                  final long timeout) {
    //获取broker地址
    String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
    //给消息设置唯一ID
    MessageClientIDSetter.setUniqID(msg);
    //当消息体大于4M时，进行压缩，注意只压缩消息体
    UtilAll.compress(body, zipCompressLevel);
    //使用是否VIPChannel通道发送数据，vip channel的端口号为普通端口号-2。
    brokerAddr = MixAll.brokerVIPChannel(this.defaultMQProducer.isSendMessageWithVIPChannel(), brokerAddr);
    //发送消息校验
    CheckForbiddenContext checkForbiddenContext = new CheckForbiddenContext();
       this.executeCheckForbiddenHook(checkForbiddenContext);

　　　　
    //发送消息前逻辑
    SendMessageContext context = new SendMessageContext();
       this.executeSendMessageHookBefore(context);

    //构建发送消息请求
    SendMessageRequestHeader requestHeader = new SendMessageRequestHeader();
    //底层调用netty发送消息给broker。
    sendResult = this.mQClientFactory.getMQClientAPIImpl().sendMessage();
    return sendResult;
}