RocketMQ之NameServer架构设计及启动关闭流程源码分析

更新时间：2021年11月15日 14:07:28 作者：又蠢又笨的懒羊羊程序猿

这篇文章主要为大家介绍了RocketMQ之NameServer架构设计及启动关闭流程源码分析详解，有需要的朋友可以借鉴参考下，希望能够有所帮助，祝大家多多进步

NameServer

1.架构设计

消息中间件的设计思路一般都是基于主题订阅与发布的机制，RocketMQ也不例外。RocketMQ中，消息生产者（Producer）发送某主题的消息到消息服务器，消息服务器对消息进行持久化存储，而消息消费者（Consumer）订阅所需要的主题，消息服务器根据订阅信息（路由信息）将消息推送至消息消费者（Push模式）或者消息消费者主动向消息服务器进行拉取（Pull模式），从而实现消息生产者与消息消费者之间解耦。

为了避免消息服务器单点故障而导致的系统瘫痪，消息服务器常常会集群分布，部署多台服务器共同处理消息并且承担消息的存储，消息生产者如何知道要将消息发送至哪台服务器和消息消费者如何知道要从哪台消息服务器进行消息的拉取等等问题，都要由NameServer来处理，其实NameServer充当的角色与Zookeeper十分相似。

在这里插入图片描述

Broker消息服务器启动时，需要向NameServer集群进行信息注册，消息生产者Producer发送消息之前主动向NameServer获取Broker服务器地址列表，然后根据负载均衡算法从列表中选出一台服务器进行消息的发送。NameServer与每台Broker保持长连接，并每隔30s对Broker存活状态进行检测，如果检测到Broker宕机并且长时间没有进行连接重试，则会将该Broker从路由注册表中删除，以此保证Broker集群的高可用，但是路由变化不会立马对生产者进行通知，需要Producer一段时间之后重新向NameServer进行获取并更新路由信息。这也是NameServer与Zookeeper的不同，NameServer这样的设计降低了整个NameServer实现的复杂度，整个NameServer代码实现不超过一千行，简单而高效！

以下是NameServer整个项目预览：

在这里插入图片描述

可以看到NameServer主要有以下几个作用：

配置信息管理

请求处理

路由信息管理

2.核心类与配置

NamesrvController

NameserController 是 NameServer 模块的核心控制类。

private final NamesrvConfig namesrvConfig;//主要指定 nameserver 的相关配置属性
private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("NSScheduledThread"));//NameServer定时任务执行线程池-->每隔10s扫描broker，对存活的Broker信息进行维护并且打印KVConfig
private final KVConfigManager kvConfigManager;//读取或变更NameServer的配置属性，加载 NamesrvConfig中配置到内存
private final RouteInfoManager routeInfoManager;//NameServer 数据的载体，记录 Broker、Topic 等信息。

private final NettyServerConfig nettyServerConfig;//与网络通讯相关的配置
private RemotingServer remotingServer;//网络通信服务
private ExecutorService remotingExecutor;//网络通信服务

NamesrvConfig

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

rocketmqHome：rocketmq主目录

kvConfigPath：NameServer存储KV配置属性的持久化路径

configStorePath：nameServer默认配置文件路径

orderMessageEnable：是否支持顺序消息

NettyServerConfig

private int listenPort = 8888;
private int serverWorkerThreads = 8;
private int serverCallbackExecutorThreads = 0;
private int serverSelectorThreads = 3;
private int serverOnewaySemaphoreValue = 256;
private int serverAsyncSemaphoreValue = 64;
private int serverChannelMaxIdleTimeSeconds = 120;

private int serverSocketSndBufSize = NettySystemConfig.socketSndbufSize;
private int serverSocketRcvBufSize = NettySystemConfig.socketRcvbufSize;
private boolean serverPooledByteBufAllocatorEnable = true;

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

RouteInfoManager

private final static long BROKER_CHANNEL_EXPIRED_TIME = 1000 * 60 * 2;
private final ReadWriteLock lock = new ReentrantReadWriteLock();
private final HashMap<String/* topic */, List<QueueData>> topicQueueTable;
private final HashMap<String/* brokerName */, BrokerData> brokerAddrTable;
private final HashMap<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable;
private final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;
private final HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;

BROKER_CHANNEL_EXPIRED_TIME：NameServer与Broker空闲连接时长，在2 minNameServer之内没有收到Broker的心跳包，则NameServer会关闭与该Broker的连接并删除Broker的路由信息。

lock：读写锁，用来保护以下用于存储关键信息的非线程安全容器HashMap。

