RocketMQ消息发送篇

朱雀 2022-11-04 15:29 227阅读 0赞

**RocketMQ支持3种消息发送方式：同步（sync）、异步（async）、单向（oneway）。**

*  **同步：发送者向MQ执行发送消息API时，同步等待，直到消息服务器返回发送结果。**
 *  **异步：发送者向MQ执行发送消息API时，指定消息发送成功后的回调函数，然后调用消息发送API后，立即返回，消息发送者线程不阻塞，直到运行结束，消息发送成功或失败的回调任务在一个新的线程中执行。**
 *  **单向：消息发送者向MQ执行发送消息API时，直接返回，不等待消息服务器的结果，也不注册回调函数，简单地说，就是只管发，不在乎消息是否成功存储在消息服务器上。**

#  Message #

**RocketMQ消息封装类是org.apache.rocketmq.common.message.Message。**

**Message的基础属性主要包括消息所属主题topic、消息Flag（RocketMQ不做处理）、扩展属性、消息体。**

**Message扩展属性主要包含下面几个。**

*  **tag：消息TAG，用于消息过滤。**
 *  **keys:Message索引键，多个用空格隔开，RocketMQ可以根据这些key快速检索到消息。**
 *  **waitStoreMsgOK：消息发送时是否等消息存储完成后再返回。**
 *  **delayTimeLevel：消息延迟级别，用于定时消息或消息重试。**

**这些扩展属性存储在Message的properties中。**

# **生产者启动流程** #

**DefaultMQProducer是默认的消息生产者实现类，它实现MQAdmin的接口。**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_16_color_FFFFFF_t_70][]

public void start() throws MQClientException {
            this.setProducerGroup(withNamespace(this.producerGroup));
            this.defaultMQProducerImpl.start();
            if (null != traceDispatcher) {
                try {
                    traceDispatcher.start(this.getNamesrvAddr(), this.getAccessChannel());
                } catch (MQClientException e) {
                    log.warn("trace dispatcher start failed ", e);
                }
            }
        }
    
    
    
        public void start(final boolean startFactory) throws MQClientException {
            switch (this.serviceState) {
                case CREATE_JUST:
                    this.serviceState = ServiceState.START_FAILED;
    
                    this.checkConfig();
    
                    if (!this.defaultMQProducer.getProducerGroup().equals(MixAll.CLIENT_INNER_PRODUCER_GROUP)) {
                        this.defaultMQProducer.changeInstanceNameToPID();
                    }
    
                    this.mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(this.defaultMQProducer, rpcHook);
    
                    boolean registerOK = mQClientFactory.registerProducer(this.defaultMQProducer.getProducerGroup(), this);
                    if (!registerOK) {
                        this.serviceState = ServiceState.CREATE_JUST;
                        throw new MQClientException("The producer group[" + this.defaultMQProducer.getProducerGroup()
                            + "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL),
                            null);
                    }
    
                    this.topicPublishInfoTable.put(this.defaultMQProducer.getCreateTopicKey(), new TopicPublishInfo());
    
                    if (startFactory) {
                        mQClientFactory.start();
                    }
    
                    log.info("the producer [{}] start OK. sendMessageWithVIPChannel={}", this.defaultMQProducer.getProducerGroup(),
                        this.defaultMQProducer.isSendMessageWithVIPChannel());
                    this.serviceState = ServiceState.RUNNING;
                    break;
                case RUNNING:
                case START_FAILED:
                case SHUTDOWN_ALREADY:
                    throw new MQClientException("The producer service state not OK, maybe started once, "
                        + this.serviceState
                        + FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK),
                        null);
                default:
                    break;
            }
    
            this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
    
            this.timer.scheduleAtFixedRate(new TimerTask() {
                @Override
                public void run() {
                    try {
                        RequestFutureTable.scanExpiredRequest();
                    } catch (Throwable e) {
                        log.error("scan RequestFutureTable exception", e);
                    }
                }
            }, 1000 * 3, 1000);
        }

