这六个场景下RocketMQ会找不到Broker

今天来分享一个最近生产环境遇到的一个 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);
}

当前文章：这六个场景下RocketMQ会找不到Broker
标题网址：http://www.shufengxianlan.com/qtweb/news13/476863.html

网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 创新互联

猜你还喜欢下面的内容

域名注册知识

同城分类信息

成都网站托管　　 　重庆托管服务器　　 　成都移动服务器托管　　 　广汉锦华橡塑　　 　成都圣月制冷　　 　中国电信西部信息中心机房　　 　成都外贸网站建设公司　　 　重庆app软件开发　　 　郫县封阳台　　 　微官网开发　　 　乐山服务器托管　　 　和正兴机械　　 　成都营销型网站建设　　 　网络推广公司　　 　成都印刷厂　　 　成都网络推广　　 　app制作　　 　注册域名　　 　免备案虚拟主机　　 　金融行业网站建设方案