topicQueueTable：用于存储主题与队列的映射关系，记录一个主题topic的队列分布在哪些Broker上。以下是QueueData属性值：

private String brokerName;			//broker名称
private int readQueueNums;			//读队列个数
private int writeQueueNums;			//写队列个数
private int perm;				   //操作权限	
private int topicSysFlag;			//同步复制还是异步复制的标识

brokerAddrTable：用于记录所有Broker信息。以下是BrokerData属性值：

private String cluster;				//当前Broker所属集群
private String brokerName;			//Broker名称
//BrokerId=0表示主节点，BrokerId>0表示从节点
//记录BrokerId与对应节点地址的映射信息
private HashMap<Long/* brokerId */, String/* broker address */> brokerAddrs;

clusterAddrTable：用于记录Broker集群信息

brokerLiveTable：用于记录活跃状态的Broker，NameServer每隔10s对所有Broker进行扫描，如果有Broker宕机，会将该Broker从该表中删去，以此维护可用的Broker列表信息。以下是BrokerLiveInfo的属性值：

private long lastUpdateTimestamp;		//上次发送心跳包的时间戳
private DataVersion dataVersion;		//记录数据版本信息
private Channel channel;
private String haServerAddr;			//Master节点地址

3.启动与关闭流程

NameServer启动时序图：

在这里插入图片描述

启动类：org.apache.rocketmq.namesrv.NamesrvStartup.java

3.1.步骤一

解析配置文件，填充NamesrvConfig、NettyServerConfig并创建NamesrvController：

启动类：

public static void main(String[] args) {
    main0(args);
}
public static NamesrvController main0(String[] args) {

    try {
        //创建NamesrvController的入口
        NamesrvController controller = createNamesrvController(args);
        start(controller);
        String tip = "The Name Server boot success. serializeType=" + RemotingCommand.getSerializeTypeConfigInThisServer();
        log.info(tip);
        System.out.printf("%s%n", tip);
        return controller;
    } catch (Throwable e) {
        e.printStackTrace();
        System.exit(-1);
    }
    return null;
}

NamesrvController#createNamesrvController：

public static NamesrvController createNamesrvController(String[] args) throws IOException, JoranException {
    //....    
    //创建namesrvConfig
    final NamesrvConfig namesrvConfig = new NamesrvConfig();
    //创建nettyServerConfig
    final NettyServerConfig nettyServerConfig = new NettyServerConfig();
    
    //设置默认端口9876
    nettyServerConfig.setListenPort(9876);
    //-c 指定属性配置文件的位置
    if (commandLine.hasOption('c')) {
        String file = commandLine.getOptionValue('c');
        if (file != null) {
            InputStream in = new BufferedInputStream(new FileInputStream(file));
            properties = new Properties();
            properties.load(in);
            MixAll.properties2Object(properties, namesrvConfig);
            MixAll.properties2Object(properties, nettyServerConfig);

            namesrvConfig.setConfigStorePath(file);

            System.out.printf("load config properties file OK, %s%n", file);
            in.close();
        }
    }
    //-p 属性名=属性值
    if (commandLine.hasOption('p')) {
        InternalLogger console = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_CONSOLE_NAME);
        MixAll.printObjectProperties(console, namesrvConfig);
        MixAll.printObjectProperties(console, nettyServerConfig);
        System.exit(0);
    }
    //将启动参数填充到namesrvConfig中
    MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), namesrvConfig);

    //如果未指定'ROCKETMQ_HOME'环境变量
    if (null == namesrvConfig.getRocketmqHome()) {
        System.out.printf("Please set the %s variable in your environment to match the location of the RocketMQ installation%n", MixAll.ROCKETMQ_HOME_ENV);
        System.exit(-2);
    }
	//....
    //打印配置信息日志
    MixAll.printObjectProperties(log, namesrvConfig);
    MixAll.printObjectProperties(log, nettyServerConfig);
    //根据namesrvConfig和nettyServerConfig创建NamesrvController
    final NamesrvController controller = new NamesrvController(namesrvConfig, nettyServerConfig);
    // 将配置存入controller.configuration以防止配置丢失
    controller.getConfiguration().registerConfig(properties);
    return controller;
}

3.2.步骤二

根据配置创建好NamesrvController之后，对其进行初始化：