### Step1 ###

**检查productGroup是否符合要求；并改变生产者的instanceName为进程ID。**

this.checkConfig();
    
                    if (!this.defaultMQProducer.getProducerGroup().equals(MixAll.CLIENT_INNER_PRODUCER_GROUP)) {
                        this.defaultMQProducer.changeInstanceNameToPID();
                    }

### Step2 ###

**创建MQClientInstance实例。整个JVM实例中只存在一个MQClientManager实例，维护一个MQClientInstance缓存表ConcurrentMap<String/\* clientId \*/,MQClientInstance>factoryTable =newConcurrentHashMap<String, MQClientInstance>（），也就是同一个clientId只会创建一个MQClientInstance。**

this.mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(this.defaultMQProducer, rpcHook);
    
    
      public MQClientInstance getOrCreateMQClientInstance(final ClientConfig clientConfig, RPCHook rpcHook) {
            //clientId为客户端IP+instance+（unitname可选）
            //（如果instance为默认值DEFAULT的话，RocketMQ会自动将instance设置为进程ID，这样避免了不同进程的相互影响，但同一个JVM中的不同消费者和不同生产者在启动时获取到的MQClientInstane实例都是同一个。）
            String clientId = clientConfig.buildMQClientId();
            MQClientInstance instance = this.factoryTable.get(clientId);
            if (null == instance) {
                instance =
                    new MQClientInstance(clientConfig.cloneClientConfig(),
                        this.factoryIndexGenerator.getAndIncrement(), clientId, rpcHook);
                MQClientInstance prev = this.factoryTable.putIfAbsent(clientId, instance);
                if (prev != null) {
                    instance = prev;
                    log.warn("Returned Previous MQClientInstance for clientId:[{}]", clientId);
                } else {
                    log.info("Created new MQClientInstance for clientId:[{}]", clientId);
                }
            }
    
            return instance;
        }

### Step3 ###

**向MQClientInstance注册，将当前生产者加入到MQClientInstance管理中，方便后续调用网络请求、进行心跳检测等。**

boolean registerOK = mQClientFactory.registerProducer(this.defaultMQProducer.getProducerGroup(), this);
                    if (!registerOK) {
                        this.serviceState = ServiceState.CREATE_JUST;
                        throw new MQClientException("The producer group[" + this.defaultMQProducer.getProducerGroup()
                            + "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL),
                            null);
                    }

### **Step4** ###

**启动MQClientInstance，如果MQClientInstance已经启动，则本次启动不会真正执行。**

this.topicPublishInfoTable.put(this.defaultMQProducer.getCreateTopicKey(), new TopicPublishInfo());
    
                    if (startFactory) {
                        mQClientFactory.start();
                    }
    
                    log.info("the producer [{}] start OK. sendMessageWithVIPChannel={}", this.defaultMQProducer.getProducerGroup(),
                        this.defaultMQProducer.isSendMessageWithVIPChannel());
                    this.serviceState = ServiceState.RUNNING;

# 消息发送基本流程 #

**消息发送流程主要的步骤：验证消息、查找路由、消息发送（包含异常处理机制）。**

## **消息长度验证** ##

**消息发送之前，首先确保生产者处于运行状态，然后验证消息是否符合相应的规范，具体的规范要求是主题名称、消息体不能为空、消息长度不能等于0且默认不能超过允许发送消息的最大长度4M（maxMessageSize=1024 \* 1024 \* 4）。**

## **查找主题路由信息** ##

**消息发送之前，首先需要获取主题的路由信息，只有获取了这些信息我们才知道消息要发送到具体的Broker节点。**

private TopicPublishInfo tryToFindTopicPublishInfo(final String topic) {
            TopicPublishInfo topicPublishInfo = this.topicPublishInfoTable.get(topic);
            if (null == topicPublishInfo || !topicPublishInfo.ok()) {
                this.topicPublishInfoTable.putIfAbsent(topic, new TopicPublishInfo());
                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;
            }
        }

