RocketMQ源码详解(NameServer、Producer)

NameServer

架构设计

消息中间件的设计思路一般是基于主题订阅发布的机制，消息生产者（Producer）发送某一个主题到消息服务器，消息服务器负责将消息持久化存储，消息消费者（Consumer）订阅该兴趣的主题，消息服务器根据订阅信息（路由信息）将消息推送到消费者（Push模式）或者消费者主动向消息服务器拉去（Pull模式），从而实现消息生产者与消息消费者解耦。为了避免消息服务器的单点故障导致的整个系统瘫痪，通常会部署多台消息服务器共同承担消息的存储。那消息生产者如何知道消息要发送到哪台消息服务器呢？如果某一台消息服务器宕机了，那么消息生产者如何在不重启服务情况下感知呢？
NameServer就是为了解决以上问题设计的。

• Broker消息服务器在启动的时向所有NameServer注册
  
• 消息生产者（Producer）在发送消息时之前先从NameServer获取Broker服务器地址列表，然后根据负载均衡算法从列表中选择一台服务器进行发送。
  
• NameServer与每台Broker保持长连接，并间隔30S检测Broker是否存活，如果检测到Broker宕机，则从路由注册表中删除。但是路由变化不会马上通知消息生产者。这样设计的目的是为了降低NameServer实现的复杂度，在消息发送端提供容错机制保证消息发送的可用性。
  

NameServer本身的高可用是通过部署多台NameServer来实现，但彼此之间不通讯，也就是NameServer服务器之间在某一个时刻的数据并不完全相同，但这对消息发送并不会造成任何影响，这也是NameServer设计的一个亮点，总之，RocketMQ设计追求简单高效。
启动流程

启动类：org.apache.rocketmq.namesrv.NamesrvStartup
步骤一

解析配置文件，填充NameServerConfig、NettyServerConfig属性值，并创建NamesrvController
代码：NamesrvController#createNamesrvController

1. //创建NamesrvConfig
2. final NamesrvConfig namesrvConfig = new NamesrvConfig();
3. //创建NettyServerConfig
4. final NettyServerConfig nettyServerConfig = new NettyServerConfig();
5. //设置启动端口号
6. nettyServerConfig.setListenPort(9876);
7. //解析启动-c参数
8. if (commandLine.hasOption('c')) {
9.     String file = commandLine.getOptionValue('c');
10.     if (file != null) {
11.         InputStream in = new BufferedInputStream(new FileInputStream(file));
12.         properties = new Properties();
13.         properties.load(in);
14.         MixAll.properties2Object(properties, namesrvConfig);
15.         MixAll.properties2Object(properties, nettyServerConfig);
17.         namesrvConfig.setConfigStorePath(file);
19.         System.out.printf("load config properties file OK, %s%n", file);
20.         in.close();
21.     }
22. }
23. //解析启动-p参数
24. if (commandLine.hasOption('p')) {
25.     InternalLogger console = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_CONSOLE_NAME);
26.     MixAll.printObjectProperties(console, namesrvConfig);
27.     MixAll.printObjectProperties(console, nettyServerConfig);
28.     System.exit(0);
29. }
30. //将启动参数填充到namesrvConfig,nettyServerConfig
31. MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), namesrvConfig);
33. //创建NameServerController
34. final NamesrvController controller = new NamesrvController(namesrvConfig, nettyServerConfig);

复制代码

NamesrvConfig属性

1. private String rocketmqHome = System.getProperty(MixAll.ROCKETMQ_HOME_PROPERTY, System.getenv(MixAll.ROCKETMQ_HOME_ENV));
2. private String kvConfigPath = System.getProperty("user.home") + File.separator + "namesrv" + File.separator + "kvConfig.json";
3. private String configStorePath = System.getProperty("user.home") + File.separator + "namesrv" + File.separator + "namesrv.properties";
4. private String productEnvName = "center";
5. private boolean clusterTest = false;
6. private boolean orderMessageEnable = false;

复制代码

rocketmqHome：rocketmq主目录
kvConfig：NameServer存储KV配置属性的持久化路径
configStorePath：nameServer默认配置文件路径
orderMessageEnable：是否支持顺序消息
NettyServerConfig属性

1. private int listenPort = 8888;
2. private int serverWorkerThreads = 8;
3. private int serverCallbackExecutorThreads = 0;
4. private int serverSelectorThreads = 3;
5. private int serverOnewaySemaphoreValue = 256;
6. private int serverAsyncSemaphoreValue = 64;
7. private int serverChannelMaxIdleTimeSeconds = 120;
8. private int serverSocketSndBufSize = NettySystemConfig.socketSndbufSize;
9. private int serverSocketRcvBufSize = NettySystemConfig.socketRcvbufSize;
10. private boolean serverPooledByteBufAllocatorEnable = true;
11. private boolean useEpollNativeSelector = false;

复制代码

listenPort：NameServer监听端口，该值默认会被初始化为9876
serverWorkerThreads：Netty业务线程池线程个数
serverCallbackExecutorThreads：Netty public任务线程池线程个数，Netty网络设计，根据业务类型会创建不同的线程池，比如处理消息发送、消息消费、心跳检测等。如果该业务类型未注册线程池，则由public线程池执行。
serverSelectorThreads：IO线程池个数，主要是NameServer、Broker端解析请求、返回相应的线程个数，这类线程主要是处理网路请求的，解析请求包，然后转发到各个业务线程池完成具体的操作，然后将结果返回给调用方;
serverOnewaySemaphoreValue：send oneway消息请求并发读（Broker端参数）;
serverAsyncSemaphoreValue：异步消息发送最大并发度;
serverChannelMaxIdleTimeSeconds ：网络连接最大的空闲时间，默认120s。
serverSocketSndBufSize：网络socket发送缓冲区大小。
serverSocketRcvBufSize： 网络接收端缓存区大小。
serverPooledByteBufAllocatorEnable：ByteBuffer是否开启缓存;
useEpollNativeSelector：是否启用Epoll IO模型。
步骤二

根据启动属性创建NamesrvController实例，并初始化该实例。NameServerController实例为NameServer核心控制器
代码：NamesrvController#initialize

1. public boolean initialize() {
2.         //加载KV配置
3.     this.kvConfigManager.load();
4.         //创建NettyServer网络处理对象
5.     this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);
6.         //开启定时任务:每隔10s扫描一次Broker,移除不活跃的Broker
7.     this.remotingExecutor =
8.         Executors.newFixedThreadPool(nettyServerConfig.getServerWorkerThreads(), new ThreadFactoryImpl("RemotingExecutorThread_"));
9.     this.registerProcessor();
10.     this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
11.         @Override
12.         public void run() {
13.             NamesrvController.this.routeInfoManager.scanNotActiveBroker();
14.         }
15.     }, 5, 10, TimeUnit.SECONDS);
16.         //开启定时任务:每隔10min打印一次KV配置
17.         this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
19.         @Override
20.         public void run() {
21.             NamesrvController.this.kvConfigManager.printAllPeriodically();
22.         }
23.     }, 1, 10, TimeUnit.MINUTES);
24.     return true;
25. }

复制代码

步骤三

在JVM进程关闭之前，先将线程池关闭，及时释放资源
代码：NamesrvStartup#start

1. //注册JVM钩子函数代码
2. Runtime.getRuntime().addShutdownHook(new ShutdownHookThread(log, new Callable<Void>() {
3.     @Override
4.     public Void call() throws Exception {
5.         //释放资源
6.         controller.shutdown();
7.         return null;
8.     }
9. }));

