ICode9

精准搜索请尝试: 精确搜索
首页 > 编程语言> 文章详细

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

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

标签: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

专注分享技术,共同学习,共同进步。侵权联系[admin#icode9.com]

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

ICode9版权所有