引言

前面我们介绍了RocketMQ在发送消息之前做了一系列的准备事项，其中包括路由选择、队列选择以及坏点Broker退避等等。本文将开始阐述RocketMQ的消息发送过程。

• 消息发送基本流程
• 总结

一、消息发送基本流程

下面我们看下进行消息发送的最核心的API，即DefaultMQProducerImpl类中的sendKernelImpl方法如下所示(相关字段注释如下)：

private SendResult sendKernelImpl(
//需要发送的消息
final Message msg,
									  //消息需要发送到的消息队列
                                      final MessageQueue mq,
                                      //消息发送模式
                                      final CommunicationMode communicationMode,
                                      //异步消息回调
                                      final SendCallback sendCallback,
                                      //主题路由信息
                                      final TopicPublishInfo topicPublishInfo,
                                      //超时时间
                                      final long timeout)

涉及的具体步骤如下所示：

我们具体看下源码，相关注释如下所示：

//根据选择的MessageQueue获取对应Broker地址
 String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
 		//如果为获取到
        if (null == brokerAddr) {
        	//从NameServer进行主动更新TOPIC信息
            tryToFindTopicPublishInfo(mq.getTopic());
            brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
        }

进行全局ID设置，源码如下：

if (!(msg instanceof MessageBatch)) {
 					//设置全局唯一ID
                    MessageClientIDSetter.setUniqID(msg);
                }
				//消息提的默认大小大小超过4K，则进行zip压缩，并设置消息系统标记
                int sysFlag = 0;
                boolean msgBodyCompressed = false;
                if (this.tryToCompressMessage(msg)) {
                    sysFlag |= MessageSysFlag.COMPRESSED_FLAG;
                    msgBodyCompressed = true;
                }
				//如果事务，则进行系统事务标记
                final String tranMsg = msg.getProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED);
                if (tranMsg != null && Boolean.parseBoolean(tranMsg)) {
                    sysFlag |= MessageSysFlag.TRANSACTION_PREPARED_TYPE;
                }

是否执行钩子函数，源码如下所示：

//判断是否进行消息发送钩子函数注册了，为一个列表
if (this.hasSendMessageHook()) {
                    context = new SendMessageContext();
                    context.setProducer(this);
                    context.setProducerGroup(this.defaultMQProducer.getProducerGroup());
                    context.setCommunicationMode(communicationMode);
                    context.setBornHost(this.defaultMQProducer.getClientIP());
                    context.setBrokerAddr(brokerAddr);
                    context.setMessage(msg);
                    context.setMq(mq);
                    String isTrans = msg.getProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED);
                    if (isTrans != null && isTrans.equals("true")) {
                        context.setMsgType(MessageType.Trans_Msg_Half);
                    }

                    if (msg.getProperty("__STARTDELIVERTIME") != null || msg.getProperty(MessageConst.PROPERTY_DELAY_TIME_LEVEL) != null) {
                        context.setMsgType(MessageType.Delay_Msg);
                    }
                    this.executeSendMessageHookBefore(context);
                }

进行消息发送请求包的构建，源码如下所示：

//消息发送请求头
SendMessageRequestHeader requestHeader = new SendMessageRequestHeader();
//设置消息生产组 requestHeader.setProducerGroup(this.defaultMQProducer.getProducerGroup());
                requestHeader.setTopic(msg.getTopic());
//设置主题名称 
requestHeader.setDefaultTopic(this.defaultMQProducer.getCreateTopicKey());
//设置默认 在单个Broker默认队列数 
requestHeader.setDefaultTopicQueueNums(this.defaultMQProducer.getDefaultTopicQueueNums());
                requestHeader.setQueueId(mq.getQueueId());
                //设置系统标记
                requestHeader.setSysFlag(sysFlag);
                requestHeader.setBornTimestamp(System.currentTimeMillis());
                
                requestHeader.setFlag(msg.getFlag());
                requestHeader.setProperties(MessageDecoder.messageProperties2String(msg.getProperties()));
                requestHeader.setReconsumeTimes(0);
                requestHeader.setUnitMode(this.isUnitMode());
                requestHeader.setBatch(msg instanceof MessageBatch);
                if (requestHeader.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
                    String reconsumeTimes = MessageAccessor.getReconsumeTime(msg);
                    if (reconsumeTimes != null) {
                        requestHeader.setReconsumeTimes(Integer.valueOf(reconsumeTimes));
                        MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_RECONSUME_TIME);
                    }

                    String maxReconsumeTimes = MessageAccessor.getMaxReconsumeTimes(msg);
                    if (maxReconsumeTimes != null) {
//设置最大重试次数 
requestHeader.setMaxReconsumeTimes(Integer.valueOf(maxReconsumeTimes));
                        MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_MAX_RECONSUME_TIMES);
                    }
                }

最后根据communicationMode，进行消息发送模式的选择，其中包括了同步发送方式、异步发送方式以及单向发送方式，大致的源码如下所示：

SendResult sendResult = null;
                switch (communicationMode) {
                	//异步方式
                    case ASYNC:
                        Message tmpMessage = msg;
                        if (msgBodyCompressed) {
                            //If msg body was compressed, msgbody should be reset using prevBody.
                            //Clone new message using commpressed message body and recover origin massage.
                            //Fix bug:https://github.com/apache/rocketmq-externals/issues/66
                            tmpMessage = MessageAccessor.cloneMessage(msg);
                            msg.setBody(prevBody);
                        }
                        long costTimeAsync = System.currentTimeMillis() - beginStartTime;
                        if (timeout < costTimeAsync) {
                            throw new RemotingTooMuchRequestException("sendKernelImpl call timeout");
                        }
                        sendResult = this.mQClientFactory.getMQClientAPIImpl().sendMessage(
                            brokerAddr,
                            mq.getBrokerName(),
                            tmpMessage,
                            requestHeader,
                            timeout - costTimeAsync,
                            communicationMode,
                            sendCallback,
                            topicPublishInfo,
                            this.mQClientFactory,
                            this.defaultMQProducer.getRetryTimesWhenSendAsyncFailed(),
                            context,
                            this);
                        break;
                    //单向以及同步
                    case ONEWAY:
                    case SYNC:
                        long costTimeSync = System.currentTimeMillis() - beginStartTime;
                        if (timeout < costTimeSync) {
                            throw new RemotingTooMuchRequestException("sendKernelImpl call timeout");
                        }
                        sendResult = this.mQClientFactory.getMQClientAPIImpl().sendMessage(
                            brokerAddr,
                            mq.getBrokerName(),
                            msg,
                            requestHeader,
                            timeout - costTimeSync,
                            communicationMode,
                            context,
                            this);
                        break;
                    default:
                        assert false;
                        break;
                }

二、总结

本文主要阐述了消息发送的基本流程。其中包括了获取Broker地址、设置全局ID、构建请求包以及发送消息了。在下篇文章中我们继续来看消息发送的底层流程。