复制代码

路由管理

NameServer的主要作用是为消息的生产者和消息消费者提供关于主题Topic的路由信息，那么NameServer需要存储路由的基础信息，还要管理Broker节点，包括路由注册、路由删除等。
路由元信息

代码：RouteInfoManager

1. private final HashMap<String/* topic */, List<QueueData>> topicQueueTable;
2. private final HashMap<String/* brokerName */, BrokerData> brokerAddrTable;
3. private final HashMap<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable;
4. private final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;
5. private final HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;

复制代码

topicQueueTable：Topic消息队列路由信息，消息发送时根据路由表进行负载均衡
brokerAddrTable：Broker基础信息，包括brokerName、所属集群名称、主备Broker地址
clusterAddrTable：Broker集群信息，存储集群中所有Broker名称
brokerLiveTable：Broker状态信息，NameServer每次收到心跳包是会替换该信息
filterServerTable：Broker上的FilterServer列表，用于类模式消息过滤。

RocketMQ基于定于发布机制，一个Topic拥有多个消息队列，一个Broker为每一个主题创建4个读队列和4个写队列。多个Broker组成一个集群，集群由相同的多台Broker组成Master-Slave架构，brokerId为0代表Master，大于0为Slave。BrokerLiveInfo中的lastUpdateTimestamp存储上次收到Broker心跳包的时间。

路由注册

发送心跳包

RocketMQ路由注册是通过Broker与NameServer的心跳功能实现的。Broker启动时向集群中所有的NameServer发送心跳信息，每隔30s向集群中所有NameServer发送心跳包，NameServer收到心跳包时会更新brokerLiveTable缓存中BrokerLiveInfo的lastUpdataTimeStamp信息，然后NameServer每隔10s扫描brokerLiveTable，如果连续120S没有收到心跳包，NameServer将移除Broker的路由信息同时关闭Socket连接。
代码：BrokerController#start

1. //注册Broker信息
2. this.registerBrokerAll(true, false, true);
3. //每隔30s上报Broker信息到NameServer
4. this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
6.     @Override
7.     public void run() {
8.         try {
9.             BrokerController.this.registerBrokerAll(true, false, brokerConfig.isForceRegister());
10.         } catch (Throwable e) {
11.             log.error("registerBrokerAll Exception", e);
12.         }
13.     }
14. }, 1000 * 10, Math.max(10000, Math.min(brokerConfig.getRegisterNameServerPeriod(), 60000)),
15.                                                   TimeUnit.MILLISECONDS);

复制代码

代码：BrokerOuterAPI#registerBrokerAll

1. //获得nameServer地址信息
2. List<String> nameServerAddressList = this.remotingClient.getNameServerAddressList();
3. //遍历所有nameserver列表
4. if (nameServerAddressList != null && nameServerAddressList.size() > 0) {
6.     //封装请求头
7.     final RegisterBrokerRequestHeader requestHeader = new RegisterBrokerRequestHeader();
8.     requestHeader.setBrokerAddr(brokerAddr);
9.     requestHeader.setBrokerId(brokerId);
10.     requestHeader.setBrokerName(brokerName);
11.     requestHeader.setClusterName(clusterName);
12.     requestHeader.setHaServerAddr(haServerAddr);
13.     requestHeader.setCompressed(compressed);
14.         //封装请求体
15.     RegisterBrokerBody requestBody = new RegisterBrokerBody();
16.     requestBody.setTopicConfigSerializeWrapper(topicConfigWrapper);
17.     requestBody.setFilterServerList(filterServerList);
18.     final byte[] body = requestBody.encode(compressed);
19.     final int bodyCrc32 = UtilAll.crc32(body);
20.     requestHeader.setBodyCrc32(bodyCrc32);
21.     final CountDownLatch countDownLatch = new CountDownLatch(nameServerAddressList.size());
22.     for (final String namesrvAddr : nameServerAddressList) {
23.         brokerOuterExecutor.execute(new Runnable() {
24.             @Override
25.             public void run() {
26.                 try {
27.                     //分别向NameServer注册
28.                     RegisterBrokerResult result = registerBroker(namesrvAddr,oneway, timeoutMills,requestHeader,body);
29.                     if (result != null) {
30.                         registerBrokerResultList.add(result);
31.                     }
33.                     log.info("register broker[{}]to name server {} OK", brokerId, namesrvAddr);
34.                 } catch (Exception e) {
35.                     log.warn("registerBroker Exception, {}", namesrvAddr, e);
36.                 } finally {
37.                     countDownLatch.countDown();
38.                 }
39.             }
40.         });
41.     }
43.     try {
44.         countDownLatch.await(timeoutMills, TimeUnit.MILLISECONDS);
45.     } catch (InterruptedException e) {
46.     }
47. }

复制代码

代码：BrokerOutAPI#registerBroker

1. if (oneway) {
2.     try {
3.         this.remotingClient.invokeOneway(namesrvAddr, request, timeoutMills);
4.     } catch (RemotingTooMuchRequestException e) {
5.         // Ignore
6.     }
7.     return null;
8. }
9. RemotingCommand response = this.remotingClient.invokeSync(namesrvAddr, request, timeoutMills);

复制代码

处理心跳包

org.apache.rocketmq.namesrv.processor.DefaultRequestProcessor网路处理类解析请求类型，如果请求类型是为REGISTER_BROKER，则将请求转发到RouteInfoManager#regiesterBroker
代码：DefaultRequestProcessor#processRequest

1. //判断是注册Broker信息
2. case RequestCode.REGISTER_BROKER:
3.         Version brokerVersion = MQVersion.value2Version(request.getVersion());
4.         if (brokerVersion.ordinal() >= MQVersion.Version.V3_0_11.ordinal()) {
5.             return this.registerBrokerWithFilterServer(ctx, request);
6.         } else {
7.         //注册Broker信息
8.             return this.registerBroker(ctx, request);
9.         }

复制代码

代码：DefaultRequestProcessor#registerBroker

1. RegisterBrokerResult result = this.namesrvController.getRouteInfoManager().registerBroker(
2.     requestHeader.getClusterName(),
3.     requestHeader.getBrokerAddr(),
4.     requestHeader.getBrokerName(),
5.     requestHeader.getBrokerId(),
6.     requestHeader.getHaServerAddr(),
7.     topicConfigWrapper,
8.     null,
9.     ctx.channel()
10. );

复制代码

代码：RouteInfoManager#registerBroker
维护路由信息

1. //加锁
2. this.lock.writeLock().lockInterruptibly();
3. //维护clusterAddrTable
4. Set<String> brokerNames = this.clusterAddrTable.get(clusterName);
5. if (null == brokerNames) {
6.     brokerNames = new HashSet<String>();
7.     this.clusterAddrTable.put(clusterName, brokerNames);
8. }
9. brokerNames.add(brokerName);

复制代码

1. //维护brokerAddrTable
2. BrokerData brokerData = this.brokerAddrTable.get(brokerName);
3. //第一次注册,则创建brokerData
4. if (null == brokerData) {
5.     registerFirst = true;
6.     brokerData = new BrokerData(clusterName, brokerName, new HashMap<Long, String>());
7.     this.brokerAddrTable.put(brokerName, brokerData);
8. }
9. //非第一次注册,更新Broker
10. Map<Long, String> brokerAddrsMap = brokerData.getBrokerAddrs();
11. Iterator<Entry<Long, String>> it = brokerAddrsMap.entrySet().iterator();
12. while (it.hasNext()) {
13.     Entry<Long, String> item = it.next();
14.     if (null != brokerAddr && brokerAddr.equals(item.getValue()) && brokerId != item.getKey()) {
15.         it.remove();
16.     }
17. }
18. String oldAddr = brokerData.getBrokerAddrs().put(brokerId, brokerAddr);
19. registerFirst = registerFirst || (null == oldAddr);

