java

关注公众号 jb51net

关闭
首页 > 软件编程 > java > RocketMQ Namesrv架构

RocketMQ Namesrv架构工作原理详解

作者:奔跑的毛球

这篇文章主要为大家介绍了RocketMQ Namesrv架构工作原理详解,有需要的朋友可以借鉴参考下,希望能够有所帮助,祝大家多多进步,早日升职加薪

1 概念

Namesrv的作用是保存元数据提高Broker的可用性

Namesrv的主要功能是临时存储管理Topic路由信息,各个Namesrv节点之间是不通信无状态的,互相不知道对方的存在。

当Broker,生产者,消费者启动的时候,会轮询全部的Namesrv节点,获取路由信息。

2 核心数据结构和API

2.1 Namesrv的核心数据结构

Namesrv中保存的信息是Topic的路由信息,Topic的路由决定了Topic的信息发送给哪些Broker,或者从哪些Broker获取消息。

路由数据结构的实现代码都在org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager中

public class RouteInfoManager {
    private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);
    //Broker存活的时间周期,默认120秒
    private final static long DEFAULT_BROKER_CHANNEL_EXPIRED_TIME = 1000 * 60 * 2;
    private final ReadWriteLock lock = new ReentrantReadWriteLock();
    //保存Topic和队列的路由信息
    private final Map<String/* topic */, Map<String, QueueData>> topicQueueTable;
    //Broker名字和Broker信息的对应信息
    private final Map<String/* brokerName */, BrokerData> brokerAddrTable;
    //集群和Broker的对应关系
    private final Map<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable;
    //在线的Broker地址和Broker信息的对应关系
    private final Map<BrokerAddrInfo/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;
    //过滤服务器消息
    private final Map<BrokerAddrInfo/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;
    private final Map<String/* topic */, Map<String/*brokerName*/, TopicQueueMappingInfo>> topicQueueMappingInfoTable;
    private final BatchUnRegisterService unRegisterService;
    private final NamesrvController namesrvController;
    private final NamesrvConfig namesrvConfig;

2.2 Namesrv的API

Namesrv的的API在org.apache.rocketmq.namesrv.processor.DefaultRequestProcessor中,根据方法名很容易判断出来方法的作用。

switch (request.getCode()) {
    case RequestCode.PUT_KV_CONFIG:
        return this.putKVConfig(ctx, request);
    case RequestCode.GET_KV_CONFIG:
        return this.getKVConfig(ctx, request);
    case RequestCode.DELETE_KV_CONFIG:
        return this.deleteKVConfig(ctx, request);
    case RequestCode.QUERY_DATA_VERSION:
        return this.queryBrokerTopicConfig(ctx, request);
    case RequestCode.REGISTER_BROKER:
        //Broker注册自身信息到Namesrv
        return this.registerBroker(ctx, request);
    case RequestCode.UNREGISTER_BROKER:
        //Broker取消注册自身信息到Namesrv
        return this.unregisterBroker(ctx, request);
    case RequestCode.BROKER_HEARTBEAT:
        return this.brokerHeartbeat(ctx, request);
    case RequestCode.GET_BROKER_MEMBER_GROUP:
        return this.getBrokerMemberGroup(ctx, request);
    case RequestCode.GET_BROKER_CLUSTER_INFO:
        return this.getBrokerClusterInfo(ctx, request);
    case RequestCode.WIPE_WRITE_PERM_OF_BROKER:
        return this.wipeWritePermOfBroker(ctx, request);
    case RequestCode.ADD_WRITE_PERM_OF_BROKER:
        return this.addWritePermOfBroker(ctx, request);
    case RequestCode.GET_ALL_TOPIC_LIST_FROM_NAMESERVER:
        return this.getAllTopicListFromNameserver(ctx, request);
    case RequestCode.DELETE_TOPIC_IN_NAMESRV:
        return this.deleteTopicInNamesrv(ctx, request);
    case RequestCode.REGISTER_TOPIC_IN_NAMESRV:
        return this.registerTopicToNamesrv(ctx, request);
    case RequestCode.GET_KVLIST_BY_NAMESPACE:
        return this.getKVListByNamespace(ctx, request);
    case RequestCode.GET_TOPICS_BY_CLUSTER:
        return this.getTopicsByCluster(ctx, request);
    case RequestCode.GET_SYSTEM_TOPIC_LIST_FROM_NS:
        return this.getSystemTopicListFromNs(ctx, request);
    case RequestCode.GET_UNIT_TOPIC_LIST:
        return this.getUnitTopicList(ctx, request);
    case RequestCode.GET_HAS_UNIT_SUB_TOPIC_LIST:
        return this.getHasUnitSubTopicList(ctx, request);
    case RequestCode.GET_HAS_UNIT_SUB_UNUNIT_TOPIC_LIST:
        return this.getHasUnitSubUnUnitTopicList(ctx, request);
    case RequestCode.UPDATE_NAMESRV_CONFIG:
        return this.updateConfig(ctx, request);
    case RequestCode.GET_NAMESRV_CONFIG:
        return this.getConfig(ctx, request);
    case RequestCode.GET_CLIENT_CONFIG:
        return this.getClientConfigs(ctx, request);
    default:
        String error = " request type " + request.getCode() + " not supported";
        return RemotingCommand.createResponseCommand(RemotingSysResponseCode.REQUEST_CODE_NOT_SUPPORTED, error);
}

3 Namesrv架构

下图是一个消息的常规流转过程,生产者,消费者,Broker通过与Namesrv交换信息来实现自己的功能。

3.1组件

Broker在启动的时候,将自己的元数据信息,上报给Namesrv,这部分信息也就是Topic路由。

这里的元数据包含Broker本身的元数据和该Broker中Topic的信息。

生产者只关注Topic路由,从namesrv获取到Topic路由后就可以知道这个Topic的消息存放到了哪些Broker中。

消费者也只关注Topic路由,从namesrv获取到获取到Topic路由之后,才能知道自己订阅的Topic的Broker地址,从而获取消息。

3.2 Namesrv四个功能模块

这是Namesrv最核心的模块,Topic路由决定,Topic的数据会保存在哪些Broker上。Broker启动的时候,会将自身的信息注册到Namesrv中,以供消费者和生产者获取。生产者和消费者与Namesrv之间会有心跳通信,从而获取最新的Broker信息。

这个模块是基于Netty的网络通信封装,担任各个组件之间的网络通信任务。

定时任务模块包括:定时扫描宕机的Broker,定时打印KV配置,定时扫描超时请求。

Namesrv维护了一个全局的KV配置魔窟啊,方便全局配置。

以上就是RocketMQ Namesrv架构工作原理详解的详细内容,更多关于RocketMQ Namesrv架构的资料请关注脚本之家其它相关文章!

您可能感兴趣的文章:
阅读全文