**tryToFindTopicPublishInfo是查找主题的路由信息的方法。如果生产者中缓存了topic的路由信息，且该路由信息中包含了消息队列，则直接返回该路由信息，如果没有缓存或没有包含消息队列，则向NameServer查询该topic的路由信息。如果最终未找到路由信息，则抛出异常：无法找到主题相关路由信息异常。**

**第一次发送消息时，本地没有缓存topic的路由信息，查询NameServer尝试获取，如果路由信息未找到，再次尝试用默认主题DefaultMQProducerImpl\#createTopicKey去查询，如果BrokerConfig\#autoCreateTopicEnable为true时，NameServer将返回路由信息，如果autoCreateTopicEnable为false将抛出无法找到topic路由异常。**

**代码MQClientInstance\#updateTopicRouteInfoFromNameServer这个方法的功能是消息生产者更新和维护路由缓存，具体代码如下。**

public boolean updateTopicRouteInfoFromNameServer(final String topic, boolean isDefault,
            DefaultMQProducer defaultMQProducer) {
            try {
                if (this.lockNamesrv.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) {
                    try {
                        TopicRouteData topicRouteData;
                        if (isDefault && defaultMQProducer != null) {
                            topicRouteData = this.mQClientAPIImpl.getDefaultTopicRouteInfoFromNameServer(defaultMQProducer.getCreateTopicKey(),
                                1000 * 3);
                            if (topicRouteData != null) {
                                for (QueueData data : topicRouteData.getQueueDatas()) {
                                    int queueNums = Math.min(defaultMQProducer.getDefaultTopicQueueNums(), data.getReadQueueNums());
                                    data.setReadQueueNums(queueNums);
                                    data.setWriteQueueNums(queueNums);
                                }
                            }
                        } else {
                            topicRouteData = this.mQClientAPIImpl.getTopicRouteInfoFromNameServer(topic, 1000 * 3);
                        }
                        if (topicRouteData != null) {
                            TopicRouteData old = this.topicRouteTable.get(topic);
                            boolean changed = topicRouteDataIsChange(old, topicRouteData);
                            if (!changed) {
                                changed = this.isNeedUpdateTopicRouteInfo(topic);
                            } else {
                                log.info("the topic[{}] route info changed, old[{}] ,new[{}]", topic, old, topicRouteData);
                            }
    
                            if (changed) {
                                TopicRouteData cloneTopicRouteData = topicRouteData.cloneTopicRouteData();
    
                                for (BrokerData bd : topicRouteData.getBrokerDatas()) {
                                    this.brokerAddrTable.put(bd.getBrokerName(), bd.getBrokerAddrs());
                                }
    
                                // Update Pub info
                                {
                                    TopicPublishInfo publishInfo = topicRouteData2TopicPublishInfo(topic, topicRouteData);
                                    publishInfo.setHaveTopicRouterInfo(true);
                                    Iterator<Entry<String, MQProducerInner>> it = this.producerTable.entrySet().iterator();
                                    while (it.hasNext()) {
                                        Entry<String, MQProducerInner> entry = it.next();
                                        MQProducerInner impl = entry.getValue();
                                        if (impl != null) {
                                            impl.updateTopicPublishInfo(topic, publishInfo);
                                        }
                                    }
                                }
    
                                // Update sub info
                                {
                                    Set<MessageQueue> subscribeInfo = topicRouteData2TopicSubscribeInfo(topic, topicRouteData);
                                    Iterator<Entry<String, MQConsumerInner>> it = this.consumerTable.entrySet().iterator();
                                    while (it.hasNext()) {
                                        Entry<String, MQConsumerInner> entry = it.next();
                                        MQConsumerInner impl = entry.getValue();
                                        if (impl != null) {
                                            impl.updateTopicSubscribeInfo(topic, subscribeInfo);
                                        }
                                    }
                                }
                                log.info("topicRouteTable.put. Topic = {}, TopicRouteData[{}]", topic, cloneTopicRouteData);
                                this.topicRouteTable.put(topic, cloneTopicRouteData);
                                return true;
                            }
                        } else {
                            log.warn("updateTopicRouteInfoFromNameServer, getTopicRouteInfoFromNameServer return null, Topic: {}. [{}]", topic, this.clientId);
                        }
                    } catch (MQClientException e) {
                        if (!topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
                            log.warn("updateTopicRouteInfoFromNameServer Exception", e);
                        }
                    } catch (RemotingException e) {
                        log.error("updateTopicRouteInfoFromNameServer Exception", e);
                        throw new IllegalStateException(e);
                    } finally {
                        this.lockNamesrv.unlock();
                    }
                } else {
                    log.warn("updateTopicRouteInfoFromNameServer tryLock timeout {}ms. [{}]", LOCK_TIMEOUT_MILLIS, this.clientId);
                }
            } catch (InterruptedException e) {
                log.warn("updateTopicRouteInfoFromNameServer Exception", e);
            }
    
            return false;
        }