复制代码

1. //维护topicQueueTable
2. if (null != topicConfigWrapper && MixAll.MASTER_ID == brokerId) {
3.     if (this.isBrokerTopicConfigChanged(brokerAddr, topicConfigWrapper.getDataVersion()) ||
4.         registerFirst) {
5.         ConcurrentMap<String, TopicConfig> tcTable = topicConfigWrapper.getTopicConfigTable();
6.         if (tcTable != null) {
7.             for (Map.Entry<String, TopicConfig> entry : tcTable.entrySet()) {
8.                 this.createAndUpdateQueueData(brokerName, entry.getValue());
9.             }
10.         }
11.     }
12. }

复制代码

代码：RouteInfoManager#createAndUpdateQueueData

1. private void createAndUpdateQueueData(final String brokerName, final TopicConfig topicConfig) {
2.     //创建QueueData
3.         QueueData queueData = new QueueData();
4.         queueData.setBrokerName(brokerName);
5.         queueData.setWriteQueueNums(topicConfig.getWriteQueueNums());
6.         queueData.setReadQueueNums(topicConfig.getReadQueueNums());
7.         queueData.setPerm(topicConfig.getPerm());
8.         queueData.setTopicSynFlag(topicConfig.getTopicSysFlag());
9.         //获得topicQueueTable中队列集合
10.         List<QueueData> queueDataList = this.topicQueueTable.get(topicConfig.getTopicName());
11.     //topicQueueTable为空,则直接添加queueData到队列集合
12.         if (null == queueDataList) {
13.             queueDataList = new LinkedList<QueueData>();
14.             queueDataList.add(queueData);
15.             this.topicQueueTable.put(topicConfig.getTopicName(), queueDataList);
16.             log.info("new topic registered, {} {}", topicConfig.getTopicName(), queueData);
17.         } else {
18.         //判断是否是新的队列
19.             boolean addNewOne = true;
20.             Iterator<QueueData> it = queueDataList.iterator();
21.             while (it.hasNext()) {
22.                 QueueData qd = it.next();
23.             //如果brokerName相同,代表不是新的队列
24.                 if (qd.getBrokerName().equals(brokerName)) {
25.                     if (qd.equals(queueData)) {
26.                         addNewOne = false;
27.                 } else {
28.                             log.info("topic changed, {} OLD: {} NEW: {}", topicConfig.getTopicName(), qd,
29.                                 queueData);
30.                             it.remove();
31.                         }
32.                     }
33.                 }
34.                 //如果是新的队列,则添加队列到queueDataList
35.         if (addNewOne) {
36.             queueDataList.add(queueData);
37.         }
38.     }
39. }

复制代码

1. //维护brokerLiveTable
2. BrokerLiveInfo prevBrokerLiveInfo = this.brokerLiveTable.put(brokerAddr,new BrokerLiveInfo(
3.     System.currentTimeMillis(),
4.     topicConfigWrapper.getDataVersion(),
5.     channel,
6.     haServerAddr));

复制代码

1. //维护filterServerList
2. if (filterServerList != null) {
3.     if (filterServerList.isEmpty()) {
4.         this.filterServerTable.remove(brokerAddr);
5.     } else {
6.         this.filterServerTable.put(brokerAddr, filterServerList);
7.     }
8. }
10. if (MixAll.MASTER_ID != brokerId) {
11.     String masterAddr = brokerData.getBrokerAddrs().get(MixAll.MASTER_ID);
12.     if (masterAddr != null) {
13.         BrokerLiveInfo brokerLiveInfo = this.brokerLiveTable.get(masterAddr);
14.         if (brokerLiveInfo != null) {
15.             result.setHaServerAddr(brokerLiveInfo.getHaServerAddr());
16.             result.setMasterAddr(masterAddr);
17.         }
18.     }
19. }

复制代码

路由删除

Broker每隔30s向NameServer发送一个心跳包，心跳包包含BrokerId，Broker地址，Broker名称，Broker所属集群名称、Broker关联的FilterServer列表。但是如果Broker宕机，NameServer无法收到心跳包，此时NameServer如何来剔除这些失效的Broker呢？NameServer会每隔10s扫描brokerLiveTable状态表，如果BrokerLive的lastUpdateTimestamp的时间戳距当前时间超过120s，则认为Broker失效，移除该Broker，关闭与Broker连接，同时更新topicQueueTable、brokerAddrTable、brokerLiveTable、filterServerTable。
RocketMQ有两个触发点来删除路由信息：

• NameServer定期扫描brokerLiveTable检测上次心跳包与当前系统的时间差，如果时间超过120s，则需要移除broker。
  
• Broker在正常关闭的情况下，会执行unregisterBroker指令
  

这两种方式路由删除的方法都是一样的，就是从相关路由表中删除与该broker相关的信息。

代码：NamesrvController#initialize

1. //每隔10s扫描一次为活跃Broker
2. this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
4.     @Override
5.     public void run() {
6.         NamesrvController.this.routeInfoManager.scanNotActiveBroker();
7.     }
8. }, 5, 10, TimeUnit.SECONDS);

复制代码

代码：RouteInfoManager#scanNotActiveBroker

1. public void scanNotActiveBroker() {
2.     //获得brokerLiveTable
3.     Iterator<Entry<String, BrokerLiveInfo>> it = this.brokerLiveTable.entrySet().iterator();
4.     //遍历brokerLiveTable
5.     while (it.hasNext()) {
6.         Entry<String, BrokerLiveInfo> next = it.next();
7.         long last = next.getValue().getLastUpdateTimestamp();
8.         //如果收到心跳包的时间距当时时间是否超过120s
9.         if ((last + BROKER_CHANNEL_EXPIRED_TIME) < System.currentTimeMillis()) {
10.             //关闭连接
11.             RemotingUtil.closeChannel(next.getValue().getChannel());
12.             //移除broker
13.             it.remove();
14.             //维护路由表
15.             this.onChannelDestroy(next.getKey(), next.getValue().getChannel());
16.         }
17.     }
18. }

复制代码

代码：RouteInfoManager#onChannelDestroy

1. //申请写锁,根据brokerAddress从brokerLiveTable和filterServerTable移除
2. this.lock.writeLock().lockInterruptibly();
3. this.brokerLiveTable.remove(brokerAddrFound);
4. this.filterServerTable.remove(brokerAddrFound);

复制代码

