RocketMQ源码：producer 同步发送和单向发送

偏执的太偏执、 2024-03-25 14:54 116阅读 0赞

RocketMQ生产者发送消息分为三种模式，分别是`同步发送，异步发送和单向发送`。

*  `单向发送`，这个就是发送之后不用接收结果的，就是你发出去一个消息，然后就返回了，就算有结果返回也不会接收了，这是站在消息生产者的角度；
 *  `同步发送`的话，就是发出去一个消息，这个线程要等着它返回消息发送结果，然后你这个线程再根据这个消息发送结果再做一些业务操作等等；
 *  `异步发送`，这个就是在你发送消息之前要给一个`callback`，发送的时候，你这个线程就不用等着，该干什么就干什么，然后发送结果回来的时候，是由其他线程调用你这个callback来处理的，你可以把这个callback看作是一个回调函数，回调方法，这个方法里面的业务逻辑就是你对这个消息发送结果的处理。注意，本文介绍的消息发送只是普通的消息发送，那种事务类型的消息，我们以后会有介绍。

## 1. 同步发送 ##

`producer`同步发送消息的示例在`org.apache.rocketmq.example.simple.Producer`类中，代码如下：

public class Producer {
        public static void main(String[] args) throws MQClientException, InterruptedException {
            // 1. 创建 DefaultMQProducer 对象
            DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
    
           
            producer.setNamesrvAddr("127.0.0.1:9876");
            /*
             * Launch the instance.
             */
            // todo 2. 启动 producer
            producer.start();
    
            for (int i = 0; i < 1000; i++) {
                try {
    
                  
                    Message msg = new Message("TopicTest" /* Topic */,
                        "TagA" /* Tag */,
                        ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
                    );
    
    
                    // 3. 发送消息
                    SendResult sendResult = producer.send(msg);
    
                    System.out.printf("%s%n", sendResult);
                } 
                ...
            }
            producer.shutdown();
        }
    }
    复制代码

我们可以看到这个代码，你是同步消息你是需要在你自己的业务线程里面接收这个`sendResult`的，然后在做一些业务处理，比如我这里就是打印了一下这个`sendResult`。

接下来我们看下它是怎样发送的，这里是调用了这个`producer`的`send`方法。

@Override
    public SendResult send(
        Message msg) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
        // topic 和消息长度 校验
        Validators.checkMessage(msg, this);
        msg.setTopic(withNamespace(msg.getTopic()));
        // todo
        return this.defaultMQProducerImpl.send(msg);
    }
    复制代码

我们可以看到，这个 `DefaultMQProducer` 将这个消息给了`defaultMQProducerImpl` 这个实现的`send`方法来处理了。

public SendResult send(
        Message msg) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
        // todo 默认超时时间3s
        return send(msg, this.defaultMQProducer.getSendMsgTimeout());
    }
    复制代码

`defaultMQProducerImpl`的`send`方法，加了个超时时间 ，然后有调用它的重载方法`send(msg,timeout)`

public SendResult send(Message msg,
        long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
        // todo 同步模式
        return this.sendDefaultImpl(msg, CommunicationMode.SYNC, null, timeout);
    }
    复制代码

这个`send(msg,timeout)`又调用了`sendDefaultImpl` 方法，然后他这里加了个通信模式是同步，`CommunicationMode.SYNC`。

### 1.1 DefaultMQProducerImpl\#sendDefaultImpl ###

`sendDefaultImpl` 方法就比较长了了我们分成几部分来介绍：

private SendResult sendDefaultImpl(
        Message msg,
        final CommunicationMode communicationMode,
        final SendCallback sendCallback,
        final long timeout
    ) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
        // 判断状态是否是running
        this.makeSureStateOK();
        // 检查消息合法性
        Validators.checkMessage(msg, this.defaultMQProducer);
        // 随机的invokeID
        final long invokeID = random.nextLong();
        long beginTimestampFirst = System.currentTimeMillis();
        long beginTimestampPrev = beginTimestampFirst;
        long endTimestamp = beginTimestampFirst;
        // todo 获取topic信息
        TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
        ...
    }
    复制代码

这一小段代码其实就是做了一些准备检查工作，注意第二行的个检查消息合法性，它要检查你`topic`，消息长度的，你不能发空消息，消息长度也不能太长，默认是不超过`4m`，接下来这些就是记录一下时间了，再看最后一行，就是根据你这个消息发送的`topic`，然后获取`topic` 发送消息的这么一个信息，这里面就有这`topic` 有几个`MessageQueue`，然后每个`MessageQueue`对应在哪个`broker`上面，`broker` 的地址又是啥的，它这个方法会先从本地的一个缓存中获取下，没有的话就从`nameserv`更新下这个本地缓存，再找找，要是再找不到，它就认为你没有这个`topic`了，然后就去nameserv上面拉取一个默认topic的一些配置信息给你用（这个其实就是在`新建一个topic`）。 接着这个方法往下看，接着就是判断 这个`TopicPublishInfo`是否存在了，如果不存在的话就抛出异常了，没有后续了就，如果存在的话：

...
    if (topicPublishInfo != null && topicPublishInfo.ok()) {
        boolean callTimeout = false;
        MessageQueue mq = null;
        Exception exception = null;
        SendResult sendResult = null;
        // 重试次数 区分同步、其他
        int timesTotal = communicationMode == CommunicationMode.SYNC ? 1 + this.defaultMQProducer.getRetryTimesWhenSendFailed() : 1;
        int times = 0;
        // 存放发送过的broker name
        String[] brokersSent = new String[timesTotal];
        // 重试发送
        for (; times < timesTotal; times++) {
            String lastBrokerName = null == mq ? null : mq.getBrokerName();
            // todo 选择message queue
            MessageQueue mqSelected = this.selectOneMessageQueue(topicPublishInfo, lastBrokerName);
            if (mqSelected != null) {
                mq = mqSelected;
                brokersSent[times] = mq.getBrokerName();
                try {
                    beginTimestampPrev = System.currentTimeMillis();
                    if (times > 0) {
                        //Reset topic with namespace during resend.
                        msg.setTopic(this.defaultMQProducer.withNamespace(msg.getTopic()));
                    }
                    long costTime = beginTimestampPrev - beginTimestampFirst;
                    if (timeout < costTime) {
                        callTimeout = true;
                        break;
                    }
    
                    // todo 进行发送
                    sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout - costTime);
                    endTimestamp = System.currentTimeMillis();
                    // todo isolation 参数为false(看一下异常情况)
                    this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false);
                    switch (communicationMode) {
                        case ASYNC:
                            return null;
                        case ONEWAY:
                            return null;
                        case SYNC:
                            if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
                                if (this.defaultMQProducer.isRetryAnotherBrokerWhenNotStoreOK()) {
                                    continue;
                                }