引言
在上篇博文中，我们介绍了消息发送之前，消费生产者启动的流程。生产者启动后，就正式进入消息发送的的流程。本文主要阐述消息的发送的初步流程。
PS：消息生产者的代码模块在cilent模块中。如下：
- 消息发送基本流程
- 总结
一、消息发送基本流程
在介绍消息发送流程之前，我们先来看下RocketMQ的架构图，如下所示：
架构图中的各个模块的大致作用在前几篇文章中已经介绍过了，这里不再进行赘述。主要通过架构图让大家对RocketMQ中的消息发送有个整体的理解。
消息的发送方式主要包括三种方式：
（1）同步方式；
（2）异步方式：
（3）Oneway方式；
消息的发送流程主要步骤为：消息验证、路由信息查找以及消息发送三个基本步骤。在DefaultMQProducer中可以查看到对应的消息发送方法，如下所示：
@Override public SendResult send( Message msg) throws MQClientException, RemotingException, MQBrokerException, InterruptedException { return this.defaultMQProducerImpl.send(msg); } ... @Override public SendResult send(Message msg, long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException { return this.defaultMQProducerImpl.send(msg, timeout); } ... @Override public void send(Message msg, SendCallback sendCallback) throws MQClientException, RemotingException, InterruptedException { this.defaultMQProducerImpl.send(msg, sendCallback); }
大致的调用方式如下图所示：
消息发动的方式默认采用同步的方式进行，同时默认的超时时间为3s。在发送消息时，我们首先需要知道该消息要发送到哪里，这就像我们寄快递，需要先明确收件人是谁。那么在RocketMQ发送消息前会对消息进行基本的消息验证。确认消息是否符合发送要求。
Validators.checkMessage(msg, this.defaultMQProducer);
具体的代码如下所示：
public static void checkMessage(Message msg, DefaultMQProducer defaultMQProducer) throws MQClientException { //消息不能为空 if (null == msg) { throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL, "the message is null"); } // 主题不能为空 Validators.checkTopic(msg.getTopic()); // 消息的body不能为空 if (null == msg.getBody()) { throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL, "the message body is null"); } // 消息的body长度不能为0 if (0 == msg.getBody().length) { throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL, "the message body length is zero"); } // 消息的body长度不能唱过最大长度4M if (msg.getBody().length > defaultMQProducer.getMaxMessageSize()) { throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL, "the message body size over max value, MAX: " + defaultMQProducer.getMaxMessageSize()); } }
消息的校验没有问题则会调用tryToFindTopicPublishInfo(msg.getTopic())方法。我们需要获取主题的路由信息，通过路由信息我们才知道消息需要被投递到哪个具体的Broker节点之上。
我们来一起看下查找主题的路由信息方法，如下所示：
private TopicPublishInfo tryToFindTopicPublishInfo(final String topic) { //先从本地缓存变量topicPublishInfoTable中先get一次 TopicPublishInfo topicPublishInfo = this.topicPublishInfoTable.get(topic); if (null == topicPublishInfo || !topicPublishInfo.ok()) { this.topicPublishInfoTable.putIfAbsent(topic, new TopicPublishInfo()); //然后从nameServer上更新topic路由信息 this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic); //然后再从本地缓存变量topicPublishInfoTable中再get一次 topicPublishInfo = this.topicPublishInfoTable.get(topic); } if (topicPublishInfo.isHaveTopicRouterInfo() || topicPublishInfo.ok()) { return topicPublishInfo; } else { //第一次的时候isDefault为false，第二次的时候default为true，即为用默认的topic的参数进行更新 this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic, true, this.defaultMQProducer); topicPublishInfo = this.topicPublishInfoTable.get(topic); return topicPublishInfo; } }
在生产者中如果缓存了topic路由信息，路由信息中如果包含了消息队列，那么会进行路由信息的返回。如果没有缓存或者没有队列信息，那么就会向NameServer查询topic的路由信息。TopicPublishInfo属性如下所示：
public class TopicPublishInfo { //是否是顺序消息 private boolean orderTopic = false; //是否包含主题路由信息 private boolean haveTopicRouterInfo = false; //主题队列的消息队列 private List<MessageQueue> messageQueueList = new ArrayList<MessageQueue>(); // 没选择一次消息队列，值自增1 private volatile ThreadLocalIndex sendWhichQueue = new ThreadLocalIndex(); private TopicRouteData topicRouteData; public boolean isOrderTopic() { return orderTopic; } public void setOrderTopic(boolean orderTopic) { this.orderTopic = orderTopic; } public boolean ok() { return null != this.messageQueueList && !this.messageQueueList.isEmpty(); } public List<MessageQueue> getMessageQueueList() { return messageQueueList; } public void setMessageQueueList(List<MessageQueue> messageQueueList) { this.messageQueueList = messageQueueList; } public ThreadLocalIndex getSendWhichQueue() { return sendWhichQueue; } public void setSendWhichQueue(ThreadLocalIndex sendWhichQueue) { this.sendWhichQueue = sendWhichQueue; } public boolean isHaveTopicRouterInfo() { return haveTopicRouterInfo; } public void setHaveTopicRouterInfo(boolean haveTopicRouterInfo) { this.haveTopicRouterInfo = haveTopicRouterInfo; } public MessageQueue selectOneMessageQueue(final String lastBrokerName) { if (lastBrokerName == null) { return selectOneMessageQueue(); } else { 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; 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); } public int getQueueIdByBroker(final String brokerName) { for (int i = 0; i < topicRouteData.getQueueDatas().size(); i++) { final QueueData queueData = this.topicRouteData.getQueueDatas().get(i); if (queueData.getBrokerName().equals(brokerName)) { return queueData.getWriteQueueNums(); } } return -1; } @Override public String toString() { return "TopicPublishInfo [orderTopic=" + orderTopic + ", messageQueueList=" + messageQueueList + ", sendWhichQueue=" + sendWhichQueue + ", haveTopicRouterInfo=" + haveTopicRouterInfo + "]"; } public TopicRouteData getTopicRouteData() { return topicRouteData; } public void setTopicRouteData(final TopicRouteData topicRouteData) { this.topicRouteData = topicRouteData; } }
二、总结
本文主要介绍了消息发送前的准备，包括了消息的有效性验证以及路由信息查找以确认消息投递到哪个具体的Broker节点之上。下篇文章将正式阐述消息的发送。