1. //维护brokerAddrTable
2. String brokerNameFound = null;
3. boolean removeBrokerName = false;
4. Iterator<Entry<String, BrokerData>> itBrokerAddrTable =this.brokerAddrTable.entrySet().iterator();
5. //遍历brokerAddrTable
6. while (itBrokerAddrTable.hasNext() && (null == brokerNameFound)) {
7.     BrokerData brokerData = itBrokerAddrTable.next().getValue();
8.     //遍历broker地址
9.     Iterator<Entry<Long, String>> it = brokerData.getBrokerAddrs().entrySet().iterator();
10.     while (it.hasNext()) {
11.         Entry<Long, String> entry = it.next();
12.         Long brokerId = entry.getKey();
13.         String brokerAddr = entry.getValue();
14.         //根据broker地址移除brokerAddr
15.         if (brokerAddr.equals(brokerAddrFound)) {
16.             brokerNameFound = brokerData.getBrokerName();
17.             it.remove();
18.             log.info("remove brokerAddr[{}, {}] from brokerAddrTable, because channel destroyed",
19.                 brokerId, brokerAddr);
20.             break;
21.         }
22.     }
23.         //如果当前主题只包含待移除的broker,则移除该topic
24.     if (brokerData.getBrokerAddrs().isEmpty()) {
25.         removeBrokerName = true;
26.         itBrokerAddrTable.remove();
27.         log.info("remove brokerName[{}] from brokerAddrTable, because channel destroyed",
28.             brokerData.getBrokerName());
29.     }
30. }

复制代码

1. //维护clusterAddrTable
2. if (brokerNameFound != null && removeBrokerName) {
3.     Iterator<Entry<String, Set<String>>> it = this.clusterAddrTable.entrySet().iterator();
4.     //遍历clusterAddrTable
5.     while (it.hasNext()) {
6.         Entry<String, Set<String>> entry = it.next();
7.         //获得集群名称
8.         String clusterName = entry.getKey();
9.         //获得集群中brokerName集合
10.         Set<String> brokerNames = entry.getValue();
11.         //从brokerNames中移除brokerNameFound
12.         boolean removed = brokerNames.remove(brokerNameFound);
13.         if (removed) {
14.             log.info("remove brokerName[{}], clusterName[{}] from clusterAddrTable, because channel destroyed",
15.                 brokerNameFound, clusterName);
17.             if (brokerNames.isEmpty()) {
18.                 log.info("remove the clusterName[{}] from clusterAddrTable, because channel destroyed and no broker in this cluster",
19.                     clusterName);
20.                 //如果集群中不包含任何broker,则移除该集群
21.                 it.remove();
22.             }
24.             break;
25.         }
26.     }
27. }

复制代码

1. //维护topicQueueTable队列
2. if (removeBrokerName) {
3.     //遍历topicQueueTable
4.     Iterator<Entry<String, List<QueueData>>> itTopicQueueTable =
5.         this.topicQueueTable.entrySet().iterator();
6.     while (itTopicQueueTable.hasNext()) {
7.         Entry<String, List<QueueData>> entry = itTopicQueueTable.next();
8.         //主题名称
9.         String topic = entry.getKey();
10.         //队列集合
11.         List<QueueData> queueDataList = entry.getValue();
12.                 //遍历该主题队列
13.         Iterator<QueueData> itQueueData = queueDataList.iterator();
14.         while (itQueueData.hasNext()) {
15.             //从队列中移除为活跃broker信息
16.             QueueData queueData = itQueueData.next();
17.             if (queueData.getBrokerName().equals(brokerNameFound)) {
18.                 itQueueData.remove();
19.                 log.info("remove topic[{} {}], from topicQueueTable, because channel destroyed",
20.                     topic, queueData);
21.             }
22.         }
23.                 //如果该topic的队列为空,则移除该topic
24.         if (queueDataList.isEmpty()) {
25.             itTopicQueueTable.remove();
26.             log.info("remove topic[{}] all queue, from topicQueueTable, because channel destroyed",
27.                 topic);
28.         }
29.     }
30. }

复制代码

1. //释放写锁
2. finally {
3.     this.lock.writeLock().unlock();
4. }

复制代码

路由发现

RocketMQ路由发现是非实时的，当Topic路由出现变化后，NameServer不会主动推送给客户端，而是由客户端定时拉取主题最新的路由。
代码：DefaultRequestProcessor#getRouteInfoByTopic

1. public RemotingCommand getRouteInfoByTopic(ChannelHandlerContext ctx,
2.     RemotingCommand request) throws RemotingCommandException {
3.     final RemotingCommand response = RemotingCommand.createResponseCommand(null);
4.     final GetRouteInfoRequestHeader requestHeader =
5.         (GetRouteInfoRequestHeader) request.decodeCommandCustomHeader(GetRouteInfoRequestHeader.class);
6.         //调用RouteInfoManager的方法,从路由表topicQueueTable、brokerAddrTable、filterServerTable中分别填充TopicRouteData的List<QueueData>、List<BrokerData>、filterServer
7.     TopicRouteData topicRouteData = this.namesrvController.getRouteInfoManager().pickupTopicRouteData(requestHeader.getTopic());
8.         //如果找到主题对应你的路由信息并且该主题为顺序消息，则从NameServer KVConfig中获取关于顺序消息相关的配置填充路由信息
9.     if (topicRouteData != null) {
10.         if (this.namesrvController.getNamesrvConfig().isOrderMessageEnable()) {
11.             String orderTopicConf =
12.                 this.namesrvController.getKvConfigManager().getKVConfig(NamesrvUtil.NAMESPACE_ORDER_TOPIC_CONFIG,
13.                     requestHeader.getTopic());
14.             topicRouteData.setOrderTopicConf(orderTopicConf);
15.         }
17.         byte[] content = topicRouteData.encode();
18.         response.setBody(content);
19.         response.setCode(ResponseCode.SUCCESS);
20.         response.setRemark(null);
21.         return response;
22.     }
24.     response.setCode(ResponseCode.TOPIC_NOT_EXIST);
25.     response.setRemark("No topic route info in name server for the topic: " + requestHeader.getTopic()
26.         + FAQUrl.suggestTodo(FAQUrl.APPLY_TOPIC_URL));
27.     return response;
28. }

复制代码

小结

Producer

消息生产者的代码都在client模块中，相对于RocketMQ来讲，消息生产者就是客户端，也是消息的提供者。

方法和属性

1）主要方法介绍

1.   //创建主题
2.   void createTopic(final String key, final String newTopic, final int queueNum) throws MQClientException;

复制代码

1.   //根据时间戳从队列中查找消息偏移量
2.   long searchOffset(final MessageQueue mq, final long timestamp)

复制代码

1.   //查找消息队列中最大的偏移量
2.   long maxOffset(final MessageQueue mq) throws MQClientException;

复制代码

1.   //查找消息队列中最小的偏移量
2.   long minOffset(final MessageQueue mq)

复制代码

1.   //根据偏移量查找消息
2.   MessageExt viewMessage(final String offsetMsgId) throws RemotingException, MQBrokerException,
3.           InterruptedException, MQClientException;

复制代码

1.   //根据条件查找消息
2.   QueryResult queryMessage(final String topic, final String key, final int maxNum, final long begin,
3.           final long end) throws MQClientException, InterruptedException;

复制代码

1.   //根据消息ID和主题查找消息
2.   MessageExt viewMessage(String topic,String msgId) throws RemotingException, MQBrokerException, InterruptedException, MQClientException;

复制代码

1.   //启动
2.   void start() throws MQClientException;

复制代码

1.   //关闭
2.   void shutdown();

复制代码

1.   //查找该主题下所有消息
2.   List<MessageQueue> fetchPublishMessageQueues(final String topic) throws MQClientException;

复制代码

1.   //同步发送消息
2.   SendResult send(final Message msg) throws MQClientException, RemotingException, MQBrokerException,
3.           InterruptedException;

复制代码

1.   //同步超时发送消息
2.   SendResult send(final Message msg, final long timeout) throws MQClientException,
3.           RemotingException, MQBrokerException, InterruptedException;

复制代码