### Step1 ###

**如果isDefault为true，则使用默认主题去查询，如果查询到路由信息，则替换路由信息中读写队列个数为消息生产者默认的队列个数（defaultTopicQueueNums）；如果isDefault为false，则使用参数topic去查询；如果未查询到路由信息，则返回false，表示路由信息未变化。**

TopicRouteData topicRouteData;
                        if (isDefault && defaultMQProducer != null) {
                            topicRouteData = this.mQClientAPIImpl.getDefaultTopicRouteInfoFromNameServer(defaultMQProducer.getCreateTopicKey(),
                                1000 * 3);
                            if (topicRouteData != null) {
                                for (QueueData data : topicRouteData.getQueueDatas()) {
                                    int queueNums = Math.min(defaultMQProducer.getDefaultTopicQueueNums(), data.getReadQueueNums());
                                    data.setReadQueueNums(queueNums);
                                    data.setWriteQueueNums(queueNums);
                                }
                            }
                        } else {
                            topicRouteData = this.mQClientAPIImpl.getTopicRouteInfoFromNameServer(topic, 1000 * 3);
                        }

### **Step2** ###

**如果路由信息找到，与本地缓存中的路由信息进行对比，判断路由信息是否发生了改变，如果未发生变化，则直接返回false。**

TopicRouteData old = this.topicRouteTable.get(topic);
                            boolean changed = topicRouteDataIsChange(old, topicRouteData);
                            if (!changed) {
                                changed = this.isNeedUpdateTopicRouteInfo(topic);
                            } else {
                                log.info("the topic[{}] route info changed, old[{}] ,new[{}]", topic, old, topicRouteData);
                            }

### Step3 ###

**更新MQClientInstance Broker地址缓存表。**

TopicRouteData cloneTopicRouteData = topicRouteData.cloneTopicRouteData();
    
                                for (BrokerData bd : topicRouteData.getBrokerDatas()) {
                                    this.brokerAddrTable.put(bd.getBrokerName(), bd.getBrokerAddrs());
                                }

### Step4 ###

**根据topicRouteData中的List<QueueData>转换成topicPublishInfo的List<MessageQueue>列表。其具体实现在topicRouteData2TopicPublishInfo，然后会更新该MQClientInstance所管辖的所有消息发送关于topic的路由信息。**

