rocketmq-producer原理解析

Producer随机与一个NameServer建立长连接，从NameServer获取topic的最新队列情况。Producer会向提供topic服务的master建立长连接，且定时向master发送心跳。 发送消息demo:

// 构造Producer DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName"); // 初始化Producer，整个应用生命周期内，只需要初始化1次 producer.start(); // 构造Message Message msg = new Message("TopicTest1",// topic "TagA",// tag：给消息打标签,用于区分一类消息，可为null "OrderID188",// key：自定义Key，可以用于去重，可为null ("Hello MetaQ").getBytes());// body：消息内容 // 发送消息并返回结果 SendResult sendResult = producer.send(msg); // 清理资源，关闭网络连接，注销自己 producer.shutdown();

一：start() 在整个应用生命周期内，生产者只需要调用一次start方法来初始化以下工作：

如果没有指定namesrv地址，将会自动寻址启动定时任务：从namsrv更新topic路由信息、清理已经挂掉的broker、向所有在线的broker master发送心跳…启动负载均衡的服务：producer根据roundbin方式轮询topic下的所有队列来实现发送方的负载均衡。

二：DefaultMQProducer send的内部实现原理伪代码如下：

private SendResult sendDefaultImpl(Message msg,......) { // 检查Producer的状态是否是RUNNING this.makeSureStateOK(); // 检查msg是否合法：是否为null、topic,body是否为空、body是否超长 Validators.checkMessage(msg, this.defaultMQProducer); // 获取topic路由信息 TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic()); // 从路由信息中选择一个消息队列 MessageQueue mq = topicPublishInfo.selectOneMessageQueue(lastBrokerName); // 将消息发送到该队列上去 sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, timeout); }

1：producer初始化的时候会从namesrv获取topic的路由信息更新到本地缓存，所以tryToFindTopicPublishInfo会先从本地缓存取，如果没取到再从namesrv获取最新的路由信息。

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 != null && topicPublishInfo.ok())) { return topicPublishInfo; } else { this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic, true, this.defaultMQProducer); topicPublishInfo = this.topicPublishInfoTable.get(topic); return topicPublishInfo; } }

2.发送消息时，当没有指定队列或队列选择器时，调用selectOneMessageQueue：使用轮询方式，返回一个队列。当指定队列选择器时，通过selector.select(topicPublishInfo.getMessageQueueList(), msg, arg);获取队列。 3.消息发送 CommunicationMode: SYNC:同步（默认）, ASYNC：异步（callback）, ONEWAY：单向,

private SendResult sendKernelImpl(final Message msg, final MessageQueue mq, final CommunicationMode communicationMode, ......){ String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName()); SendMessageRequestHeader requestHeader = new SendMessageRequestHeader(); switch (communicationMode) { case ASYNC: sendResult = this.mQClientFactory.getMQClientAPIImpl().sendMessage( brokerAddr, // 1 mq.getBrokerName(), // 2 msg, // 3 requestHeader, // 4 timeout, // 5 communicationMode, // 6 sendCallback, // 7 topicPublishInfo, // 8 this.mQClientFactory, // 9 this.defaultMQProducer.getRetryTimesWhenSendAsyncFailed(), //10 context, // this); break; case ONEWAY: case SYNC: .... break; }