1.   //异步发送消息
2.   void send(final Message msg, final SendCallback sendCallback) throws MQClientException,
3.           RemotingException, InterruptedException;

复制代码

1.   //异步超时发送消息
2.   void send(final Message msg, final SendCallback sendCallback, final long timeout)
3.       throws MQClientException, RemotingException, InterruptedException;

复制代码

1.   //发送单向消息
2.   void sendOneway(final Message msg) throws MQClientException, RemotingException,
3.       InterruptedException;

复制代码

1.   //选择指定队列同步发送消息
2.   SendResult send(final Message msg, final MessageQueue mq) throws MQClientException,
3.       RemotingException, MQBrokerException, InterruptedException;

复制代码

1.   //选择指定队列异步发送消息
2.   void send(final Message msg, final MessageQueue mq, final SendCallback sendCallback)
3.       throws MQClientException, RemotingException, InterruptedException;

复制代码

1.   //选择指定队列单项发送消息
2.   void sendOneway(final Message msg, final MessageQueue mq) throws MQClientException,
3.       RemotingException, InterruptedException;

复制代码

1.   //批量发送消息
2.   SendResult send(final Collection<Message> msgs) throws MQClientException, RemotingException, MQBrokerException,InterruptedException;

复制代码

2）属性介绍

1. producerGroup：生产者所属组
2. createTopicKey：默认Topic
3. defaultTopicQueueNums：默认主题在每一个Broker队列数量
4. sendMsgTimeout：发送消息默认超时时间，默认3s
5. compressMsgBodyOverHowmuch：消息体超过该值则启用压缩，默认4k
6. retryTimesWhenSendFailed：同步方式发送消息重试次数，默认为2，总共执行3次
7. retryTimesWhenSendAsyncFailed：异步方法发送消息重试次数，默认为2
8. retryAnotherBrokerWhenNotStoreOK：消息重试时选择另外一个Broker时，是否不等待存储结果就返回，默认为false
9. maxMessageSize：允许发送的最大消息长度，默认为4M

复制代码

启动流程

代码：DefaultMQProducerImpl#start

1. //检查生产者组是否满足要求
2. this.checkConfig();
3. //更改当前instanceName为进程ID
4. if (!this.defaultMQProducer.getProducerGroup().equals(MixAll.CLIENT_INNER_PRODUCER_GROUP)) {
5.     this.defaultMQProducer.changeInstanceNameToPID();
6. }
7. //获得MQ客户端实例
8. this.mQClientFactory = MQClientManager.getInstance().getAndCreateMQClientInstance(this.defaultMQProducer, rpcHook);

复制代码

整个JVM中只存在一个MQClientManager实例，维护一个MQClientInstance缓存表
ConcurrentMap factoryTable = new ConcurrentHashMap();
同一个clientId只会创建一个MQClientInstance。
MQClientInstance封装了RocketMQ网络处理API，是消息生产者和消息消费者与NameServer、Broker打交道的网络通道

代码：MQClientManager#getAndCreateMQClientInstance

1. public MQClientInstance getAndCreateMQClientInstance(final ClientConfig clientConfig,
2.                                                      RPCHook rpcHook) {
3.     //构建客户端ID
4.     String clientId = clientConfig.buildMQClientId();
5.     //根据客户端ID或者客户端实例
6.     MQClientInstance instance = this.factoryTable.get(clientId);
7.     //实例如果为空就创建新的实例,并添加到实例表中
8.     if (null == instance) {
9.         instance =
10.             new MQClientInstance(clientConfig.cloneClientConfig(),
11.                 this.factoryIndexGenerator.getAndIncrement(), clientId, rpcHook);
12.         MQClientInstance prev = this.factoryTable.putIfAbsent(clientId, instance);
13.         if (prev != null) {
14.             instance = prev;
15.             log.warn("Returned Previous MQClientInstance for clientId:[{}]", clientId);
16.         } else {
17.             log.info("Created new MQClientInstance for clientId:[{}]", clientId);
18.         }
19.     }
21.     return instance;
22. }

复制代码

代码：DefaultMQProducerImpl#start

1. //注册当前生产者到到MQClientInstance管理中,方便后续调用网路请求
2. boolean registerOK = mQClientFactory.registerProducer(this.defaultMQProducer.getProducerGroup(), this);
3. if (!registerOK) {
4.     this.serviceState = ServiceState.CREATE_JUST;
5.     throw new MQClientException("The producer group[" + this.defaultMQProducer.getProducerGroup()
6.         + "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL),
7.         null);
8. }
9. //启动生产者
10. if (startFactory) {
11.     mQClientFactory.start();
12. }

复制代码

消息发送

代码：DefaultMQProducerImpl#send(Message msg)

1. //发送消息
2. public SendResult send(Message msg) {
3.     return send(msg, this.defaultMQProducer.getSendMsgTimeout());
4. }

复制代码

代码：DefaultMQProducerImpl#send(Message msg,long timeout)

1. //发送消息,默认超时时间为3s
2. public SendResult send(Message msg,long timeout){
3.     return this.sendDefaultImpl(msg, CommunicationMode.SYNC, null, timeout);
4. }

复制代码

代码：DefaultMQProducerImpl#sendDefaultImpl

1. //校验消息
2. Validators.checkMessage(msg, this.defaultMQProducer);

复制代码

1）验证消息

代码：Validators#checkMessage

1. public static void checkMessage(Message msg, DefaultMQProducer defaultMQProducer)
2.     throws MQClientException {
3.     //判断是否为空
4.     if (null == msg) {
5.         throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL, "the message is null");
6.     }
7.     // 校验主题
8.     Validators.checkTopic(msg.getTopic());
9.                
10.     // 校验消息体
11.     if (null == msg.getBody()) {
12.         throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL, "the message body is null");
13.     }
15.     if (0 == msg.getBody().length) {
16.         throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL, "the message body length is zero");
17.     }
19.     if (msg.getBody().length > defaultMQProducer.getMaxMessageSize()) {
20.         throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL,
21.             "the message body size over max value, MAX: " + defaultMQProducer.getMaxMessageSize());
22.     }
23. }

复制代码

2）查找路由

代码：DefaultMQProducerImpl#tryToFindTopicPublishInfo

1. private TopicPublishInfo tryToFindTopicPublishInfo(final String topic) {
2.     //从缓存中获得主题的路由信息
3.     TopicPublishInfo topicPublishInfo = this.topicPublishInfoTable.get(topic);
4.     //路由信息为空,则从NameServer获取路由
5.     if (null == topicPublishInfo || !topicPublishInfo.ok()) {
6.         this.topicPublishInfoTable.putIfAbsent(topic, new TopicPublishInfo());
7.         this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic);
8.         topicPublishInfo = this.topicPublishInfoTable.get(topic);
9.     }
11.     if (topicPublishInfo.isHaveTopicRouterInfo() || topicPublishInfo.ok()) {
12.         return topicPublishInfo;
13.     } else {
14.         //如果未找到当前主题的路由信息,则用默认主题继续查找
15.         this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic, true, this.defaultMQProducer);
16.         topicPublishInfo = this.topicPublishInfoTable.get(topic);
17.         return topicPublishInfo;
18.     }
19. }

复制代码

代码：TopicPublishInfo

1. public class TopicPublishInfo {
2.     private boolean orderTopic = false;        //是否是顺序消息
3.     private boolean haveTopicRouterInfo = false;
4.     private List<MessageQueue> messageQueueList = new ArrayList<MessageQueue>();        //该主题消息队列
5.     private volatile ThreadLocalIndex sendWhichQueue = new ThreadLocalIndex();//每选择一次消息队列,该值+1
6.     private TopicRouteData topicRouteData;//关联Topic路由元信息
7. }