{
                                    TopicPublishInfo publishInfo = topicRouteData2TopicPublishInfo(topic, topicRouteData);
                                    publishInfo.setHaveTopicRouterInfo(true);
                                    Iterator<Entry<String, MQProducerInner>> it = this.producerTable.entrySet().iterator();
                                    while (it.hasNext()) {
                                        Entry<String, MQProducerInner> entry = it.next();
                                        MQProducerInner impl = entry.getValue();
                                        if (impl != null) {
                                            impl.updateTopicPublishInfo(topic, publishInfo);
                                        }
                                    }
                                }
    
    
        public static TopicPublishInfo topicRouteData2TopicPublishInfo(final String topic, final TopicRouteData route) {
            TopicPublishInfo info = new TopicPublishInfo();
            info.setTopicRouteData(route);
            if (route.getOrderTopicConf() != null && route.getOrderTopicConf().length() > 0) {
                String[] brokers = route.getOrderTopicConf().split(";");
                for (String broker : brokers) {
                    String[] item = broker.split(":");
                    int nums = Integer.parseInt(item[1]);
                    for (int i = 0; i < nums; i++) {
                        MessageQueue mq = new MessageQueue(topic, item[0], i);
                        info.getMessageQueueList().add(mq);
                    }
                }
    
                info.setOrderTopic(true);
            } else {
                List<QueueData> qds = route.getQueueDatas();
                Collections.sort(qds);
                for (QueueData qd : qds) {
                    if (PermName.isWriteable(qd.getPerm())) {
                        BrokerData brokerData = null;
                        for (BrokerData bd : route.getBrokerDatas()) {
                            if (bd.getBrokerName().equals(qd.getBrokerName())) {
                                brokerData = bd;
                                break;
                            }
                        }
    
                        if (null == brokerData) {
                            continue;
                        }
    
                        if (!brokerData.getBrokerAddrs().containsKey(MixAll.MASTER_ID)) {
                            continue;
                        }
    
                        for (int i = 0; i < qd.getWriteQueueNums(); i++) {
                            MessageQueue mq = new MessageQueue(topic, qd.getBrokerName(), i);
                            info.getMessageQueueList().add(mq);
                        }
                    }
                }
    
                info.setOrderTopic(false);
            }
    
            return info;
        }

**循环遍历路由信息的QueueData信息，如果队列没有写权限，则继续遍历下一个QueueData；根据brokerName找到brokerData信息，找不到或没有找到Master节点，则遍历下一个QueueData；根据写队列个数，根据topic+序号创建MessageQueue，填充topicPublishInfo的List<QuueMessage>。完成消息发送的路由查找。**

# **选择消息队列** #

**首先消息发送端采用重试机制，由retryTimesWhenSendFailed指定同步方式重试次数，异步重试机制在收到消息发送结构后执行回调之前进行重试。由retryTimes When Send-AsyncFailed指定，接下来就是循环执行，选择消息队列、发送消息，发送成功则返回，收到异常则重试。选择消息队列有两种方式。**

*  **sendLatencyFaultEnable=false，默认不启用Broker故障延迟机制。**
 *  **sendLatencyFaultEnable=true，启用Broker故障延迟机制。**

# 消息发送 #

**消息发送API核心入口：DefaultMQProducerImpl\#sendKernelImpl。**

