RocketMQ高手之路系列之四：RocketMQ之消息发送（一）

 public Message(String topic, String tags, String keys, int flag, byte[] body, boolean waitStoreMsgOK) {
        this.topic = topic;
        this.flag = flag;
        this.body = body;
        if (tags != null && tags.length() > 0)
            this.setTags(tags);
        if (keys != null && keys.length() > 0)
            this.setKeys(keys);
        this.setWaitStoreMsgOK(waitStoreMsgOK);
    }

public class DefaultMQProducer extends ClientConfig implements MQProducer {
...
}

    /**
     * 生产者所属组
     */
  private String producerGroup;
    /**
     * 默认的topicKey
     */
    private String createTopicKey = MixAll.AUTO_CREATE_TOPIC_KEY_TOPIC;
    /**
     * 默认主题中的Broker队列的数量
     */
    private volatile int defaultTopicQueueNums = 4;
    /**
     * 消息发送默认超时时间为3s
     */
    private int sendMsgTimeout = 3000;
    /**
     * 消息体大小超过默认值则需要进行压缩，此处默认为4K
     */
    private int compressMsgBodyOverHowmuch = 1024 * 4;
    /**
     * 同步方式发送消息的重试次数，默认为为2次，总共会执行3次
     */
    private int retryTimesWhenSendFailed = 2;
    /**
     * 异方式发送消息的重试次数，默认为为2次
     */
    private int retryTimesWhenSendAsyncFailed = 2;
    /**
     * 消息重试时选择另外一个Broker，是否不等待存储结果就进行返回，默认为false
     */
    private boolean retryAnotherBrokerWhenNotStoreOK = false;
    /**
     * 允许发送消息的最大消息长度，默认为4M
     */
    private int maxMessageSize = 1024 * 1024 * 4; // 4M

public void start() throws MQClientException {
        this.defaultMQProducerImpl.start();
        if (null != traceDispatcher) {
            try {
                traceDispatcher.start(this.getNamesrvAddr());
            } 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;
        //检查producerGroup是否符合要求
                this.checkConfig();
        //修改生产者的instanceName为进程的ID
                if (!this.defaultMQProducer.getProducerGroup().equals(MixAll.CLIENT_INNER_PRODUCER_GROUP)) {
                    this.defaultMQProducer.changeInstanceNameToPID();
                }
        //创建MQClientInstance实例
                this.mQClientFactory = MQClientManager.getInstance().getAndCreateMQClientInstance(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;
        }
    //想所有的Broker发送心跳
        this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
    }

public MQClientInstance getAndCreateMQClientInstance(final ClientConfig clientConfig, RPCHook rpcHook) {
    //创建clientId
        String clientId = clientConfig.buildMQClientId();
        MQClientInstance instance = this.factoryTable.get(clientId);
        if (null == instance) {
            instance =
                new MQClientInstance(clientConfig.cloneClientConfig(),
                    this.factoryIndexGenerator.getAndIncrement(), clientId, rpcHook);
            //维护缓存表，一个clientId对应一个MQClientInstance
            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;
    }

  public String buildMQClientId() {
        StringBuilder sb = new StringBuilder();
        sb.append(this.getClientIP());
        sb.append("@");
        sb.append(this.getInstanceName());
        if (!UtilAll.isBlank(this.unitName)) {
            sb.append("@");
            sb.append(this.unitName);
        }
        return sb.toString();
    }