复制代码

代码：MQClientInstance#updateTopicRouteInfoFromNameServer

1. TopicRouteData topicRouteData;
2. //使用默认主题从NameServer获取路由信息
3. if (isDefault && defaultMQProducer != null) {
4.     topicRouteData = this.mQClientAPIImpl.getDefaultTopicRouteInfoFromNameServer(defaultMQProducer.getCreateTopicKey(),
5.         1000 * 3);
6.     if (topicRouteData != null) {
7.         for (QueueData data : topicRouteData.getQueueDatas()) {
8.             int queueNums = Math.min(defaultMQProducer.getDefaultTopicQueueNums(), data.getReadQueueNums());
9.             data.setReadQueueNums(queueNums);
10.             data.setWriteQueueNums(queueNums);
11.         }
12.     }
13. } else {
14.     //使用指定主题从NameServer获取路由信息
15.     topicRouteData = this.mQClientAPIImpl.getTopicRouteInfoFromNameServer(topic, 1000 * 3);
16. }

复制代码

代码：MQClientInstance#updateTopicRouteInfoFromNameServer

1. //判断路由是否需要更改
2. TopicRouteData old = this.topicRouteTable.get(topic);
3. boolean changed = topicRouteDataIsChange(old, topicRouteData);
4. if (!changed) {
5.     changed = this.isNeedUpdateTopicRouteInfo(topic);
6. } else {
7.     log.info("the topic[{}] route info changed, old[{}] ,new[{}]", topic, old, topicRouteData);
8. }

复制代码

代码：MQClientInstance#updateTopicRouteInfoFromNameServer

1. if (changed) {
2.     //将topicRouteData转换为发布队列
3.     TopicPublishInfo publishInfo = topicRouteData2TopicPublishInfo(topic, topicRouteData);
4.     publishInfo.setHaveTopicRouterInfo(true);
5.     //遍历生产
6.     Iterator<Entry<String, MQProducerInner>> it = this.producerTable.entrySet().iterator();
7.     while (it.hasNext()) {
8.         Entry<String, MQProducerInner> entry = it.next();
9.         MQProducerInner impl = entry.getValue();
10.         if (impl != null) {
11.             //生产者不为空时,更新publishInfo信息
12.             impl.updateTopicPublishInfo(topic, publishInfo);
13.         }
14.     }
15. }

复制代码

代码：MQClientInstance#topicRouteData2TopicPublishInfo

1. public static TopicPublishInfo topicRouteData2TopicPublishInfo(final String topic, final TopicRouteData route) {
2.             //创建TopicPublishInfo对象
3.         TopicPublishInfo info = new TopicPublishInfo();
4.             //关联topicRoute
5.         info.setTopicRouteData(route);
6.             //顺序消息,更新TopicPublishInfo
7.         if (route.getOrderTopicConf() != null && route.getOrderTopicConf().length() > 0) {
8.             String[] brokers = route.getOrderTopicConf().split(";");
9.             for (String broker : brokers) {
10.                 String[] item = broker.split(":");
11.                 int nums = Integer.parseInt(item[1]);
12.                 for (int i = 0; i < nums; i++) {
13.                     MessageQueue mq = new MessageQueue(topic, item[0], i);
14.                     info.getMessageQueueList().add(mq);
15.                 }
16.             }
18.             info.setOrderTopic(true);
19.         } else {
20.             //非顺序消息更新TopicPublishInfo
21.             List<QueueData> qds = route.getQueueDatas();
22.             Collections.sort(qds);
23.             //遍历topic队列信息
24.             for (QueueData qd : qds) {
25.                 //是否是写队列
26.                 if (PermName.isWriteable(qd.getPerm())) {
27.                     BrokerData brokerData = null;
28.                     //遍历写队列Broker
29.                     for (BrokerData bd : route.getBrokerDatas()) {
30.                         //根据名称获得读队列对应的Broker
31.                         if (bd.getBrokerName().equals(qd.getBrokerName())) {
32.                         brokerData = bd;
33.                         break;
34.                     }
35.                 }
37.                 if (null == brokerData) {
38.                     continue;
39.                 }
41.                 if (!brokerData.getBrokerAddrs().containsKey(MixAll.MASTER_ID)) {
42.                     continue;
43.                 }
44.                                 //封装TopicPublishInfo写队列
45.                 for (int i = 0; i < qd.getWriteQueueNums(); i++) {
46.                     MessageQueue mq = new MessageQueue(topic, qd.getBrokerName(), i);
47.                     info.getMessageQueueList().add(mq);
48.                 }
49.             }
50.         }
52.         info.setOrderTopic(false);
53.     }
54.         //返回TopicPublishInfo对象
55.     return info;
56. }

复制代码

3）选择队列

• 默认不启用Broker故障延迟机制
  

代码：TopicPublishInfo#selectOneMessageQueue(lastBrokerName)

1. public MessageQueue selectOneMessageQueue(final String lastBrokerName) {
2.     //第一次选择队列
3.     if (lastBrokerName == null) {
4.         return selectOneMessageQueue();
5.     } else {
6.         //sendWhichQueue
7.         int index = this.sendWhichQueue.getAndIncrement();
8.         //遍历消息队列集合
9.         for (int i = 0; i < this.messageQueueList.size(); i++) {
10.             //sendWhichQueue自增后取模
11.             int pos = Math.abs(index++) % this.messageQueueList.size();
12.             if (pos < 0)
13.                 pos = 0;
14.             //规避上次Broker队列
15.             MessageQueue mq = this.messageQueueList.get(pos);
16.             if (!mq.getBrokerName().equals(lastBrokerName)) {
17.                 return mq;
18.             }
19.         }
20.         //如果以上情况都不满足,返回sendWhichQueue取模后的队列
21.         return selectOneMessageQueue();
22.     }
23. }

复制代码

代码：TopicPublishInfo#selectOneMessageQueue()

1. //第一次选择队列
2. public MessageQueue selectOneMessageQueue() {
3.     //sendWhichQueue自增
4.     int index = this.sendWhichQueue.getAndIncrement();
5.     //对队列大小取模
6.     int pos = Math.abs(index) % this.messageQueueList.size();
7.     if (pos < 0)
8.         pos = 0;
9.     //返回对应的队列
10.     return this.messageQueueList.get(pos);
11. }

复制代码

• 启用Broker故障延迟机制
  