private SendResult sendKernelImpl(final Message msg,
            final MessageQueue mq,
            final CommunicationMode communicationMode,
            final SendCallback sendCallback,
            final TopicPublishInfo topicPublishInfo,
            final long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
            long beginStartTime = System.currentTimeMillis();
            String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
            if (null == brokerAddr) {
                tryToFindTopicPublishInfo(mq.getTopic());
                brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
            }
    
            SendMessageContext context = null;
            if (brokerAddr != null) {
                brokerAddr = MixAll.brokerVIPChannel(this.defaultMQProducer.isSendMessageWithVIPChannel(), brokerAddr);
    
                byte[] prevBody = msg.getBody();
                try {
                    //for MessageBatch,ID has been set in the generating process
                    if (!(msg instanceof MessageBatch)) {
                        MessageClientIDSetter.setUniqID(msg);
                    }
    
                    boolean topicWithNamespace = false;
                    if (null != this.mQClientFactory.getClientConfig().getNamespace()) {
                        msg.setInstanceId(this.mQClientFactory.getClientConfig().getNamespace());
                        topicWithNamespace = true;
                    }
    
                    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 (hasCheckForbiddenHook()) {
                        CheckForbiddenContext checkForbiddenContext = new CheckForbiddenContext();
                        checkForbiddenContext.setNameSrvAddr(this.defaultMQProducer.getNamesrvAddr());
                        checkForbiddenContext.setGroup(this.defaultMQProducer.getProducerGroup());
                        checkForbiddenContext.setCommunicationMode(communicationMode);
                        checkForbiddenContext.setBrokerAddr(brokerAddr);
                        checkForbiddenContext.setMessage(msg);
                        checkForbiddenContext.setMq(mq);
                        checkForbiddenContext.setUnitMode(this.isUnitMode());
                        this.executeCheckForbiddenHook(checkForbiddenContext);
                    }
    
                    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);
                        context.setNamespace(this.defaultMQProducer.getNamespace());
                        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());
                    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);
                        }
                    }
    
                    SendResult sendResult = null;
                    switch (communicationMode) {
                        case ASYNC:
                            Message tmpMessage = msg;
                            boolean messageCloned = false;
                            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);
                                messageCloned = true;
                                msg.setBody(prevBody);
                            }
    
                            if (topicWithNamespace) {
                                if (!messageCloned) {
                                    tmpMessage = MessageAccessor.cloneMessage(msg);
                                    messageCloned = true;
                                }
                                msg.setTopic(NamespaceUtil.withoutNamespace(msg.getTopic(), this.defaultMQProducer.getNamespace()));
                            }
    
                            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;
                    }
    
                    if (this.hasSendMessageHook()) {
                        context.setSendResult(sendResult);
                        this.executeSendMessageHookAfter(context);
                    }
    
                    return sendResult;
                } catch (RemotingException e) {
                    if (this.hasSendMessageHook()) {
                        context.setException(e);
                        this.executeSendMessageHookAfter(context);
                    }
                    throw e;
                } catch (MQBrokerException e) {
                    if (this.hasSendMessageHook()) {
                        context.setException(e);
                        this.executeSendMessageHookAfter(context);
                    }
                    throw e;
                } catch (InterruptedException e) {
                    if (this.hasSendMessageHook()) {
                        context.setException(e);
                        this.executeSendMessageHookAfter(context);
                    }
                    throw e;
                } finally {
                    msg.setBody(prevBody);
                    msg.setTopic(NamespaceUtil.withoutNamespace(msg.getTopic(), this.defaultMQProducer.getNamespace()));
                }
            }
    
            throw new MQClientException("The broker[" + mq.getBrokerName() + "] not exist", null);
        }

### Step1 ###

**根据MessageQueue获取Broker的网络地址。如果MQClientInstance的brokerAddrTable未缓存该Broker的信息，则从NameServer主动更新一下topic的路由信息。如果路由更新后还是找不到Broker信息，则抛出MQClientException，提示Broker不存在。**

String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
            if (null == brokerAddr) {
                tryToFindTopicPublishInfo(mq.getTopic());
                brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
            }

### Step2 ###

**为消息分配全局唯一ID，如果消息体默认超过4K（compressMsgBodyOverHowmuch），会对消息体采用zip压缩，并设置消息的系统标记为MessageSysFlag.COMPRESSED\_FLAG。如果是事务Prepared消息，则设置消息的系统标记为MessageSysFlag.TRANSACTION\_PREPARED\_TYPE。**

//for MessageBatch,ID has been set in the generating process
                    if (!(msg instanceof MessageBatch)) {
                        MessageClientIDSetter.setUniqID(msg);
                    }
    
                    boolean topicWithNamespace = false;
                    if (null != this.mQClientFactory.getClientConfig().getNamespace()) {
                        msg.setInstanceId(this.mQClientFactory.getClientConfig().getNamespace());
                        topicWithNamespace = true;
                    }
    
                    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;
                    }

###  Step3 ###

**如果注册了消息发送钩子函数，则执行消息发送之前的增强逻辑。通过DefaultMQProducerImpl\#registerSendMessageHook注册钩子处理类，并且可以注册多个。简单看一下钩子处理类接口。**

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);
                        context.setNamespace(this.defaultMQProducer.getNamespace());
                        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);
                    }

### Step4 ###

