1、NameServer的作用
Name Server 是专为 RocketMQ 设计的轻量级名称服务，具有简单、可集群横吐扩展、无状态，节点之间互不通信等特点。
整个Rocketmq集群的工作原理如下图所示：
任何Producer、Consumer、Broker与所有NameServer通信，向NameServer请求或者发送数据。而且都是单向的，Producer和Consumer请求数据，Broker发送数据。正是因为这种单向的通信，RocketMQ水平扩容变得很容易。
Broker集群
Broker用于接收生产者发送消息，或者消费者消费消息的请求。一个Broker集群由多组Master/Slave组成，Master可写可读，Slave只可以读，Master将写入的数据同步给Slave。每个Broker节点，在启动时，都会遍历NameServer列表，与每个NameServer建立长连接，注册自己的信息，之后定时上报。
Producer集群
消息的生产者，通过NameServer集群获得Topic的路由信息，包括Topic下面有哪些Queue，这些Queue分布在哪些Broker上等。Producer只会将消息发送到Master节点上，因此只需要与Master节点建立连接。
Consumer集群
消息的消费者，通过NameServer集群获得Topic的路由信息，连接到对应的Broker上消费消息。注意，由于Master和Slave都可以读取消息，因此Consumer会与Master和Slave都建立连接。
2、NameServer类结构
- NamesrvStartup： NameServer的启动类；
- NamesrvController： NameServer的核心控制类；
- KVConfigManager： 读取或变更NameServer的配置属性，加载NamesrvConfig中配置的配置文件到内存；
- KVConfigSerializeWrapper： NameServer配置信息序列化包装类；
- RouteInfoManager： NameServer数据的载体，记录Broker，Topic等信息；
- DefaultRequestProcessor： NameServer处理请求的请求类，负责处理所有与NameServer交互的请求；
- BrokerHousekeepingService： BrokerHouseKeepingService实现ChannelEventListener接口，可以说是通道在发送异常时的回调方法（Nameserver与Broker的连接通道在关闭、通道发送异常、通道空闲时）；
- NamesrvConfig： NamesrvConfig,主要指定nameserver的相关配置目录属性；
- NettyRemotingServer： Netty服务类；
（1）NameServer启动流程
NameServer的启动是由NamesrvStartup完成的，启动过程如下：
- 获取并解析配置参数，包括NamesrvConfig和NettyServerConfig；
- 调用NamesrvController.initialize()初始化NamesrvController；若初始化失败，则直接关闭NamesrvController；
- 然后调用NamesrvController.start()方法来开启NameServer服务；
- 注册ShutdownHookThread服务。在JVM退出之前，调用NamesrvController.shutdown()来进行关闭服务，释放资源；
public class NamesrvStartup {
private static InternalLogger log;
private static Properties properties = null;
private static CommandLine commandLine = null;
public static void main(String[] args) {
main0(args);
}
public static NamesrvController main0(String[] args) {
try {
// 创建NamesrvController
NamesrvController controller = createNamesrvController(args);
// 启动NamesrvController
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;
}
public static NamesrvController createNamesrvController(String[] args) throws IOException, JoranException {
System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION));
//PackageConflictDetect.detectFastjson();
// 构建命令行
Options options = ServerUtil.buildCommandlineOptions(new Options());
commandLine = ServerUtil.parseCmdLine("mqnamesrv", args, buildCommandlineOptions(options), new PosixParser());
if (null == commandLine) {
System.exit(-1);
return null;
}
// nameServer配置参数
final NamesrvConfig namesrvConfig = new NamesrvConfig();
// netty server 配置参数
final NettyServerConfig nettyServerConfig = new NettyServerConfig();
nettyServerConfig.setListenPort(9876);
// 命令行参数是否包含配置文件
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();
}
}
// 是否打印参数
if (commandLine.hasOption('p')) {
// 都不打印
MixAll.printObjectProperties(null, namesrvConfig);
MixAll.printObjectProperties(null, nettyServerConfig);
System.exit(0);
}
// 设置命令行的参数，优先级高（会覆盖掉配置文件的配置项）
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);
}
// 配置Logger
LoggerContext lc = (LoggerContext) LoggerFactory.getILoggerFactory();
JoranConfigurator configurator = new JoranConfigurator();
configurator.setContext(lc);
lc.reset();
configurator.doConfigure(namesrvConfig.getRocketmqHome() + "/conf/logback_namesrv.xml");
log = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);
// 控制台打印参数
MixAll.printObjectProperties(log, namesrvConfig);
MixAll.printObjectProperties(log, nettyServerConfig);
// 创建 NamesrvController
final NamesrvController controller = new NamesrvController(namesrvConfig, nettyServerConfig);
// 注册配置参数，防止丢失
// remember all configs to prevent discard
controller.getConfiguration().registerConfig(properties);
return controller;
}
public static NamesrvController start(final NamesrvController controller) throws Exception {
if (null == controller) {
throw new IllegalArgumentException("NamesrvController is null");
}
// 初始化NamesrvController
boolean initResult = controller.initialize();
// 初始化失败
if (!initResult) {
// 关闭NamesrvController
controller.shutdown();
// 关闭JVM
System.exit(-3);
}
// 注册关闭钩子方法：当JVM关闭的时候，先关闭NamesrvController
Runtime.getRuntime().addShutdownHook(new ShutdownHookThread(log, new Callable<Void>() {
@Override
public Void call() throws Exception {
// 关闭NamesrvController
controller.shutdown();
return null;
}
}));
// 启动NamesrvController
controller.start();
return controller;
}
public static void shutdown(final NamesrvController controller) {
controller.shutdown();
}
public static Options buildCommandlineOptions(final Options options) {
Option opt = new Option("c", "configFile", true, "Name server config properties file");
opt.setRequired(false);
options.addOption(opt);
opt = new Option("p", "printConfigItem", false, "Print all config item");
opt.setRequired(false);
options.addOption(opt);
return options;
}
}
调用NamesrvController.initialize()初始化NamesrvController
public boolean initialize() {
// 加载KV配置
this.kvConfigManager.load();
// 初始化通信层
this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);
// 初始化线程池
this.remotingExecutor =
Executors.newFixedThreadPool(nettyServerConfig.getServerWorkerThreads(),
new ThreadFactoryImpl("RemotingExecutorThread_"));
this.registerProcessor();
// 增加定时任务
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
NamesrvController.this.routeInfoManager.scanNotActiveBroker();
}
}, 5, 10, TimeUnit.SECONDS);
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
NamesrvController.this.kvConfigManager.printAllPeriodically();
}
}, 1, 10, TimeUnit.MINUTES);
// this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
//
// @Override
// public void run() {
// NamesrvController.this.routeInfoManager.printAllPeriodically();
// }
// }, 1, 5, TimeUnit.MINUTES);
return true;
}
加载KV配置，创建NettyServer网络处理对象，然后开启两个定时任务，此类定时任务统称为心跳检测
- NameServer每隔10秒扫描一次Broker，移除处于不激活状态的Broker
- NameServer每隔10分钟打印一次KV配置
3、NameServer如何保证数据的最终一致
NameServer作为一个名称服务，需要提供服务注册、服务剔除、服务发现这些基本功能，但是NameServer节点之间并不通信，在某个时刻各个节点数据可能不一致的情况下，如何保证客户端可以最终拿到正确的数据。下面分别从路由元信息、路由注册、路由剔除，路由发现四个角度进行介绍。
路由元信息
NameServer路由实现类: RoutelnfoManager
RouteInfoManager作为NameServer数据的载体，记录Broker、Topic、QueueData等信息。
Broker在启动时会将Broker信息、Topic信息、QueueData信息注册到所有的NameServer上，并和所有NameServer节点保持长连接，之后也会定时注册信息；
Producer、Consumer也会和其中一个NameServer节点保持长连接，定时从NameServer中获取Topic路由信息；
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;
- topicQueueTable: Topic 消息队列路由信息，消息发送时根据路由表进行负 载均衡 。
- brokerAddrTable : Broker 基础信息， 包含 brokerName、 所属集群名称 、 主备 Broker
地址。 - clusterAddrTable: Broker 集群信息，存储集群中所有 Broker 名称 。
- brokerLiveTable: Broker 状态信息 。 NameServer 每次 收到心跳包时会 替换该信 息 。
- filterServerTable : Broker上的 FilterServer列表，用于类模式消息过滤，
RocketMQ 基于订阅发布机制 ， 一个 Topic 拥有 多 个消息队 列 ，一 个 Broker 为每一主 题默 认创建 4 个读队列 4 个写队列 。 多个 Broker 组成 一个集群 ， BrokerName 由相同的多台 Broker 组 成 Master-Slave 架构 ， brokerId 为 0 代表 Master， 大于 0 表示 Slave。 BrokerLivelnfo 中 的 lastUpdateTimestamp 存储上次收到 Broker 心跳包的时间 。
路由注册
对于Zookeeper、Etcd这样强一致性组件，数据只要写到主节点，内部会通过状态机将数据复制到其他节点，Zookeeper使用的是Zab协议，etcd使用的是raft协议。
但是NameServer节点之间是互不通信的，无法进行数据复制。RocketMQ采取的策略是，在Broker节点在启动的时候，轮训NameServer列表，与每个NameServer节点建立长连接，发起注册请求。NameServer内部会维护一个Broker表，用来动态存储Broker的信息。
同时，Broker节点为了证明自己是存活的，会将最新的信息上报给NameServer，然后每隔30秒向NameServer发送心跳包，心跳包中包含 BrokerId、Broker地址、Broker名称、Broker所属集群名称等等，然后NameServer接收到心跳包后，会更新时间戳，记录这个Broker的最新存活时间。
NameServer在处理心跳包的时候，存在多个Broker同时操作一张Broker表，为了防止并发修改Broker表导致不安全，路由注册操作引入了ReadWriteLock读写锁，这个设计亮点允许多个消息生产者并发读，保证了消息发送时的高并发，但是同一时刻NameServer只能处理一个Broker心跳包，多个心跳包串行处理。这也是读写锁的经典使用场景，即读多写少。
Broker端心跳包发送
Broker端心跳包发送( BrokerController#start)
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
try {
BrokerController.this.registerBrokerAll(true, false);
}
catch (Exception e) {
log.error("registerBrokerAll Exception", e);
}
}
}, 1000 * 10, 1000 * 30, TimeUnit.MILLISECONDS);
该方法主要是遍历 NameServer列表， Broker消息服务器依次向 NameServer发送心跳包。
public RegisterBrokerResult registerBrokerAll(//
final String clusterName,// 1
final String brokerAddr,// 2
final String brokerName,// 3
final long brokerId,// 4
final String haServerAddr,// 5
final TopicConfigSerializeWrapper topicConfigWrapper,// 6
final List<String> filterServerList,// 7
final boolean oneway// 8
) {
RegisterBrokerResult registerBrokerResult = null;
List<String> nameServerAddressList = this.remotingClient.getNameServerAddressList();
if (nameServerAddressList != null) {
for (String namesrvAddr : nameServerAddressList) {
try {
RegisterBrokerResult result =
this.registerBroker(namesrvAddr, clusterName, brokerAddr, brokerName, brokerId,
haServerAddr, topicConfigWrapper, filterServerList, oneway);
if (result != null) {
registerBrokerResult = result;
}
log.info("register broker to name server {} OK", namesrvAddr);
}
catch (Exception e) {
log.warn("registerBroker Exception, " + namesrvAddr, e);
}
}
}
NameServer端处理心跳包
Step1:路由 注册需要加写锁 ，防止并发修改 RoutelnfoManager 中的路由 表 。 Broker 所属 集群是否存在， 如果不存在，则创 建，然 后将 broker 名加入到集群 合中。
Step2 :维护 BrokerData信息，首先从 brokerAddrTable根据 BrokerName尝试获取 Broker信息，如果不存在， 则新建 BrokerData并放入到 brokerAddrTable, registerFirst设 置为 true;如果存在 ， 直接替换原先的， registerFirst设置为 false，表示非第一次注册 。
Step3 :如果Broker为Master，并且BrokerTopic配置信息发生变化或者是初次注册， 则需要创建或更新 Topic路由元数据，填充 topicQueueTable， 其实就是为默认主题自动注 册路由信息，其中包含 MixAII.DEFAULT TOPIC 的路由信息。 当消息生产者发送主题时， 如果该主题未创建并且BrokerConfig的autoCreateTopicEnable为true时， 将返回MixAII. DEFAULT TOPIC的路由信息。
路由剔除
正常情况下，如果Broker关闭，则会与NameServer断开长连接，Netty的通道关闭监听器会监听到连接断开事件，然后会将这个Broker信息剔除掉。
Broker 每隔 30s 向 NameServer 发送一个心跳包，心跳包中包含 BrokerId、Broker地址、Broker名称、 Broker所属集群名称、Broker关联的 FilterServer列表。 但是如果 Broker若机 ， NameServer无法收到心跳包，此时 NameServer如何来剔除这些失 效的 Broker 呢? Name Server会每隔 IOs 扫描 brokerLiveTable状态表，如果 BrokerLive 的 lastUpdateTimestamp 的时间戳距当前时间超过 120s，则认为 Broker失效，移除该 Broker, 关闭与Broker连接，并同时更新topicQueueTable、 brokerAddrTable、 brokerLiveTable、 filterServerTable。
RocktMQ 有两个触发点来触发路由删除 。
- NameServer定时扫描 brokerLiveTable检测上次心跳包与 当前系统时间的时间差， 如果时间戳大于 120s，则需要移除该 Broker 信息 。
- Broker在正常被关闭的情况下，会执行 unregisterBroker指令。
路由发现
RocketMQ 路由发现是非实时的，当 Topic 路由出现变化后， NameServer不主动推送给客户端 ， 而 是由客户端定时拉取主题最新的路由 。
com.alibaba.rocketmq.filtersrv.processor.DefaultRequestProcessor#pullMessageForward
private RemotingCommand pullMessageForward(final ChannelHandlerContext ctx, final RemotingCommand request)
throws Exception {
final RemotingCommand response =
RemotingCommand.createResponseCommand(PullMessageResponseHeader.class);
final PullMessageResponseHeader responseHeader =
(PullMessageResponseHeader) response.readCustomHeader();
final PullMessageRequestHeader requestHeader =
(PullMessageRequestHeader) request.decodeCommandCustomHeader(PullMessageRequestHeader.class);
// 由于异步返回，所以必须要设置
response.setOpaque(request.getOpaque());
DefaultMQPullConsumer pullConsumer = this.filtersrvController.getDefaultMQPullConsumer();
final FilterClassInfo findFilterClass =
this.filtersrvController.getFilterClassManager().findFilterClass(
requestHeader.getConsumerGroup(), requestHeader.getTopic());
if (null == findFilterClass) {
response.setCode(ResponseCode.SYSTEM_ERROR);
response.setRemark("Find Filter class failed, not registered");
return response;
}
if (null == findFilterClass.getMessageFilter()) {
response.setCode(ResponseCode.SYSTEM_ERROR);
response.setRemark("Find Filter class failed, registered but no class");
return response;
}
responseHeader.setSuggestWhichBrokerId(MixAll.MASTER_ID);
// 构造从Broker拉消息的参数
MessageQueue mq = new MessageQueue();
mq.setTopic(requestHeader.getTopic());
mq.setQueueId(requestHeader.getQueueId());
mq.setBrokerName(this.filtersrvController.getBrokerName());
long offset = requestHeader.getQueueOffset();
int maxNums = requestHeader.getMaxMsgNums();
final PullCallback pullCallback = new PullCallback() {
@Override
public void onSuccess(PullResult pullResult) {
responseHeader.setMaxOffset(pullResult.getMaxOffset());
responseHeader.setMinOffset(pullResult.getMinOffset());
responseHeader.setNextBeginOffset(pullResult.getNextBeginOffset());
response.setRemark(null);
switch (pullResult.getPullStatus()) {
case FOUND:
response.setCode(ResponseCode.SUCCESS);
List<MessageExt> msgListOK = new ArrayList<MessageExt>();
try {
for (MessageExt msg : pullResult.getMsgFoundList()) {
boolean match = findFilterClass.getMessageFilter().match(msg);
if (match) {
msgListOK.add(msg);
}
}
// 有消息返回
if (!msgListOK.isEmpty()) {
returnResponse(requestHeader.getConsumerGroup(), requestHeader.getTopic(), ctx,
response, msgListOK);
return;
}
// 全部都被过滤掉了
else {
response.setCode(ResponseCode.PULL_RETRY_IMMEDIATELY);
}
}
// 只要抛异常，就终止过滤，并返回客户端异常
catch (Throwable e) {
final String error =
String.format("do Message Filter Exception, ConsumerGroup: %s Topic: %s ",
requestHeader.getConsumerGroup(), requestHeader.getTopic());
log.error(error, e);
response.setCode(ResponseCode.SYSTEM_ERROR);
response.setRemark(error + RemotingHelper.exceptionSimpleDesc(e));
returnResponse(requestHeader.getConsumerGroup(), requestHeader.getTopic(), ctx,
response, null);
return;
}
break;
case NO_MATCHED_MSG:
response.setCode(ResponseCode.PULL_RETRY_IMMEDIATELY);
break;
case NO_NEW_MSG:
response.setCode(ResponseCode.PULL_NOT_FOUND);
break;
case OFFSET_ILLEGAL:
response.setCode(ResponseCode.PULL_OFFSET_MOVED);
break;
default:
break;
}
returnResponse(requestHeader.getConsumerGroup(), requestHeader.getTopic(), ctx, response,
null);
}
@Override
public void onException(Throwable e) {
response.setCode(ResponseCode.SYSTEM_ERROR);
response.setRemark("Pull Callback Exception, " + RemotingHelper.exceptionSimpleDesc(e));
returnResponse(requestHeader.getConsumerGroup(), requestHeader.getTopic(), ctx, response,
null);
return;
}
};
pullConsumer.pullBlockIfNotFound(mq, null, offset, maxNums, pullCallback);
return null;
}
- 调用 RouterlnfoManager 的方法，从路由 表 topicQueueTable、 brokerAddrTable、 filterServerTable中分别填充TopicRouteData中的List<Queu巳Data>、List<BrokerData>和 filterServer 地址表 。
- 如果找到主题对应的路由信息并且该主题为顺序消息，则从 NameServer KVconfig 中获取关于顺序消息相关 的配置填充路由信息 。
如果找不到路由信息 CODE 则使用 TOPIC NOT_EXISTS ，表示没有找到对应的路由 。