1. public MessageQueue selectOneMessageQueue(final TopicPublishInfo tpInfo, final String lastBrokerName) {
2.     //Broker故障延迟机制
3.     if (this.sendLatencyFaultEnable) {
4.         try {
5.             //对sendWhichQueue自增
6.             int index = tpInfo.getSendWhichQueue().getAndIncrement();
7.             //对消息队列轮询获取一个队列
8.             for (int i = 0; i < tpInfo.getMessageQueueList().size(); i++) {
9.                 int pos = Math.abs(index++) % tpInfo.getMessageQueueList().size();
10.                 if (pos < 0)
11.                     pos = 0;
12.                 MessageQueue mq = tpInfo.getMessageQueueList().get(pos);
13.                 //验证该队列是否可用
14.                 if (latencyFaultTolerance.isAvailable(mq.getBrokerName())) {
15.                     //可用
16.                     if (null == lastBrokerName || mq.getBrokerName().equals(lastBrokerName))
17.                         return mq;
18.                 }
19.             }
20.                         //从规避的Broker中选择一个可用的Broker
21.             final String notBestBroker = latencyFaultTolerance.pickOneAtLeast();
22.             //获得Broker的写队列集合
23.             int writeQueueNums = tpInfo.getQueueIdByBroker(notBestBroker);
24.             if (writeQueueNums > 0) {
25.                 //获得一个队列,指定broker和队列ID并返回
26.                 final MessageQueue mq = tpInfo.selectOneMessageQueue();
27.                 if (notBestBroker != null) {
28.                     mq.setBrokerName(notBestBroker);
29.                     mq.setQueueId(tpInfo.getSendWhichQueue().getAndIncrement() % writeQueueNums);
30.                 }
31.                 return mq;
32.             } else {
33.                 latencyFaultTolerance.remove(notBestBroker);
34.             }
35.         } catch (Exception e) {
36.             log.error("Error occurred when selecting message queue", e);
37.         }
39.         return tpInfo.selectOneMessageQueue();
40.     }
42.     return tpInfo.selectOneMessageQueue(lastBrokerName);
43. }

复制代码

• 延迟机制接口规范
  

1. public interface LatencyFaultTolerance<T> {
2.     //更新失败条目
3.     void updateFaultItem(final T name, final long currentLatency, final long notAvailableDuration);
4.         //判断Broker是否可用
5.     boolean isAvailable(final T name);
6.         //移除Fault条目
7.     void remove(final T name);
8.         //尝试从规避的Broker中选择一个可用的Broker
9.     T pickOneAtLeast();
10. }

复制代码

• FaultItem：失败条目
  

1. class FaultItem implements Comparable<FaultItem> {
2.     //条目唯一键,这里为brokerName
3.     private final String name;
4.     //本次消息发送延迟
5.     private volatile long currentLatency;
6.     //故障规避开始时间
7.     private volatile long startTimestamp;
8. }

复制代码

• 消息失败策略
  

1. public class MQFaultStrategy {
2.    //根据currentLatency本地消息发送延迟,从latencyMax尾部向前找到第一个比currentLatency小的索引,如果没有找到,返回0
3.         private long[] latencyMax = {50L, 100L, 550L, 1000L, 2000L, 3000L, 15000L};
4.     //根据这个索引从notAvailableDuration取出对应的时间,在该时长内,Broker设置为不可用
5.         private long[] notAvailableDuration = {0L, 0L, 30000L, 60000L, 120000L, 180000L, 600000L};
6. }

复制代码

原理分析
代码：DefaultMQProducerImpl#sendDefaultImpl

1. sendResult = this.sendKernelImpl(msg,
2.                                  mq,
3.                                  communicationMode,
4.                                  sendCallback,
5.                                  topicPublishInfo,
6.                                  timeout - costTime);
7. endTimestamp = System.currentTimeMillis();
8. this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false);

复制代码

如果上述发送过程出现异常，则调用DefaultMQProducerImpl#updateFaultItem

1. public void updateFaultItem(final String brokerName, final long currentLatency, boolean isolation) {
2.     //参数一：broker名称
3.     //参数二:本次消息发送延迟时间
4.     //参数三:是否隔离
5.     this.mqFaultStrategy.updateFaultItem(brokerName, currentLatency, isolation);
6. }

复制代码

代码：MQFaultStrategy#updateFaultItem

1. public void updateFaultItem(final String brokerName, final long currentLatency, boolean isolation) {
2.     if (this.sendLatencyFaultEnable) {
3.         //计算broker规避的时长
4.         long duration = computeNotAvailableDuration(isolation ? 30000 : currentLatency);
5.         //更新该FaultItem规避时长
6.         this.latencyFaultTolerance.updateFaultItem(brokerName, currentLatency, duration);
7.     }
8. }

复制代码

代码：MQFaultStrategy#computeNotAvailableDuration

1. private long computeNotAvailableDuration(final long currentLatency) {
2.     //遍历latencyMax
3.     for (int i = latencyMax.length - 1; i >= 0; i--) {
4.         //找到第一个比currentLatency的latencyMax值
5.         if (currentLatency >= latencyMax[i])
6.             return this.notAvailableDuration[i];
7.     }
8.     //没有找到则返回0
9.     return 0;
10. }

复制代码

代码：LatencyFaultToleranceImpl#updateFaultItem

1. public void updateFaultItem(final String name, final long currentLatency, final long notAvailableDuration) {
2.     //获得原FaultItem
3.     FaultItem old = this.faultItemTable.get(name);
4.     //为空新建faultItem对象,设置规避时长和开始时间
5.     if (null == old) {
6.         final FaultItem faultItem = new FaultItem(name);
7.         faultItem.setCurrentLatency(currentLatency);
8.         faultItem.setStartTimestamp(System.currentTimeMillis() + notAvailableDuration);
10.         old = this.faultItemTable.putIfAbsent(name, faultItem);
11.         if (old != null) {
12.             old.setCurrentLatency(currentLatency);
13.             old.setStartTimestamp(System.currentTimeMillis() + notAvailableDuration);
14.         }
15.     } else {
16.         //更新规避时长和开始时间
17.         old.setCurrentLatency(currentLatency);
18.         old.setStartTimestamp(System.currentTimeMillis() + notAvailableDuration);
19.     }
20. }

复制代码

4）发送消息

消息发送API核心入口DefaultMQProducerImpl#sendKernelImpl

1. private SendResult sendKernelImpl(
2.     final Message msg,        //待发送消息
3.     final MessageQueue mq,        //消息发送队列
4.     final CommunicationMode communicationMode,                //消息发送内模式
5.     final SendCallback sendCallback,        pp        //异步消息回调函数
6.     final TopicPublishInfo topicPublishInfo,        //主题路由信息
7.     final long timeout        //超时时间
8.     )

复制代码

代码：DefaultMQProducerImpl#sendKernelImpl

1. //获得broker网络地址信息
2. String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
3. if (null == brokerAddr) {
4.     //没有找到从NameServer更新broker网络地址信息
5.     tryToFindTopicPublishInfo(mq.getTopic());
6.     brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
7. }

复制代码

1. //为消息分类唯一ID
2. if (!(msg instanceof MessageBatch)) {
3.     MessageClientIDSetter.setUniqID(msg);
4. }
6. boolean topicWithNamespace = false;
7. if (null != this.mQClientFactory.getClientConfig().getNamespace()) {
8.     msg.setInstanceId(this.mQClientFactory.getClientConfig().getNamespace());
9.     topicWithNamespace = true;
10. }
11. //消息大小超过4K,启用消息压缩
12. int sysFlag = 0;
13. boolean msgBodyCompressed = false;
14. if (this.tryToCompressMessage(msg)) {
15.     sysFlag |= MessageSysFlag.COMPRESSED_FLAG;
16.     msgBodyCompressed = true;
17. }
18. //如果是事务消息,设置消息标记MessageSysFlag.TRANSACTION_PREPARED_TYPE
19. final String tranMsg = msg.getProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED);
20. if (tranMsg != null && Boolean.parseBoolean(tranMsg)) {
21.     sysFlag |= MessageSysFlag.TRANSACTION_PREPARED_TYPE;
22. }

复制代码

