源码下载地址:https://github.com/alibaba/nacos
从官网架构图中可以看出nacos内部提供了nacos-namign和nacos-config两个服务,作为注册中心和配置中心,nacos-core作为nacos-naming和nacos-config两个模块的公共支持部分,提供了一些相关工具类
Nacos通过提供一系列的http接口来提供Naming服务和Config服务的实现
服务注册URI:/nacos/v1/ns/instance POST
服务取消注册URI:/nacos/v1/ns/instance DELETE
心跳检测URI:/nacos/v1/ns/instance/beat PUT
可以看到其遵循了REST API的风格。
并且我们可以直观的认识到,Nacos通过http这样无状态的协议来进行client-server端的通信。
那么差不多可以开始进入源码分析的部分了,先了解一下core模块,为Config Service和Naming Service的公共支持组件,core仅仅提供了一些工具类以及使用spring boot starter的方式将nacos配置文件加载到Environment。
首先介绍一个处理请求参数转化的一个常用的类
required方法通过参数名key,解析HttpServletRequest
请求中的参数,并转码为UTF-8编码。
optional方法在required方法的基础上增加了默认值,如果获取不到,则返回默认值。
nacos server-client使用了http协议来交互,那么在server端必定提供了http接口的入口,并且在core模块看到其依赖了spring boot starter,一个合理的猜想是:其http接口由集成了Spring的web服务器支持,简单地说就是像我们平时写的业务服务一样,有controller层和service层。
以OpenAPI作为入口来学习,找到/nacos/v1/ns/instance服务注册接口:
com.alibaba.nacos.naming.controllers.InstanceController
RequestMapping(value = "/instance", method = RequestMethod.POST)
public String register(HttpServletRequest request) throws Exception {
OverrideParameterRequestWrapper requestWrapper = OverrideParameterRequestWrapper.buildRequest(request);
String serviceJson = WebUtils.optional(request, "service", StringUtils.EMPTY);
// set service info:
if (StringUtils.isNotEmpty(serviceJson)) {
JSONObject service = JSON.parseObject(serviceJson);
requestWrapper.addParameter("serviceName", service.getString("name"));
}
return regService(requestWrapper);
}
可以看到在controller层只是将请求参数解析出来,封装成requestWrapper
后交给下层的服务来处理
public String regService(HttpServletRequest request) throws Exception {
String dom = WebUtils.required(request, "serviceName");
String tenant = WebUtils.optional(request, "tid", StringUtils.EMPTY);
String app = WebUtils.optional(request, "app", "DEFAULT");
String env = WebUtils.optional(request, "env", StringUtils.EMPTY);
String metadata = WebUtils.optional(request, "metadata", StringUtils.EMPTY);
String namespaceId = WebUtils.optional(request, Constants.REQUEST_PARAM_NAMESPACE_ID, UtilsAndCommons.getDefaultNamespaceId());
VirtualClusterDomain virtualClusterDomain = (VirtualClusterDomain) domainsManager.getDomain(namespaceId, dom);
IpAddress ipAddress = getIPAddress(request);
ipAddress.setApp(app);
ipAddress.setServiceName(dom);
ipAddress.setInstanceId(ipAddress.generateInstanceId());
ipAddress.setLastBeat(System.currentTimeMillis());
if (StringUtils.isNotEmpty(metadata)) {
ipAddress.setMetadata(UtilsAndCommons.parseMetadata(metadata));
}
if (virtualClusterDomain == null) {
Lock lock = domainsManager.addLockIfAbsent(UtilsAndCommons.assembleFullServiceName(namespaceId, dom));
Condition condition = domainsManager.addCondtion(UtilsAndCommons.assembleFullServiceName(namespaceId, dom));
UtilsAndCommons.RAFT_PUBLISH_EXECUTOR.execute(new Runnable() {
@Override
public void run() {
try {
regDom(request);
} catch (Exception e) {
Loggers.SRV_LOG.error("[REG-SERIVCE] register service failed, service:" + dom, e);
}
}
});
try {
lock.lock();
condition.await(5000, TimeUnit.MILLISECONDS);
} finally {
lock.unlock();
}
virtualClusterDomain = (VirtualClusterDomain) domainsManager.getDomain(namespaceId, dom);
}
if (virtualClusterDomain != null) {
if (!virtualClusterDomain.getClusterMap().containsKey(ipAddress.getClusterName())) {
doAddCluster4Dom(request);
}
if (Loggers.SRV_LOG.isDebugEnabled()) {
Loggers.SRV_LOG.debug("reg-service add ip: {}|{}", dom, ipAddress.toJSON());
}
Map<String, String[]> stringMap = new HashMap<>(16);
stringMap.put("dom", Arrays.asList(dom).toArray(new String[1]));
stringMap.put("ipList", Arrays.asList(JSON.toJSONString(Arrays.asList(ipAddress))).toArray(new String[1]));
stringMap.put("json", Arrays.asList("true").toArray(new String[1]));
stringMap.put("token", Arrays.asList(virtualClusterDomain.getToken()).toArray(new String[1]));
addIP4Dom(OverrideParameterRequestWrapper.buildRequest(request, stringMap));
} else {
throw new IllegalArgumentException("dom not found: " + dom);
}
return "ok";
}
这里有几个参数需要理解,dom/app/metadata
dom:域,实质上就是服务名,微服务中一个服务会有多个实例
app:Property of service which can be used to identify the service provider.
层级关系是这样的
一个namespace -> 多个cluster
一个cluster -> 多个dom(服务)
一个dom(服务)-> 多个实例
app、metadata则是服务上的属性,用于标志服务所属应用以及标志服务特性。
metadata是一个map,以key、value的形式存储服务的特殊属性,例如:
可以通过enableSSL:true来标志是否开启身份验证
简单而言,metadata是提供了用户*扩展的属性,类似于数据库中预留的表字段
分解这一段代码的重要逻辑:
VirtualClusterDomain virtualClusterDomain = (VirtualClusterDomain) domainsManager.getDomain(namespaceId, dom);
regDom(request);
if (!virtualClusterDomain.getClusterMap().containsKey(ipAddress.getClusterName())) {
doAddCluster4Dom(request);
}
addIP4Dom(OverrideParameterRequestWrapper.buildRequest(request, stringMap));
- 获取集群域,可以理解为根据服务名和命名空间构建的一个集群域
- 如果不存在这样的集群域,构建一个新的
- 根据获取到的集群域,筛选出该clusterName的集群,如果不存在,则创建
- 构建新的IP,将服务实例更新到该集群域中
前面说了:
一个namespace下会存在多个cluster
一个cluster下会存在多个dom
这里的VirtualClusterDomain则是根据namespace和dom定位到dom集合
一个dom下存在多个实例
那么注册服务或更新服务信息,则是注册到该dom集合上
直接进入主要逻辑
if (RaftCore.isLeader()) {
try {
RaftCore.OPERATE_LOCK.lock();
OverrideParameterRequestWrapper requestWrapper = OverrideParameterRequestWrapper.buildRequest(request);
requestWrapper.addParameter("clientIP", NetUtils.localServer());
requestWrapper.addParameter("notify", "true");
requestWrapper.addParameter("term", String.valueOf(RaftCore.getPeerSet().local().term));
requestWrapper.addParameter("timestamp", String.valueOf(timestamp));
onAddIP4Dom(requestWrapper);
proxyParams.put("clientIP", NetUtils.localServer());
proxyParams.put("notify", "true");
proxyParams.put("term", String.valueOf(RaftCore.getPeerSet().local().term));
proxyParams.put("timestamp", String.valueOf(timestamp));
if (domain.getEnableHealthCheck() && !domain.getEnableClientBeat()) {
syncOnAddIP4Dom(namespaceId, dom, proxyParams);
} else {
asyncOnAddIP4Dom(proxyParams);
}
} finally {
RaftCore.OPERATE_LOCK.unlock();
}
}
onAddIP4Dom(requestWrapper);由自身处理服务注册的任务
syncOnAddIP4Dom(namespaceId, dom, proxyParams);
asyncOnAddIP4Dom(proxyParams);//根据域的属性来决定使用同步或异步逻辑
onAddIP4Dom(requestWrapper);最终进入com.alibaba.nacos.naming.core.DomainsManager#easyUpdateIP4Dom
下面的两个同步、异步的逻辑,则是将本机处理成功后的请求以leader身份转发给其他follower,同步增量请求。
这样就完成了一次服务注册过程,简单的缕一缕:
当Naming Server接收服务注册请求时,如果当前的身份不是leader,则转发给leader来处理,如果当前的身份是leader,则在本地直接处理服务注册的请求,并发送同步请求给其他follower。
那么,处理服务注册的请求具体又是如何进行的?
在com.alibaba.nacos.naming.core.DomainsManager#easyUpdateIP4Dom能找到这样的一段逻辑
Datum datum = new Datum();
datum.key = key;
datum.value = value;
...
RaftCore.onPublish(datum, peer, increaseTerm);
继续进入com.alibaba.nacos.naming.raft.RaftCore#onPublish(com.alibaba.nacos.naming.raft.Datum, com.alibaba.nacos.naming.raft.RaftPeer, boolean)
notifier.addTask(datum, Notifier.ApplyAction.CHANGE);
在RaftCore类中有一个内部类,用于单线程不断轮询来处理notifier task:com.alibaba.nacos.naming.raft.RaftCore.Notifier
当有新的notifier task进入,则会执行:
for (RaftListener listener : listeners) {
...
try {
if (action == ApplyAction.CHANGE) {
listener.onChange(datum.key, getDatum(datum.key).value);
continue;
}
if (action == ApplyAction.DELETE) {
listener.onDelete(datum.key, datum.value);
continue;
}
} catch (Throwable e) {
Loggers.RAFT.error("[NACOS-RAFT] error while notifying listener of key: {} {}", datum.key, e);
}
}
由listener执行监听逻辑,而这里的listener又包含了哪些呢?
在com.alibaba.nacos.naming.core.VirtualClusterDomain初始化时,将自身作为监听器注册到RaftCore.Notifier的listeners中来监听notifier事件。
@Override
public void init() {
RaftCore.listen(this);
HealthCheckReactor.scheduleCheck(clientBeatCheckTask);
for (Map.Entry<String, Cluster> entry : clusterMap.entrySet()) {
entry.getValue().init();
}
}
所以,这里在本机执行服务注册的过程中会令notifier监听到服务实例列表的变化而作出反应
最终执行com.alibaba.nacos.naming.core.VirtualClusterDomain#onChange
->com.alibaba.nacos.naming.core.VirtualClusterDomain#updateIPs
->com.alibaba.nacos.naming.core.Cluster#updateIPs
最终更新内存中的IPs,似乎还是没有找到数据持久化的地方,难道是像eureka一样保存在内存中?
原来是我忽略了一段代码,在com.alibaba.nacos.naming.raft.RaftCore#onPublish(com.alibaba.nacos.naming.raft.Datum, com.alibaba.nacos.naming.raft.RaftPeer, boolean)中
// do apply
if (datum.key.startsWith(UtilsAndCommons.DOMAINS_DATA_ID_PRE) || UtilsAndCommons.INSTANCE_LIST_PERSISTED) {
RaftStore.write(datum);
}
继续进入com.alibaba.nacos.naming.raft.RaftStore#write
FileChannel fc = null;
ByteBuffer data = ByteBuffer.wrap(JSON.toJSONString(datum).getBytes("UTF-8"));
try {
fc = new FileOutputStream(cacheFile, false).getChannel();
fc.write(data, data.position());
fc.force(true);
} catch (Exception e) {
MetricsMonitor.getDiskException().increment();
throw e;
} finally {
if (fc != null) {
fc.close();
}
}
最初以为是异步持久化,还在疑惑这样似乎会存在很多问题的时候…
找到光明大道了,代码证明了,同步写入!
并通过fc.force(true)的方式确保文件同步写入完成!
那么我们知道了Nacos的存储机制,不仅在内存维护了一份数据,还在文件保存了一份数据。
并且在内存中维护的数据是以异步的方式更新,而在文件中保存的数据则是同步执行,这意味着数据是完全以文件数据为准的。
Nacos是可以检测服务健康状况的,来看看它是怎么检测的:
还记得前面提到的类吗?
VirtualClusterDomain,其内部维护了多个cluster对象
而每个cluster对象中维护了一个Set
IpAddress则代表了一个实例的ip等信息
在cluster对象初始化时会初始化心跳检测任务并启动心跳检测的线程
这个线程会不断的轮询当前的IpAddress列表,并对每个IpAddress进行心跳检测
Nacos提供了三种心跳检测的实现:mysql、HTTP、TCP
默认使用TCP的实现,那就来简单了解一下TCP的实现吧
com.alibaba.nacos.naming.healthcheck.TcpSuperSenseProcessor
代码感兴趣的可以去了解一下,这里就不贴了,开始解读
它首先提供了一个抽象的心跳检测处理器,这里的处理器会根据用户配置来决定使用哪个处理器
处理器处理的逻辑由一个线程池来定期的执行,TCP的实现则是从cluster对象获取最新的内存中的IPs,然后对于每个IPs创建一个Beat对象,这个Beat对象则是用于一次心跳检测,将Beat对象加入队列,由负责心跳检测的线程池来轮询取出Beat对象,并进行批量处理,在批量处理时与该实例的IP建立nio socket连接(这里需要注意的是server端此时的角色是client,其通过socket.connect来连接client),并通过定时检测channel、key是否有效来判断client是否健康。
又有一个大问题,既然Nacos的服务注册是通过http协议这样无状态的协议来注册的,又如何让客户端收到最新的服务列表信息?
关注这个类com.alibaba.nacos.naming.push.PushService
其在内部根据服务名维护了一个{key=serviceName,value=Client}的map
当有服务注册到Nacos时,还不会将自身信息封装成Client加入到PushService的map中,而是在服务注册后进行服务列表拉取时,将client服务封装成Client加入其中。
当server端服务列表发生变化时,会根据serviceName来给map中的Client推送最新的服务列表信息。
推送方式是采用udp协议,当有服务列表有变化发生时会延迟1秒向client发送udp数据包,udp协议优点在于速度快、代价小,缺点在于不可靠。
虽然可能存在数据包丢失的问题,但由于服务列表一般而言在压力较小时不会出现网络故障,所以一般不会出现udp数据丢包的问题,万一服务列表压力较大时,就相当于牺牲了一致性,换取了性能,让服务节点使用本地缓存的服务列表信息。
到这里,了解到Nacos Naming Service与client端的通信方式以及数据的存储机制,下面继续学习它的集群模式,它是如何实现前文提到的Raft协议的。
Nacos的支持Raft协议的几个类的命名与eureka有些相似:
com.alibaba.nacos.naming.raft.PeerSet:维护当前节点持有的同伴节点信息
com.alibaba.nacos.naming.raft.RaftPeer:维护单个节点的信息,包括term、voteFor、ip、state等
com.alibaba.nacos.naming.raft.RaftCore:作为核心类,负责请求的转发、分发、接收其他节点消息及集群选举
com.alibaba.nacos.naming.raft.RaftProxy:工具类,用于向leader发送请求
com.alibaba.nacos.naming.raft.RaftStore:存储节点信息,与本机文件系统交互
那么要了解选举是如何进行的,就要了解RaftCore和它的内部类com.alibaba.nacos.naming.raft.RaftCore.MasterElection
在RaftCore初始化时会初始化MasterElection,并定时执行该选举任务
->com.alibaba.nacos.naming.raft.RaftCore#init
->124行->com.alibaba.nacos.naming.raft.GlobalExecutor#register(java.lang.Runnable)
那么问题在于这个com.alibaba.nacos.naming.raft.RaftCore.MasterElection是如何工作的
@Override
public void run() {
try {
RaftPeer local = peers.local();
local.leaderDueMs -= GlobalExecutor.TICK_PERIOD_MS;
if (local.leaderDueMs > 0) {
return;
}
// reset timeout
local.resetLeaderDue();
local.resetHeartbeatDue();
sendVote();
} catch (Exception e) {
Loggers.RAFT.warn("[RAFT] error while master election {}", e);
}
}
通过判断自身节点的leaderDueMs
信息来决定是否要执行选举
public volatile long leaderDueMs = RandomUtils.nextLong(0, GlobalExecutor.LEADER_TIMEOUT_MS);
该leaderDueMs默认为(0s-15s)的随机数,而GlobalExecutor.TICK_PERIOD_MS默认为0.5s,这样的效果就是会随机出一个(0-15s)的延迟时间来进入选举状态,这样的好处则是避免不同的节点同时启动,同时进入选举状态而出现多个竞争者,如果一个节点接受到先进入状态的竞争者消息,则会直接放弃当leader。
继续进入com.alibaba.nacos.naming.raft.RaftCore.MasterElection#sendVote
当一个节点经过随机时间后进入选举阶段,开始向其他节点发送选票。
这段逻辑做了几件事:
令本地的term加1
投自己一票
修改自身状态为CANDIDATE
通过其他节点提供的http接口,发送给其他节点自身最新的投票信息
当对方节点接受到投票信息并同意后返回了同意响应消息时,更新本地维护的peers列表信息,并统计投票,当某一节点的得票数大于等于所有节点的一半时,设置该节点为leader节点
这样就完成了一次选举,更新本地维护的peers列表,并决定了leader。
无论本次选举是否成功,选举都会结束,如果选举失败,该节点将无法正常的提供服务。
虽然选举失败的概率非常低,但也有可能存在这样的情况,当由于网络不稳定等原因短时间无法与其他节点正常通信时,就会选举失败。
com.alibaba.nacos.naming.web.RaftCommands
@NeedAuth
@RequestMapping("/vote")
public JSONObject vote(HttpServletRequest request, HttpServletResponse response) throws Exception {
RaftPeer peer = RaftCore.MasterElection.receivedVote(
JSON.parseObject(WebUtils.required(request, "vote"), RaftPeer.class));
return JSON.parseObject(JSON.toJSONString(peer));
}
最终进入com.alibaba.nacos.naming.raft.RaftCore.MasterElection#receivedVote
逻辑
- 判断发送端term是否大于自身term,如果大于,则投票给发送者,如果小于投票给自身
- 返回自身最新的投票信息
选举过程结束后,就形成了集群,集群接收请求时也会按照raft协议来进行
如果自身不是leader节点,则将请求转发给leader
如果自身是leader节点,则处理该请求,并分发给其他follower。(如果有半数的follower节点成功处理该分发请求,则认为本次请求处理成功)
另外,在选举结束后,leader节点还会与follower节点通过http接口的方式进行心跳检测,并同步自身存储的数据信息