(四)Rocketmq消息发送-蒲公英云

文章目录

- - 一.消息发送
  - 二.消息结构
  - 三.生产者启动流程
  - 四.消息发送基本流程
  - - 4.1消息长度验证
    - 4.2查找主题路由信息
    - 4.3选择消息队列
    - 4.4消息发送
  - 五.批量发送

Rocketmq发送消息有三种实现方式:可靠同步发送,可靠异步发送,单向(oneway)发送

一.消息发送

支持3种消息发送 sync同步 async异步 oneway单向
同步:发送者向MQ执行发送消息api,同步等待,知道消息服务器返回发送结果
异步:发送者向MQ执行发送消息api,指定消息发送成功后回调函数,然后调用消息发送API后,立即返回,消息发送者线程不阻塞,直到运行结束,消息发送成功或者失败的毁掉任务在一个新的线程中执行
单向:发送者向MQ执行发送消息api,直接返回,不等待消息服务器的结果,也不注册回调函数,简单说,就是只管发,不在乎是否成功存储在消息服务器上

二.消息结构

org.apache.rocketmq.common.message.Message

public class Message implements Serializable { 
    private static final long serialVersionUID = 8445773977080406428L;
    private String topic;
    /** * @see org.apache.rocketmq.common.sysflag.MessageSysFlag */
    private int flag;
    /** * 扩展属性 */
    private Map<String, String> properties;
    private byte[] body;
    private String transactionId;
    public Message() { 
    }
    public Message(String topic, byte[] body) { 
        this(topic, "", "", 0, body, true);
    }
    public Message(String topic, String tags, String keys, int flag, byte[] body, boolean waitStoreMsgOK) { 
        this.topic = topic;
        this.flag = flag;
        this.body = body;
        //消息的tag用于消息过滤
        if (tags != null && tags.length() > 0)
            this.setTags(tags);
        //Message索引键,多个用空格隔开,rocketmq根据这些key快速检索消息
        if (keys != null && keys.length() > 0)
            this.setKeys(keys);
        //发送时是否等消息存储完成再返回
        this.setWaitStoreMsgOK(waitStoreMsgOK);
    }

三.生产者启动流程

Producer的代码在client模块中,对于Rocketmq来说,它就是客户端,也是消息的提供者;
org.apache.rocketmq.client.producer.DefaultMQProducer
我们可以从quickstart中producer.start()开始追寻消息发送者的启动流程
org.apache.rocketmq.client.producer.DefaultMQProducer#start

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);
        }
    }
}

org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl#start(boolean)

public void start(final boolean startFactory) throws MQClientException { 
    //某人CREATE_JUST
    switch (this.serviceState) { 
        case CREATE_JUST:
            this.serviceState = ServiceState.START_FAILED;
            //检查配置
            this.checkConfig();
            //转换成进程id ->pid
            if (!this.defaultMQProducer.getProducerGroup().equals(MixAll.CLIENT_INNER_PRODUCER_GROUP)) { 
                this.defaultMQProducer.changeInstanceNameToPID();
            }
            //创建MQClientInstance实例 是producer cunsumer NameServer broker的网络通道
            this.mQClientFactory = MQClientManager.getInstance().getAndCreateMQClientInstance(this.defaultMQProducer, rpcHook);
            //生产组 生产者 放入producerTable
            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) { 
                //启动MQClientInstance
                mQClientFactory.start();
            }
            this.serviceState = ServiceState.RUNNING;
            break;
        .......
        default:
            break;
    }
    this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();

四.消息发送基本流程

消息发送的主要流程是:验证消息,查找路由,消息发送
org.apache.rocketmq.client.producer.DefaultMQProducer#send(org.apache.rocketmq.common.message.Message)

public SendResult send(
    Message msg) throws MQClientException, RemotingException, MQBrokerException, InterruptedException { 
    //验证消息
    Validators.checkMessage(msg, this);
    msg.setTopic(withNamespace(msg.getTopic()));
    return this.defaultMQProducerImpl.send(msg);
}

org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl#send(org.apache.rocketmq.common.message.Message) 默认的同步发送