1. //如果注册了消息发送钩子函数,在执行消息发送前的增强逻辑
2. if (this.hasSendMessageHook()) {
3.     context = new SendMessageContext();
4.     context.setProducer(this);
5.     context.setProducerGroup(this.defaultMQProducer.getProducerGroup());
6.     context.setCommunicationMode(communicationMode);
7.     context.setBornHost(this.defaultMQProducer.getClientIP());
8.     context.setBrokerAddr(brokerAddr);
9.     context.setMessage(msg);
10.     context.setMq(mq);
11.     context.setNamespace(this.defaultMQProducer.getNamespace());
12.     String isTrans = msg.getProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED);
13.     if (isTrans != null && isTrans.equals("true")) {
14.         context.setMsgType(MessageType.Trans_Msg_Half);
15.     }
17.     if (msg.getProperty("__STARTDELIVERTIME") != null || msg.getProperty(MessageConst.PROPERTY_DELAY_TIME_LEVEL) != null) {
18.         context.setMsgType(MessageType.Delay_Msg);
19.     }
20.     this.executeSendMessageHookBefore(context);
21. }

复制代码

代码：SendMessageHook

1. public interface SendMessageHook {
2.     String hookName();
4.     void sendMessageBefore(final SendMessageContext context);
6.     void sendMessageAfter(final SendMessageContext context);
7. }

复制代码

代码：DefaultMQProducerImpl#sendKernelImpl

1. //构建消息发送请求包
2. SendMessageRequestHeader requestHeader = new SendMessageRequestHeader();
3. //生产者组
4. requestHeader.setProducerGroup(this.defaultMQProducer.getProducerGroup());
5. //主题
6. requestHeader.setTopic(msg.getTopic());
7. //默认创建主题Key
8. requestHeader.setDefaultTopic(this.defaultMQProducer.getCreateTopicKey());
9. //该主题在单个Broker默认队列树
10. requestHeader.setDefaultTopicQueueNums(this.defaultMQProducer.getDefaultTopicQueueNums());
11. //队列ID
12. requestHeader.setQueueId(mq.getQueueId());
13. //消息系统标记
14. requestHeader.setSysFlag(sysFlag);
15. //消息发送时间
16. requestHeader.setBornTimestamp(System.currentTimeMillis());
17. //消息标记
18. requestHeader.setFlag(msg.getFlag());
19. //消息扩展信息
20. requestHeader.setProperties(MessageDecoder.messageProperties2String(msg.getProperties()));
21. //消息重试次数
22. requestHeader.setReconsumeTimes(0);
23. requestHeader.setUnitMode(this.isUnitMode());
24. //是否是批量消息等
25. requestHeader.setBatch(msg instanceof MessageBatch);
26. if (requestHeader.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
27.     String reconsumeTimes = MessageAccessor.getReconsumeTime(msg);
28.     if (reconsumeTimes != null) {
29.         requestHeader.setReconsumeTimes(Integer.valueOf(reconsumeTimes));
30.         MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_RECONSUME_TIME);
31.     }
33.     String maxReconsumeTimes = MessageAccessor.getMaxReconsumeTimes(msg);
34.     if (maxReconsumeTimes != null) {
35.         requestHeader.setMaxReconsumeTimes(Integer.valueOf(maxReconsumeTimes));
36.         MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_MAX_RECONSUME_TIMES);
37.     }
38. }

复制代码

1. case ASYNC:                //异步发送
2.     Message tmpMessage = msg;
3.     boolean messageCloned = false;
4.     if (msgBodyCompressed) {
5.         //If msg body was compressed, msgbody should be reset using prevBody.
6.         //Clone new message using commpressed message body and recover origin massage.
7.         //Fix bug:https://github.com/apache/rocketmq-externals/issues/66
8.         tmpMessage = MessageAccessor.cloneMessage(msg);
9.         messageCloned = true;
10.         msg.setBody(prevBody);
11.     }
13.     if (topicWithNamespace) {
14.         if (!messageCloned) {
15.             tmpMessage = MessageAccessor.cloneMessage(msg);
16.             messageCloned = true;
17.         }
18.         msg.setTopic(NamespaceUtil.withoutNamespace(msg.getTopic(),
19.                                                     this.defaultMQProducer.getNamespace()));
20.     }
22.                 long costTimeAsync = System.currentTimeMillis() - beginStartTime;
23.                 if (timeout < costTimeAsync) {
24.                     throw new RemotingTooMuchRequestException("sendKernelImpl call timeout");
25.                 }
26.                 sendResult = this.mQClientFactory.getMQClientAPIImpl().sendMessage(
27.                                 brokerAddr,
28.                                 mq.getBrokerName(),
29.                                 tmpMessage,
30.                                 requestHeader,
31.                                 timeout - costTimeAsync,
32.                                 communicationMode,
33.                                 sendCallback,
34.                                 topicPublishInfo,
35.                                 this.mQClientFactory,
36.                                 this.defaultMQProducer.getRetryTimesWhenSendAsyncFailed(),
37.                                 context,
38.                                 this);
39.             break;
40. case ONEWAY:
41. case SYNC:                //同步发送
42.     long costTimeSync = System.currentTimeMillis() - beginStartTime;
43.         if (timeout < costTimeSync) {
44.             throw new RemotingTooMuchRequestException("sendKernelImpl call timeout");
45.         }
46.         sendResult = this.mQClientFactory.getMQClientAPIImpl().sendMessage(
47.             brokerAddr,
48.             mq.getBrokerName(),
49.             msg,
50.             requestHeader,
51.             timeout - costTimeSync,
52.             communicationMode,
53.             context,
54.             this);
55.         break;
56.     default:
57.         assert false;
58.         break;
59. }

复制代码

1. //如果注册了钩子函数,则发送完毕后执行钩子函数
2. if (this.hasSendMessageHook()) {
3.     context.setSendResult(sendResult);
4.     this.executeSendMessageHookAfter(context);
5. }

复制代码

批量消息发送

批量消息发送是将同一个主题的多条消息一起打包发送到消息服务端，减少网络调用次数，提高网络传输效率。当然，并不是在同一批次中发送的消息数量越多越好，其判断依据是单条消息的长度，如果单条消息内容比较长，则打包多条消息发送会影响其他线程发送消息的响应时间，并且单批次消息总长度不能超过DefaultMQProducer#maxMessageSize。
批量消息发送要解决的问题是如何将这些消息编码以便服务端能够正确解码出每条消息的消息内容。
代码：DefaultMQProducer#send

1. public SendResult send(Collection<Message> msgs)
2.     throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
3.     //压缩消息集合成一条消息,然后发送出去
4.     return this.defaultMQProducerImpl.send(batch(msgs));
5. }

复制代码

代码：DefaultMQProducer#batch

1. private MessageBatch batch(Collection<Message> msgs) throws MQClientException {
2.     MessageBatch msgBatch;
3.     try {
4.         //将集合消息封装到MessageBatch
5.         msgBatch = MessageBatch.generateFromList(msgs);
6.         //遍历消息集合,检查消息合法性,设置消息ID,设置Topic
7.         for (Message message : msgBatch) {
8.             Validators.checkMessage(message, this);
9.             MessageClientIDSetter.setUniqID(message);
10.             message.setTopic(withNamespace(message.getTopic()));
11.         }
12.         //压缩消息,设置消息body
13.         msgBatch.setBody(msgBatch.encode());
14.     } catch (Exception e) {
15.         throw new MQClientException("Failed to initiate the MessageBatch", e);
16.     }
17.     //设置msgBatch的topic
18.     msgBatch.setTopic(withNamespace(msgBatch.getTopic()));
19.     return msgBatch;
20. }

复制代码

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！