西安网站建设的费用,网站制作评价指标,免费怎么制作公司网站,泰安做网络推广的公司NameServer功能简述
主要功能如下
服务注册与发现#xff1a;Nameserver扮演了RocketMQ集群中服务注册中心的角色。当RocketMQ中的Broker、Producer和Consumer启动时#xff0c;它们会向Nameserver注册自己的网络地址和角色信息。Nameserver维护着集群中所有活跃实例的信息…NameServer功能简述
主要功能如下
服务注册与发现Nameserver扮演了RocketMQ集群中服务注册中心的角色。当RocketMQ中的Broker、Producer和Consumer启动时它们会向Nameserver注册自己的网络地址和角色信息。Nameserver维护着集群中所有活跃实例的信息并提供查询接口供其他组件发现这些实例。这样Producer和Consumer可以根据Nameserver提供的信息找到Broker实现消息的发布和消费。Topic路由信息管理RocketMQ中的消息按照Topic进行分类。Nameserver维护着Topic到Broker之间的路由关系包括哪些Topic被哪些Broker负责。这样当Producer发送消息时它会根据Topic在Nameserver查询路由信息得知应该将消息发送到哪些Broker上。负载均衡Nameserver通过周期性地检查Broker的状态信息包括负载、存活状态等来维护一个Broker的列表供Producer和Consumer使用。这样做的目的是为了在消息生产和消费过程中实现负载均衡避免某个Broker过载影响整个系统性能。权限控制Nameserver支持对连接到集群的客户端进行权限控制。通过配置管理员可以限制哪些客户端可以连接到Broker增加系统的安全性。集群管理Nameserver允许管理员对Broker和Topic进行动态管理包括增加、删除Broker节点创建和删除Topic等操作。这使得RocketMQ集群能够根据实际业务需求进行扩展和调整。
综上所述RockerMQ Nameserver是RocketMQ集群中起着至关重要、不可或缺的作用。其服务注册与发现、Topic路由信息管理、负载均衡、权限控制和集群管理等功能Nameserver确保整个RocketMQ系统的稳定、高效和安全运行。
核心源码
main0
public static void main(String[] args) {// 本地启动需设置环境变量System.setProperty(MixAll.ROCKETMQ_HOME_PROPERTY, /Users/config/rocketmq);main0(args);controllerManagerMain();}1.parseCommandlineAndConfigFile解析命令行和配置文件初始化namesrvConfig、nettyServerConfig、nettyClientConfig等配置对象
public static void parseCommandlineAndConfigFile(String[] args) throws Exception {System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION));Options options ServerUtil.buildCommandlineOptions(new Options());CommandLine commandLine ServerUtil.parseCmdLine(mqnamesrv, args, buildCommandlineOptions(options), new DefaultParser());if (null commandLine) {System.exit(-1);return;}namesrvConfig new NamesrvConfig();nettyServerConfig new NettyServerConfig();nettyClientConfig new NettyClientConfig();nettyServerConfig.setListenPort(9876);if (commandLine.hasOption(c)) {String file commandLine.getOptionValue(c);if (file ! null) {InputStream in new BufferedInputStream(Files.newInputStream(Paths.get(file)));properties new Properties();properties.load(in);MixAll.properties2Object(properties, namesrvConfig);MixAll.properties2Object(properties, nettyServerConfig);MixAll.properties2Object(properties, nettyClientConfig);if (namesrvConfig.isEnableControllerInNamesrv()) {controllerConfig new ControllerConfig();MixAll.properties2Object(properties, controllerConfig);}namesrvConfig.setConfigStorePath(file);System.out.printf(load config properties file OK, %s%n, file);in.close();}}MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), namesrvConfig);if (commandLine.hasOption(p)) {MixAll.printObjectProperties(logConsole, namesrvConfig);MixAll.printObjectProperties(logConsole, nettyServerConfig);MixAll.printObjectProperties(logConsole, nettyClientConfig);if (namesrvConfig.isEnableControllerInNamesrv()) {MixAll.printObjectProperties(logConsole, controllerConfig);}System.exit(0);}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);}2.createAndStartNamesrvController创建并运行NamesrvController
public static NamesrvController start(final NamesrvController controller) throws Exception {if (null controller) {throw new IllegalArgumentException(NamesrvController is null);}boolean initResult controller.initialize();if (!initResult) {controller.shutdown();System.exit(-3);}Runtime.getRuntime().addShutdownHook(new ShutdownHookThread(log, (CallableVoid) () - {controller.shutdown();return null;}));controller.start();return controller;}一、controller.initialize()初始化
加载配置
private final HashMapString/* Namespace */, HashMapString/* Key */, String/* Value */ configTable new HashMap();
public void load() {String content null;try {content MixAll.file2String(this.namesrvController.getNamesrvConfig().getKvConfigPath());} catch (IOException e) {log.warn(Load KV config table exception, e);}if (content ! null) {KVConfigSerializeWrapper kvConfigSerializeWrapper KVConfigSerializeWrapper.fromJson(content, KVConfigSerializeWrapper.class);if (null ! kvConfigSerializeWrapper) {this.configTable.putAll(kvConfigSerializeWrapper.getConfigTable());log.info(load KV config table OK);}}}初始化网络组件
private void initiateNetworkComponents() {// remotingServer是一个NettyRemotingServer对象用于接收和处理来自其他服务的请求或响应// BrokerHouseKeepingService对象用于处理Broker的连接和断开事件this.remotingServer new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);// remotingClient是一个NettyRemotingClient对象它用于向其他服务发送请求或响应this.remotingClient new NettyRemotingClient(this.nettyClientConfig);}初始化线程执行器
private void initiateThreadExecutors() {this.defaultThreadPoolQueue new LinkedBlockingQueue(this.namesrvConfig.getDefaultThreadPoolQueueCapacity());// 用于处理默认的远程请求this.defaultExecutor new ThreadPoolExecutor(this.namesrvConfig.getDefaultThreadPoolNums(), this.namesrvConfig.getDefaultThreadPoolNums(), 1000 * 60, TimeUnit.MILLISECONDS, this.defaultThreadPoolQueue, new ThreadFactoryImpl(RemotingExecutorThread_)) {Overrideprotected T RunnableFutureT newTaskFor(final Runnable runnable, final T value) {return new FutureTaskExt(runnable, value);}};this.clientRequestThreadPoolQueue new LinkedBlockingQueue(this.namesrvConfig.getClientRequestThreadPoolQueueCapacity());// 处理客户端的路由信息请求this.clientRequestExecutor new ThreadPoolExecutor(this.namesrvConfig.getClientRequestThreadPoolNums(), this.namesrvConfig.getClientRequestThreadPoolNums(), 1000 * 60, TimeUnit.MILLISECONDS, this.clientRequestThreadPoolQueue, new ThreadFactoryImpl(ClientRequestExecutorThread_)) {Overrideprotected T RunnableFutureT newTaskFor(final Runnable runnable, final T value) {return new FutureTaskExt(runnable, value);}};}注册remoting server处理器
private void registerProcessor() {if (namesrvConfig.isClusterTest()) {// ClusterTestRequestProcessor是一个用于集群测试的处理器它会在请求前后添加一些环境信息比如产品环境名称、请求时间等this.remotingServer.registerDefaultProcessor(new ClusterTestRequestProcessor(this, namesrvConfig.getProductEnvName()), this.defaultExecutor);} else {// Support get route info only temporarily// 在 namesrvConfig.isClusterTest() false 时如果收到请求的 requestCode 等于 RequestCode.GET_ROUTEINFO_BY_TOPIC 则会使用ClientRequestProcessor来处理当收到其他请求时会使用DefaultRequestProcessor来处理。ClientRequestProcessor clientRequestProcessor new ClientRequestProcessor(this);this.remotingServer.registerProcessor(RequestCode.GET_ROUTEINFO_BY_TOPIC, clientRequestProcessor, this.clientRequestExecutor);// DefaultRequestProcessor是一个用于正常运行的处理器它会根据请求的类型调用不同的方法来处理比如注册Broker、获取路由信息、更新配置等。this.remotingServer.registerDefaultProcessor(new DefaultRequestProcessor(this), this.defaultExecutor);}}启动定时任务
private void startScheduleService() {// 首次延迟5毫秒执行后续执行间隔5秒执行分两种情况// 后续执行1.当执行任务时间小于间隔时间时延迟间隔时间-任务执行时间执行// 2.当任务执行时间大于间隔时间则任务结束立即执行下一次任务// 间隔扫描不活跃的Brokerthis.scanExecutorService.scheduleAtFixedRate(NamesrvController.this.routeInfoManager::scanNotActiveBroker,5, this.namesrvConfig.getScanNotActiveBrokerInterval(), TimeUnit.MILLISECONDS);//每隔10分钟打印所有KV配置this.scheduledExecutorService.scheduleAtFixedRate(NamesrvController.this.kvConfigManager::printAllPeriodically,1, 10, TimeUnit.MINUTES);// 每隔1分钟打印WaterMarkthis.scheduledExecutorService.scheduleAtFixedRate(() - {try {NamesrvController.this.printWaterMark();} catch (Throwable e) {LOGGER.error(printWaterMark error., e);}}, 10, 1, TimeUnit.SECONDS);}初始化ssl上下文配置remotingServer使用TLS协议进行安全通信注册RPC钩子
private void initiateRpcHooks() {// 注册RPC钩子 在remotingServer处理请求之前或之后执行一些自定义的逻辑this.remotingServer.registerRPCHook(new ZoneRouteRPCHook());}二、注册JVM钩子
controllerManager.shutdown()
// 注册一个ShutdownHookThread对象JVM钩子在程序终止时调用controllerManager.shutdown()释放资源
Runtime.getRuntime().addShutdownHook(new ShutdownHookThread(log, (CallableVoid) () - {controllerManager.shutdown();return null;}));
public void shutdown() {this.heartbeatManager.shutdown();this.controllerRequestExecutor.shutdown();this.notifyService.shutdown();this.controller.shutdown();this.remotingClient.shutdown();}三、controller.start()运行 运行NettyRemotingServer启动一个NettyRemotingServer用于接收和处理客户端的请求 运行remotingClient启动一个NettyRemotingClient用于向其他服务发送请求 运行FileWatch调用它的start方法启动一个文件监视服务用于动态加载证书文件 用来跟踪SSL 运行RouteInfoManager启动一个路由信息管理器用于维护Broker和Topic的路由关系 unRegisterService.start() 提供了一种以批处理方式注销Broker的机制。扫描离线的Broker BlockingQueuetake()获取数据while循环获取不活跃的broker。take()是阻塞的drainTo()方法获取队列中的全部数据。 registerBroker() 接收Broker每隔30秒上报Broker信息