**构建消息发送请求包。主要包含如下重要信息：生产者组、主题名称、默认创建主题Key、该主题在单个Broker默认队列数、队列ID（队列序号）、消息系统标记（MessageSysFlag）、消息发送时间、消息标记（RocketMQ对消息中的flag不做任何处理，供应用程序使用）、消息扩展属性、消息重试次数、是否是批量消息等。**

SendMessageRequestHeader requestHeader = new SendMessageRequestHeader();
                    requestHeader.setProducerGroup(this.defaultMQProducer.getProducerGroup());
                    requestHeader.setTopic(msg.getTopic());
                    requestHeader.setDefaultTopic(this.defaultMQProducer.getCreateTopicKey());
                    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);
                        }
                    }

### Step5 ###

**根据消息发送方式，同步、异步、单向方式进行网络传输。**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_16_color_FFFFFF_t_70 1][]

public SendResult sendMessage(
            final String addr,
            final String brokerName,
            final Message msg,
            final SendMessageRequestHeader requestHeader,
            final long timeoutMillis,
            final CommunicationMode communicationMode,
            final SendCallback sendCallback,
            final TopicPublishInfo topicPublishInfo,
            final MQClientInstance instance,
            final int retryTimesWhenSendFailed,
            final SendMessageContext context,
            final DefaultMQProducerImpl producer
        ) throws RemotingException, MQBrokerException, InterruptedException {
            long beginStartTime = System.currentTimeMillis();
            RemotingCommand request = null;
            String msgType = msg.getProperty(MessageConst.PROPERTY_MESSAGE_TYPE);
            boolean isReply = msgType != null && msgType.equals(MixAll.REPLY_MESSAGE_FLAG);
            if (isReply) {
                if (sendSmartMsg) {
                    SendMessageRequestHeaderV2 requestHeaderV2 = SendMessageRequestHeaderV2.createSendMessageRequestHeaderV2(requestHeader);
                    request = RemotingCommand.createRequestCommand(RequestCode.SEND_REPLY_MESSAGE_V2, requestHeaderV2);
                } else {
                    request = RemotingCommand.createRequestCommand(RequestCode.SEND_REPLY_MESSAGE, requestHeader);
                }
            } else {
                if (sendSmartMsg || msg instanceof MessageBatch) {
                    SendMessageRequestHeaderV2 requestHeaderV2 = SendMessageRequestHeaderV2.createSendMessageRequestHeaderV2(requestHeader);
                    request = RemotingCommand.createRequestCommand(msg instanceof MessageBatch ? RequestCode.SEND_BATCH_MESSAGE : RequestCode.SEND_MESSAGE_V2, requestHeaderV2);
                } else {
                    request = RemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE, requestHeader);
                }
            }
            request.setBody(msg.getBody());
    
            switch (communicationMode) {
                case ONEWAY:
                    this.remotingClient.invokeOneway(addr, request, timeoutMillis);
                    return null;
                case ASYNC:
                    final AtomicInteger times = new AtomicInteger();
                    long costTimeAsync = System.currentTimeMillis() - beginStartTime;
                    if (timeoutMillis < costTimeAsync) {
                        throw new RemotingTooMuchRequestException("sendMessage call timeout");
                    }
                    this.sendMessageAsync(addr, brokerName, msg, timeoutMillis - costTimeAsync, request, sendCallback, topicPublishInfo, instance,
                        retryTimesWhenSendFailed, times, context, producer);
                    return null;
                case SYNC:
                    long costTimeSync = System.currentTimeMillis() - beginStartTime;
                    if (timeoutMillis < costTimeSync) {
                        throw new RemotingTooMuchRequestException("sendMessage call timeout");
                    }
                    return this.sendMessageSync(addr, brokerName, msg, timeoutMillis - costTimeSync, request);
                default:
                    assert false;
                    break;
            }
    
            return null;
        }

###  Step6 ###

**如果注册了消息发送钩子函数，执行after逻辑。注意，就算消息发送过程中发生RemotingException、MQBrokerException、InterruptedException时该方法也会执行。**