//NamesrvStartup#start
public static NamesrvController start(final NamesrvController controller) throws Exception {

    //进行简单的检查
    if (null == controller) {
        throw new IllegalArgumentException("NamesrvController is null");
    }	
    //controller初始化
    boolean initResult = controller.initialize();
    if (!initResult) {
        controller.shutdown();
        System.exit(-3);
    }    
    //....
    controller.start();		//开启远程服务-this.remotingServer.start();

    return controller;
}
//NamesrvController#initialize
public boolean initialize() {

    	//加载配置管理器
        this.kvConfigManager.load();
		//创建Netty远程服务
        this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);
		//创建远程服务线程池
        this.remotingExecutor =
            Executors.newFixedThreadPool(nettyServerConfig.getServerWorkerThreads(), new ThreadFactoryImpl("RemotingExecutorThread_"));

        //注册线程池
        this.registerProcessor();
        //定时任务线程池--->每隔十秒扫描活跃状态异常的Broker信息
        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            /**
             * 对Not Active Broker 进行扫描
             */
            @Override
            public void run() {
                NamesrvController.this.routeInfoManager.scanNotActiveBroker();
            }
        }, 5, 10, TimeUnit.SECONDS);
        //定时任务线程池--->每隔十秒打印KVConfig信息
        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                NamesrvController.this.kvConfigManager.printAllPeriodically();
            }
        }, 1, 10, TimeUnit.MINUTES);
       //....
        return true;
    }

3.3.步骤三

在JVM进程关闭之前，先将线程池关闭，及时释放资源。

public static NamesrvController start(final NamesrvController controller) throws Exception {
	//....    
    //JVM进程关闭之前,将线程池关闭,资源释放
    Runtime.getRuntime().addShutdownHook/*注册JVM钩子函数*/(new ShutdownHookThread(log, new Callable<Void>() {
        @Override
        public Void call() throws Exception {
            controller.shutdown();
            return null;
        }
    }));	
    //....
}

以上仅供个人学习使用，如有不足请指正！

以上就是RocketMQ之NameServer架构设计及启动关闭流程源码分析的详细内容，更多关于RocketMQ之NameServer架构设计及启动关闭的资料请关注脚本之家其它相关文章！

您可能感兴趣的文章:

Springboot中使用Redisson+AOP+自定义注解实现访问限流与黑名单拦截
本文主要介绍了Springboot中使用Redisson+AOP+自定义注解实现访问限流与黑名单拦截,包含针对用户IP限流,整个接口的访问限流,以及对某个参数字段的限流,并且支持请求限流后处理回调,感兴趣的可以了解一下
2024-02-02
Spring的@CrossOrigin注解使用与CrossFilter对象自定义详解
这篇文章主要介绍了Spring的@CrossOrigin注解使用与CrossFilter对象自定义详解,跨域,指的是浏览器不能执行其他网站的脚本,它是由浏览器的同源策略造成的,是浏览器施加的安全限制,所谓同源是指,域名,协议,端口均相同,需要的朋友可以参考下
2023-12-12
详解如何在springcloud分布式系统中实现分布式锁
最近在看分布式锁的资料，本文就介绍一下利用springcloud结合redis实现分布式锁，文中通过示例代码介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
2021-11-11
Java大数据处理的核心技术MapReduce框架
MapReduce是一种分布式计算框架，适用于大规模的数据处理。它将大数据分成多个小数据块，通过Map和Reduce两个阶段对数据进行处理和分析。MapReduce框架具有可靠、高效、可扩展等特点，已经成为大数据处理的核心技术
2023-05-05
Java反射技术原理与用法实例分析
这篇文章主要介绍了Java反射技术原理与用法,结合实例形式分析了Java反射技术的基本概念、功能、原理、用法及操作注意事项,需要的朋友可以参考下
2020-04-04
Spring MVC请求转发与请求重定向的示例详解
转发指服务器接收请求后，从一个资源跳转到另一个资源中，请求转发是一次请求，不会改变浏览器的请求地址，这篇文章主要介绍了Spring MVC请求转发与请求重定向的相关知识,需要的朋友可以参考下
2023-09-09
intellij idea旗舰版解决学生无法注册问题详解
这篇文章主要介绍了intellij idea旗舰版解决学生无法注册问题详解，文中通过图文示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2020-07-07
Spring Boot整合邮件发送与注意事项
这篇文章主要给大家介绍了关于Spring Boot整合邮件发送与注意事项的相关资料，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2018-07-07
SpringBoot创建并简单使用的实现
这篇文章主要介绍了SpringBoot创建并简单使用的实现，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2020-10-10
alibaba seata服务端具体实现
seata是来处理分布式服务之间互相调用的事务问题，本文重点给大家介绍alibaba-seata实现方法，文中通过实例代码给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下
2022-02-02