当前位置: 首页 > news >正文

RocketMQ源码分析之NameServer

1、RocketMQ组件概述

  • NameServer
    NameServer相当于配置中心,维护Broker集群、Broker信息、Broker存活信息、主题与队列信息等。NameServer彼此之间不通信,每个Broker与集群内所有的Nameserver保持长连接。

2、源码分析NameServer

本文不对 NameServer 与 Broker、Producer 集群、Consumer 集群的网络通信做详细解读(该系列后续专门进行讲解)

本文重点关注 NameServer 作为 MQ 集群的配置中心存储什么信息。

2.1 源码分析NamesrvController

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

2.1.1 NamesrvConfig

NamesrvConfig,主要指定 nameserver 的相关配置属性:

  • kvConfigPath(kvConfig.json)。
  • mqhome/namesrv/namesrv.properties。
  • orderMessageEnable,是否开启顺序消息功能,默认为false。

2.1.2 ScheduledExecutorService

private final ScheduledExecutorService scheduledExecutorService = Executors.

NameServer 定时任务执行线程池,默认定时执行两个任务:

  • 任务1、每隔 10s 扫描 broker ,维护当前存活的Broker信息。
  • 任务2、每隔 10s 打印KVConfig 信息。

2.1.3 KVConfigManager

读取或变更NameServer的配置属性,加载 NamesrvConfig 中配置的配置文件到内存,此类一个亮点就是使用轻量级的非线程安全容器,再结合读写锁对资源读写进行保护。尽最大程度提高线程的并发度。

2.1.4 RouteInfoManager

NameServer 数据的载体,记录 Broker、Topic 等信息。

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

代码@1,NameServer 与 Broker 空闲时长,默认2分钟,在2分钟内 Nameserver 没有收到 Broker 的心跳包,则关闭该连接。

代码@2,读写锁,用来保护非线程安全容器 HashMap。

代码@3,topicQueueTable,主题与队列关系,记录一个主题的队列分布在哪些Broker上,每个Broker上存在该主题的队列个数。QueueData队列描述信息,对应如下属性:

private String brokerName;           // broker的名称private int readQueueNums;           // 读队列个数private int writeQueueNums;          // 写队列个数private int perm;                    // 权限操作

代码@4,brokerAddrTable,所有 Broker 信息,使用 brokerName 当key, BrokerData 信息描述每一个 broker 信息。

// broker所属集群
private String cluster;                           // broker name
private String brokerName;broker 对应的IP:Port,brokerId=0表示Master,大于0表示Slave。             

代码@5,clusterAddrTable,broker 集群信息,每个集群包含哪些 Broker。

代码@6,brokerLiveTable,当前存活的 Broker,该信息不是实时的,NameServer 每10S扫描一次所有的 broker,根据心跳包的时间得知 broker的状态,该机制也是导致当一个 Broker 进程假死后,消息生产者无法立即感知,可能继续向其发送消息,导致失败(非高可用),如何保证消息发送高可用,请关关注该系列后续文章。

2.1.5 BrokerHousekeepingService

BrokerHouseKeepingService 实现 ChannelEventListener接口,可以说是通道在发送异常时的回调方法(Nameserver与 Broker的连接通道在关闭、通道发送异常、通道空闲时),在上述数据结构中移除已宕机的 Broker。

public interface ChannelEventListener {void onChannelConnect(final String remoteAddr, final Channel channel);void onChannelClose(final String remoteAddr, final Channel channel);void onChannelException(final String remoteAddr, final Channel channel);void onChannelIdle(final String remoteAddr, final Channel channel);

2.1.6 NettyServerConfig、RemotingServer 、ExecutorService

这三个属性与网络通信有关,NameServer 与 Broker、Producer、Consume 之间的网络通信,基于 Netty实现。本文借这个机会再次探究 Netty 线程模型与 Netty实战技巧。

源码分析网络通讯之前,我们关注如下问题:

  • NettyServerConfig 的配置含义
  • Netty 线程模型中 EventLoopGroup、EventExecutorGroup 之间的区别与作用
  • 在 Channel 的整个生命周期中,如何保证 Channel 的读写事件至始至终使用同一个线程处理

首先我们先过一下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;

我们带着上面的疑问开始源码分析 org.apache.rocketmq.remoting.netty.NettyRemotingServer。

http://www.lryc.cn/news/7472.html

相关文章:

  • 如何优化认知配比
  • WuThreat身份安全云-TVD每日漏洞情报-2023-02-15
  • Unreal Engine角色涌现行为开发教程
  • vue处理一千张图片进行分页加载
  • (三十三)Vue之消息订阅与发布
  • Http中你必须知道那点事
  • ViewBinding使用入门
  • Retrofit源码分析实践(七)【Retrofit ConvertFactory的功能实现】
  • 介电常数常用测量方法综述
  • 重生之我是赏金猎人(三)-SRC漏洞挖掘-强行多次FUZZ发现某厂商SSRF到redis密码喷洒批量反弹Shell
  • 学会分享,学会生活,分享5款简单易用的软件。
  • Redux 与 Vuex 的区别总结
  • QT(15)- QFile
  • GaiaX开源解读 | 表达式作为逻辑动态化的基础,我们是如何设计的
  • python中adb shell相关操作
  • python 之 资源使用与控制 resource模块
  • 蓝库云|8项关键让你看透企业「数字转型」,零代码是惊喜
  • (五)、编辑页面-发布长文-富文本编辑【uniapp+uinicloud多用户社区博客实战项目(完整开发文档-从零到完整项目)】
  • 你是真的“C”——【经典面试知识点】数据在内存中的大小端存储方式
  • 从零开始的数模(二十六)单因素方差分析
  • C++变量类型
  • win10 安装 vs2015(社区版本)以及opencv-4.5.5
  • 867. 转置矩阵
  • Datawahle组队学习——妙趣横生大数据 Day1
  • 网友眼中越老越吃香的行业,果然是风向变了!
  • 为什么时间序列预测这么难?本文将给你答案
  • STC15系列单片机通过串口多字节数据读写EEPROM操作
  • 计算机网络-ip数据报
  • 从零开始学C
  • 【云原生】手把手带你从零开始搭建kubernetes最新版本实战