if (this.hasSendMessageHook()) {
                        context.setSendResult(sendResult);
                        this.executeSendMessageHookAfter(context);
                    }

# 批量消息发送 #

**批量消息发送是将同一主题的多条消息一起打包发送到消息服务端，减少网络调用次数，提高网络传输效率。单条消息发送时，消息体的内容将保存在body中。批量消息发送，需要将多条消息体的内容存储在body中RocketMQ采取的方式是，对单条消息内容使用固定格式进行存储。**

![RocetMQ批量消息封装格式][RocetMQ] RocetMQ批量消息封装格式

@Override
        public SendResult send(
            Collection<Message> msgs) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
            return this.defaultMQProducerImpl.send(batch(msgs));
        }
    
        private MessageBatch batch(Collection<Message> msgs) throws MQClientException {
            MessageBatch msgBatch;
            try {
                msgBatch = MessageBatch.generateFromList(msgs);
                for (Message message : msgBatch) {
                    Validators.checkMessage(message, this);
                    MessageClientIDSetter.setUniqID(message);
                    message.setTopic(withNamespace(message.getTopic()));
                }
                msgBatch.setBody(msgBatch.encode());
            } catch (Exception e) {
                throw new MQClientException("Failed to initiate the MessageBatch", e);
            }
            msgBatch.setTopic(withNamespace(msgBatch.getTopic()));
            return msgBatch;
        }
    
    
        public byte[] encode() {
            return MessageDecoder.encodeMessages(messages);
        }

**首先在消息发送端，调用batch方法，将一批消息封装成MessageBatch对象。Message-Batch继承自Message对象，MessageBatch内部持有List<Message> messages。这样的话，批量消息发送与单条消息发送的处理流程完全一样。MessageBatch只需要将该集合中的每条消息的消息体body聚合成一个byte\[\]数值，在消息服务端能够从该byte\[\]数值中正确解析出消息即可。**

**在创建RemotingCommand对象时将调用messageBatch\#encode()方法填充到Remoting-Command的body域中。**

public static byte[] encodeMessage(Message message) {
            //only need flag, body, properties
            byte[] body = message.getBody();
            int bodyLen = body.length;
            String properties = messageProperties2String(message.getProperties());
            byte[] propertiesBytes = properties.getBytes(CHARSET_UTF8);
            //note properties length must not more than Short.MAX
            short propertiesLength = (short) propertiesBytes.length;
            int sysFlag = message.getFlag();
            int storeSize = 4 // 1 TOTALSIZE
                + 4 // 2 MAGICCOD
                + 4 // 3 BODYCRC
                + 4 // 4 FLAG
                + 4 + bodyLen // 4 BODY
                + 2 + propertiesLength;
            ByteBuffer byteBuffer = ByteBuffer.allocate(storeSize);
            // 1 TOTALSIZE
            byteBuffer.putInt(storeSize);
    
            // 2 MAGICCODE
            byteBuffer.putInt(0);
    
            // 3 BODYCRC
            byteBuffer.putInt(0);
    
            // 4 FLAG
            int flag = message.getFlag();
            byteBuffer.putInt(flag);
    
            // 5 BODY
            byteBuffer.putInt(bodyLen);
            byteBuffer.put(body);
    
            // 6 properties
            byteBuffer.putShort(propertiesLength);
            byteBuffer.put(propertiesBytes);
    
            return byteBuffer.array();
        }

**在消息发送端将会按照上述结构进行解码，然后整个发送流程与单个消息发送没什么差异。**

**参考《RocketMQ技术内幕：RocketMQ架构设计与实现原理》**

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_16_color_FFFFFF_t_70]: /images/20221023/2e14091c3a7c4d28b884404a4b913b1f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_16_color_FFFFFF_t_70 1]: /images/20221023/070307306fba4e6b9ce30289a97c7a36.png
[RocetMQ]: /images/20221023/2e29d70c469a4e308e75c386c0aa07d8.png