public SendResult send(
    Message msg) throws MQClientException, RemotingException, MQBrokerException, InterruptedException { 
    return send(msg, this.defaultMQProducer.getSendMsgTimeout());
}
public SendResult send(Message msg,
    long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException { 
    return this.sendDefaultImpl(msg, CommunicationMode.SYNC, null, timeout);
}

默认发送超时时间为3000ms

4.1消息长度验证

消息发送之前,首先确保生产者处于运行状态,然后验证消息是否符合相应的规范,具体规范为主体名称,消息体不能为空,消息长度不能小于0且不能超过4m(maxMessageSize=102410244)

4.2查找主题路由信息

在发送之前,首先获取主题的路由信息,只有获取了这些信息我们才能知道要发送到具体的Broker节点,第一次发送消息时,本地没有缓存topic路由信息,查询NameServer尝试获取,如果路由信息未获取到,再次尝试用默认主题DefaultMQProducerImpl#createTopicKey去查询,如果BrokerConfig#autoCreateTopicEnable为true,NameServer将返回路由信息,如果为false,则抛出无法找到topic异常
org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl#tryToFindTopicPublishInfo

private TopicPublishInfo tryToFindTopicPublishInfo(final String topic) { 
    //从topic信息列表获取
    TopicPublishInfo topicPublishInfo = this.topicPublishInfoTable.get(topic);
    //未获取到 或者消息队列为null
    if (null == topicPublishInfo || !topicPublishInfo.ok()) { 
        this.topicPublishInfoTable.putIfAbsent(topic, new TopicPublishInfo());
        //更新topic相关列表
        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;
    }
}

org.apache.rocketmq.client.impl.factory.MQClientInstance#updateTopicRouteInfoFromNameServer(java.lang.String, boolean, org.apache.rocketmq.client.producer.DefaultMQProducer)

public boolean updateTopicRouteInfoFromNameServer(final String topic, boolean isDefault,
    DefaultMQProducer defaultMQProducer) { 
    try { 
        if (this.lockNamesrv.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) { 
            try { 
                TopicRouteData topicRouteData;
                //isDefault 为true时查询
                if (isDefault && defaultMQProducer != null) { 
                    //根据默认主题"TBW102"
                    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 { 
                    //根据topic查询
                    topicRouteData = this.mQClientAPIImpl.getTopicRouteInfoFromNameServer(topic, 1000 * 3);
                }
                //如果有topic路由信息
                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;

4.3选择消息队列

根据路由选择消息队列,返回消息队列按着broker,序号排序,假设topicA在broker-a broker-b分别创建2个队列,返回的消息队列是
[{“broker-name”:“broker-a”,“queueId”:0},{“broker-name”:“broker-a”,“queueId”:1},{“broker-name”:“broker-b”,“queueId”:0},{“broker-name”:“broker-b”,“queueId”:1}]
消息发送端采用重试机制,由retryTimesWhenSendFailed==2指定同步方式重试次数,异步重试机制在收到消息发送结构后执行回调之前进行重试,接下来就是循环执行,选择消息队列,发送消息,发送成功则返回,收到异常则重试,选择消息队列方式有两种
1)sendLatencyFaultEnable=false,默认不启动Broker故障延迟机制
2)sendLatencyFaultEnable=true,启动Broker故障延迟机制
step1:选择队列org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl#selectOneMessageQueue
org.apache.rocketmq.client.latency.MQFaultStrategy#selectOneMessageQueue 默认的策略就是轮询
org.apache.rocketmq.client.impl.producer.TopicPublishInfo#selectOneMessageQueue(java.lang.String)
默认模式

public MessageQueue selectOneMessageQueue(final String lastBrokerName) { 
    //第一次 lastBrokerName为null
    if (lastBrokerName == null) { 
        return selectOneMessageQueue();
    } else { 
        //序号+1
        int index = this.sendWhichQueue.getAndIncrement();
        for (int i = 0; i < this.messageQueueList.size(); i++) { 
            int pos = Math.abs(index++) % this.messageQueueList.size();
            if (pos < 0)
                pos = 0;
            //选择一个不一样的broker
            MessageQueue mq = this.messageQueueList.get(pos);
            if (!mq.getBrokerName().equals(lastBrokerName)) { 
                return mq;
            }
        }
        return selectOneMessageQueue();
    }
}
public MessageQueue selectOneMessageQueue() { 
    int index = this.sendWhichQueue.getAndIncrement();
    int pos = Math.abs(index) % this.messageQueueList.size();
    if (pos < 0)
        pos = 0;
    //轮序
    return this.messageQueueList.get(pos);
}

如果Broker宕机,生产端不会立刻知晓,需要消息生产者每隔30s进行一次路由更新
org.apache.rocketmq.client.latency.MQFaultStrategy#selectOneMessageQueue

4.4消息发送

org.apache.rocketmq.client.impl.producer.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();
    //broker的ip:port
    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 为消息分配全局id
            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;
            }
            //注册了消息发送钩子函数 执行after逻辑
            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);
            }
            ......

那么Broker是如何处理的呢?
org.apache.rocketmq.broker.processor.SendMessageProcessor#processRequest
org.apache.rocketmq.broker.processor.SendMessageProcessor#sendMessage

private RemotingCommand sendMessage(final ChannelHandlerContext ctx,
                                    final RemotingCommand request,
                                    final SendMessageContext sendMessageContext,
                                    final SendMessageRequestHeader requestHeader) throws RemotingCommandException { 
    final RemotingCommand response = RemotingCommand.createResponseCommand(SendMessageResponseHeader.class);
    final SendMessageResponseHeader responseHeader = (SendMessageResponseHeader)response.readCustomHeader();
    response.setOpaque(request.getOpaque());
    response.addExtField(MessageConst.PROPERTY_MSG_REGION, this.brokerController.getBrokerConfig().getRegionId());
    response.addExtField(MessageConst.PROPERTY_TRACE_SWITCH, String.valueOf(this.brokerController.getBrokerConfig().isTraceOn()));
    log.debug("receive SendMessage request command, {}", request);
    final long startTimstamp = this.brokerController.getBrokerConfig().getStartAcceptSendRequestTimeStamp();
    if (this.brokerController.getMessageStore().now() < startTimstamp) { 
        response.setCode(ResponseCode.SYSTEM_ERROR);
        response.setRemark(String.format("broker unable to service, until %s", UtilAll.timeMillisToHumanString2(startTimstamp)));
        return response;
    }
    response.setCode(-1);
    //消息检查
    super.msgCheck(ctx, requestHeader, response);
    if (response.getCode() != -1) { 
        return response;
    }
    final byte[] body = request.getBody();
    int queueIdInt = requestHeader.getQueueId();
    TopicConfig topicConfig = this.brokerController.getTopicConfigManager().selectTopicConfig(requestHeader.getTopic());
    if (queueIdInt < 0) { 
        queueIdInt = Math.abs(this.random.nextInt() % 99999999) % topicConfig.getWriteQueueNums();
    }
    MessageExtBrokerInner msgInner = new MessageExtBrokerInner();
    msgInner.setTopic(requestHeader.getTopic());
    msgInner.setQueueId(queueIdInt);
    //如果消息重试次数达到最大的重试次数,消息将进入DLD延迟队列 主题是%DLQ%+消费组
    if (!handleRetryAndDLQ(requestHeader, response, request, msgInner, topicConfig)) { 
        return response;
    }
    msgInner.setBody(body);
    msgInner.setFlag(requestHeader.getFlag());
    MessageAccessor.setProperties(msgInner, MessageDecoder.string2messageProperties(requestHeader.getProperties()));
    msgInner.setPropertiesString(requestHeader.getProperties());
    msgInner.setBornTimestamp(requestHeader.getBornTimestamp());
    msgInner.setBornHost(ctx.channel().remoteAddress());
    msgInner.setStoreHost(this.getStoreHost());
    msgInner.setReconsumeTimes(requestHeader.getReconsumeTimes() == null ? 0 : requestHeader.getReconsumeTimes());
    PutMessageResult putMessageResult = null;
    Map<String, String> oriProps = MessageDecoder.string2messageProperties(requestHeader.getProperties());
    String traFlag = oriProps.get(MessageConst.PROPERTY_TRANSACTION_PREPARED);
    if (traFlag != null && Boolean.parseBoolean(traFlag)) { 
        if (this.brokerController.getBrokerConfig().isRejectTransactionMessage()) { 
            response.setCode(ResponseCode.NO_PERMISSION);
            response.setRemark(
                "the broker[" + this.brokerController.getBrokerConfig().getBrokerIP1()
                    + "] sending transaction message is forbidden");
            return response;
        }
        putMessageResult = this.brokerController.getTransactionalMessageService().prepareMessage(msgInner);
    } else { 
        //进行消息存储
        putMessageResult = this.brokerController.getMessageStore().putMessage(msgInner);
    }
    return handlePutMessageResult(putMessageResult, response, request, msgInner, responseHeader, sendMessageContext, ctx, queueIdInt);
}

异步发送:调用发送API之后,无需阻塞等待消息服务器返回本次消息发送结果,只需要提供一个回调函数即可,异步方式相比较同步方式来说,性能有所提高,单位了保护消息服务器的压力,rocketmq对消息异步发送进行了并发限制,通过clientAsyncSemaphoreTimes-WhenSendAsyncFailed属性来控制消息重试次数,但是重试的调用入口是在收到服务端响应包时进行的,如果出现网络异常,网络超时等将不会进行重试
单向发送:调用消息发送api之后,无须等待消息服务器返回本次消息发送结果,并且无须提供回调函数,表示消息发送压根就不关心本次消息发送是否成功,

五.批量发送

批量发送消息是同一主题的多条消息一起打包发送到消息服务器,减少网络调用次数,提高网络传输效率
单条消息发送是,消息体的内容将保存在body中,批量消息发送,需要将多条消息消息体的内容存储在body中,Rocketmq采取的方式是,对单条消息内容使用固定格式进行存储 ———->单条和批量
总长度4字节->魔数4字节->bode CRC4字节->Flag4字节->body长度4字节->消息体N字节->属性长度2字节->扩展属性N字节
org.apache.rocketmq.client.producer.DefaultMQProducer#send(java.util.Collection)

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;
}

org.apache.rocketmq.common.message.MessageDecoder#encodeMessage

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();
}