大家好，我是君哥。
今天来分享一个最近生产环境遇到的一个 RocketMQ 异常：
首先，我们回顾一下 RockemtMQ 的架构：
Broker 的主从节点都会注册到 Name Server 集群，Name Server 集群保存了 Broker 相关信息。RocketMQ client 会在本地维护一份 topic 和 Broker 地址的映射关系，放在 MQClientInstance#brokerAddrTable。
发送消息
RocketMQ client 在发送消息时，会根据 topic 首先从本地缓存（brokerAddrTable）获取 Broker，如果获取不到，就会到 Name Server 集群中获取。
//DefaultMQProducerImpl 类 private SendResult sendKernelImpl(final Message msg, final MessageQueue mq, final CommunicationMode communicationMode, final SendCallback sendCallback, final TopicPublishInfo topicPublishInfo, final long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException { String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName()); if (null == brokerAddr) { //从 Name Server 中获取 tryToFindTopicPublishInfo(mq.getTopic()); brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName()); } if (brokerAddr != null) { //省略处理逻辑 } throw new MQClientException("The broker[" + mq.getBrokerName() + "] not exist", null); }
从上面的代码可以看出，如果本地缓存和 Name Server 都没有保存 Broker 信息，则会抛出 Broker 不存在的异常。这种情况解决思路就是从 Broker 启动时是否注册成功来着手分析。
消息偏移量
获取偏移量
客户端获取消息偏移量（Consume Offset）的时候，也可能会抛出这个异常：
//RemoteBrokerOffsetStore 类 public long readOffset(final MessageQueue mq, final ReadOffsetType type) { if (mq != null) { switch (type) { case MEMORY_FIRST_THEN_STORE: case READ_FROM_MEMORY: { //省略实现逻辑 } case READ_FROM_STORE: { //省略 long brokerOffset = this.fetchConsumeOffsetFromBroker(mq); //省略 } default: break; } } return -1; }
从上面的代码中可以看到：获取偏移量的方式有 3 种：
- MEMORY_FIRST_THEN_STORE：先从内存中获取，如果获取不到，再从 Broker 请求；
- READ_FROM_MEMORY：直接从内存中获取；
- READ_FROM_STORE：直接从 Broker 请求。
从 Broker 请求的代码如下：
//RemoteBrokerOffsetStore 类 private long fetchConsumeOffsetFromBroker(MessageQueue mq) throws RemotingException, MQBrokerException, InterruptedException, MQClientException { FindBrokerResult findBrokerResult = this.mQClientFactory.findBrokerAddressInSubscribe(mq.getBrokerName(), MixAll.MASTER_ID, true); if (null == findBrokerResult) { this.mQClientFactory.updateTopicRouteInfoFromNameServer(mq.getTopic()); findBrokerResult = this.mQClientFactory.findBrokerAddressInSubscribe(mq.getBrokerName(), MixAll.MASTER_ID, false); } if (findBrokerResult != null) { //忽略处理逻辑 } else { throw new MQClientException("The broker[" + mq.getBrokerName() + "] not exist", null); } }
这段代码跟上一节发送消息时获取 Broker 地址的代码一样，首先从本地内存中获取，如果过去不到，就从 Name Server 中获取，如果取不到，就抛出 Broker 不存在的异常。
其他获取偏移量方法
除了上面的获取偏移量的方法外，还有 3 个获取偏移量的方法，在 MQAdminImpl 类：
- searchOffset：从 Broker 获取 Message-Queue 偏移量，跟上面方法类似；
- maxOffset：从 Broker 获取 MessageQ-ueue 最大偏移量；
- minOffset：从 Broker 获取 MessageQu-eue 最小偏移量。
这些方法的使用都在源码【rocketmq-tools】这个模块中。
获取最早消息的保存时间
还有一个跟偏移量相关的方法，获取最早的一条消息的保存时间，代码如下：
//RemoteBrokerOffsetStore 类 public long earliestMsgStoreTime(MessageQueue mq) throws MQClientException { String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName()); if (null == brokerAddr) { this.mQClientFactory.updateTopicRouteInfoFromNameServer(mq.getTopic()); brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName()); } //省略处理逻辑 throw new MQClientException("The broker[" + mq.getBrokerName() + "] not exist", null); }
这个方法是获取一个 MessageQueue 中最小偏移量消息的保存时间。
这些方法的使用都在源码【rocketmq-tools】这个模块中。
拉取消息
正常拉取消息
拉取消息的核心代码如下：
// PullAPIWrapper 类 public PullResult pullKernelImpl( //省略参数 ) throws MQClientException, RemotingException, MQBrokerException, InterruptedException { FindBrokerResult findBrokerResult = this.mQClientFactory.findBrokerAddressInSubscribe(mq.getBrokerName(), this.recalculatePullFromWhichNode(mq), false); if (null == findBrokerResult) { this.mQClientFactory.updateTopicRouteInfoFromNameServer(mq.getTopic()); findBrokerResult = this.mQClientFactory.findBrokerAddressInSubscribe(mq.getBrokerName(), this.recalculatePullFromWhichNode(mq), false); } if (findBrokerResult != null) { //省略处理逻辑 } throw new MQClientException("The broker[" + mq.getBrokerName() + "] not exist", null); }
从上面的代码可以看到，客户端从 Broker 拉取消息之前，首先会从本地缓存获取 Broker 地址，如果获取不到，就从 Name Server 获取 Broker 地址，如果获取失败，则抛出 Broker 不存在的异常。
偏移量不合法
如果拉取消息时返回偏移量不合法（OFFSET_ILLEGAL），这时就需要重新处理偏移量。客户端代码的调用关系如下：
这个发生在事务消息的场景，RocketMQ client 向 Broker 拉取消息时，如果 Broker 返回 PULL_OFFSET_MOVED，client 就会通过异步线程（定时 10s 后执行）通知 Broker 更新 offset 为 nextPullOffset（上次 pull 消息时 broker 返回）。代码如下：
public void updateConsumeOffsetToBroker(MessageQueue mq, long offset, boolean isOneway) throws RemotingException, MQBrokerException, InterruptedException, MQClientException { FindBrokerResult findBrokerResult = this.mQClientFactory.findBrokerAddressInSubscribe(mq.getBrokerName(), MixAll.MASTER_ID, true); if (null == findBrokerResult) { this.mQClientFactory.updateTopicRouteInfoFromNameServer(mq.getTopic()); findBrokerResult = this.mQClientFactory.findBrokerAddressInSubscribe(mq.getBrokerName(), MixAll.MASTER_ID, false); } if (findBrokerResult != null) { //省略业务代码 } else { throw new MQClientException("The broker[" + mq.getBrokerName() + "] not exist", null); } }
总结
今天根据之前遇到的一次生产环境的异常日志研究了出现【The broker[xxx] not exis】的 6 个场景，每个场景都类似，首先从本地缓存获取 Broker 地址，如果获取不到，就从 Name Server 获取。
出现这种情况一般有下面三个原因：
- Broker 挂了，客户端定时任务会判断到 Broker 离线，就会从本地缓存中移除（MQClientInstance#cleanOfflineBroker）；
- Broker 网络异常；
- Broker 发生了主备切换，客户端获取 Broker 地址时切换还没有完成。
这些场景其实也有定时任务刷新本地缓存，见下面代码：
//MQClientInstance 类 private void startScheduledTask() { this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { @Override public void run() { try { MQClientInstance.this.updateTopicRouteInfoFromNameServer(); } catch (Exception e) { log.error("ScheduledTask updateTopicRouteInfoFromNameServer exception", e); } } }, 10, this.clientConfig.getPollNameServerInterval(), TimeUnit.MILLISECONDS); this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { @Override public void run() { try { MQClientInstance.this.persistAllConsumerOffset(); } catch (Exception e) { log.error("ScheduledTask persistAllConsumerOffset exception", e); } } }, 1000 * 10, this.clientConfig.getPersistConsumerOffsetInterval(), TimeUnit.MILLISECONDS); }
·············· END ··············
