ICode9
精准搜索请尝试： 精确搜索
首页
编程语言>
数据库
系统相关
互联网
其他分享
Python
Java
JavaScript
android
PHP
首页 > 编程语言> 文章详细

微服务系列之Nacos注册中心源码解读

2020-02-25 15:07:45  阅读：247  来源： 互联网

标签：dom RaftCore Nacos alibaba 解读 源码 nacos naming com


源码下载地址: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));

 

  1. 获取集群域，可以理解为根据服务名和命名空间构建的一个集群域
  2. 如果不存在这样的集群域，构建一个新的
  3. 根据获取到的集群域，筛选出该clusterName的集群，如果不存在，则创建
  4. 构建新的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逻辑

  1. 判断发送端term是否大于自身term，如果大于，则投票给发送者，如果小于投票给自身
  2. 返回自身最新的投票信息

选举过程结束后，就形成了集群，集群接收请求时也会按照raft协议来进行

    如果自身不是leader节点，则将请求转发给leader
    如果自身是leader节点，则处理该请求，并分发给其他follower。（如果有半数的follower节点成功处理该分发请求，则认为本次请求处理成功）

另外，在选举结束后，leader节点还会与follower节点通过http接口的方式进行心跳检测，并同步自身存储的数据信息

 

TO UP! 发布了7 篇原创文章 · 获赞 3 · 访问量 373 私信 关注

标签：dom,RaftCore,Nacos,alibaba,解读,源码,nacos,naming,com
来源： https://blog.csdn.net/qq_41967563/article/details/104496151

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

关于我们 | 联系我们 | 留言反馈

专注分享技术，共同学习，共同进步。侵权联系[81616952@qq.com]

Copyright (C)ICode9.com, All Rights Reserved.

ICode9版权所有