Nacos 客户端服务发现源码分析-篇六

注册发现流程

前面的几篇探究了客户端服务注册，服务端发现，以及服务端的健康检测方面的源码。

如果通过一个图去连接他们之间的关系其实，Nacos 客户端的服务发现，其实就是客户端实体参数封装，调用服务端接口请求，获得返回实例结果列表的过程。

对于 NacosService 获取服务列表，在获取服务列表的过程中还涉及到通讯流程协议（Http / gPRC），订阅流程、故障转移等。

那么对于这些到底是怎么实现的呢？我们接着往下一点点的剖析关于 NamingTest 测试类中提供的源码流程。

@Test
public void testServiceList() throws Exception {
    ......
    NamingService namingService = NacosFactory.createNamingService(properties);
namingService.registerInstance("nacos.test.1", instance);
	ThreadUtils.sleep(5000L);
	List<Instance> list = namingService.getAllInstances("nacos.test.1");
    ......
}

前面的关于 registerInstance 方法我们前几篇已经分析完，我们接下来关注 getAllInstance 方法内部是怎么实现的，这里返回的 List 其实就是客户端在注册完毕后返回的实例列表。

我们先打断点看一下，官方提供的 MamingTest 中返回的 list 中都有些什么信息。

而这些数据就是在客户端返回的数据信息，运行之前的 9002 项目我们看观察一下。

我们进入 getAllInstances 方法一探究竟。。。

@Override
public List<Instance> getAllInstances(String serviceName) throws NacosException {
    return getAllInstances(serviceName, new ArrayList<String>());
}

@Override
public List<Instance> getAllInstances(String serviceName, List<String> clusters) throws NacosException {
    return getAllInstances(serviceName, clusters, true);
}

@Override
public List<Instance> getAllInstances(String serviceName, List<String> clusters, boolean subscribe)
        throws NacosException {
    return getAllInstances(serviceName, Constants.DEFAULT_GROUP, clusters, subscribe);
}

@Override
public List<Instance> getAllInstances(String serviceName, String groupName, List<String> clusters,
        boolean subscribe) throws NacosException {
    ServiceInfo serviceInfo;
    String clusterString = StringUtils.join(clusters, ",");
    if (subscribe) {
        serviceInfo = serviceInfoHolder.getServiceInfo(serviceName, groupName, clusterString);
        if (null == serviceInfo) {
            serviceInfo = clientProxy.subscribe(serviceName, groupName, clusterString);
        }
    } else {
        serviceInfo = clientProxy.queryInstancesOfService(serviceName, groupName, clusterString, 0, false);
    }
    List<Instance> list;
    if (serviceInfo == null || CollectionUtils.isEmpty(list = serviceInfo.getHosts())) {
        return new ArrayList<Instance>();
    }
    return list;
}

可以看到在 NacosNamingService 服务当中有大量的 getAllInstances 的重载，提供一些默认的参数。

打个断点玩玩看，内部这些参数都是怎么进行调用，并且完成处理的。

@Override
public List<Instance> getAllInstances(String serviceName, String groupName, List<String> clusters,
        boolean subscribe) throws NacosException {
    ServiceInfo serviceInfo;
    //获取当前的集群
    String clusterString = StringUtils.join(clusters, ",");
    //是否订阅模式（默认 true ）
    if (subscribe) {
        //获取客户端缓存中获取服务信息
        serviceInfo = serviceInfoHolder.getServiceInfo(serviceName, groupName, clusterString);
        if (null == serviceInfo) {
            //如果本地的缓存不存在服务信息，则进行订阅
            serviceInfo = clientProxy.subscribe(serviceName, groupName, clusterString);
        }
    } else {
        //如果未订阅服务信息，则直接从服务器进行查询
        serviceInfo = clientProxy.queryInstancesOfService(serviceName, groupName, clusterString, 0, false);
    }
    List<Instance> list;
    //从服务信息当中获取实例列表
    if (serviceInfo == null || CollectionUtils.isEmpty(list = serviceInfo.getHosts())) {
        return new ArrayList<Instance>();
    }
    return list;
}

分组名称默认：DEFAULT_GROUOP
集群列表：默认为空数组
是否订阅：订阅

可以看到上面的几个关键方法所执行的功能基本就这么几个。。。

可以通过一个流程图进行分析。

那么对于缓存获取与本地缓存内部是怎么实现，以及从服务器端进行查询服务是怎么实现的呢？我们进入具体的实现方法中瞅瞅。。。

客户端服务信息缓存的实现

进入 serviceInfoHolder.getServiceInfo() 方法当中看具体的实现

public ServiceInfo getServiceInfo(final String serviceName, final String groupName, final String clusters) {
    NAMING_LOGGER.debug("failover-mode: " + failoverReactor.isFailoverSwitch());
    String groupedServiceName = NamingUtils.getGroupedName(serviceName, groupName);
    String key = ServiceInfo.getKey(groupedServiceName, clusters);
    //当前的客户端连接是否故障，如果故障 isFailoverSwitch 返回的结果是 false
    if (failoverReactor.isFailoverSwitch()) {
        return failoverReactor.getService(key);
    }
    //返回当前本地缓存中的数据
    return serviceInfoMap.get(key);
}

serviceInfoMap 的实现

//线程安全的
private final ConcurrentMap<String, ServiceInfo> serviceInfoMap;

public ServiceInfoHolder(String namespace, Properties properties) {
    initCacheDir(namespace, properties);
    if (isLoadCacheAtStart(properties)) {
        //DiskCache.read() 从磁盘中读取缓存信息
        this.serviceInfoMap = new ConcurrentHashMap<String, ServiceInfo>(DiskCache.read(this.cacheDir));
    } else {
        //如果是第一次未有缓存则初始化一个默认大小的集合
        this.serviceInfoMap = new ConcurrentHashMap<String, ServiceInfo>(16);
    }
    this.failoverReactor = new FailoverReactor(this, cacheDir);
    this.pushEmptyProtection = isPushEmptyProtect(properties);
}

private void initCacheDir(String namespace, Properties properties) {
    String jmSnapshotPath = System.getProperty(JM_SNAPSHOT_PATH_PROPERTY);
    String namingCacheRegistryDir = "";
    if (properties.getProperty(PropertyKeyConst.NAMING_CACHE_REGISTRY_DIR) != null) {
        namingCacheRegistryDir = File.separator + properties.getProperty(PropertyKeyConst.NAMING_CACHE_REGISTRY_DIR);
    }
    
    if (!StringUtils.isBlank(jmSnapshotPath)) {
        //初始化获取磁盘中缓存信息的 key
        cacheDir = jmSnapshotPath + File.separator + FILE_PATH_NACOS + namingCacheRegistryDir
                + File.separator + FILE_PATH_NAMING + File.separator + namespace;
    } else {
        cacheDir = System.getProperty(USER_HOME_PROPERTY) + File.separator + FILE_PATH_NACOS + namingCacheRegistryDir
                + File.separator + FILE_PATH_NAMING + File.separator + namespace;
    }

通过上面的实现可以发现哈，其实在 Nacos 的内部是通过一个线程安全的 ConcurrentHashMap 进行缓存信息的维护的，其缓冲的具体 key 分两种情况一种是没有其缓存快照的一种是存在缓存快照的。

initCacheDir() 初始化缓存信息

订阅模式的实现

@Override
public ServiceInfo subscribe(String serviceName, String groupName, String clusters) throws NacosException {
    String serviceNameWithGroup = NamingUtils.getGroupedName(serviceName, groupName);
    String serviceKey = ServiceInfo.getKey(serviceNameWithGroup, clusters);
    //开启定时调度 UpdateTask
    serviceInfoUpdateService.scheduleUpdateIfAbsent(serviceName, groupName, clusters);
    //获取缓存中的 ServiceInfo
    ServiceInfo result = serviceInfoHolder.getServiceInfoMap().get(serviceKey);
    if (null == result) {
        //如果缓存中没有数据,则进行订阅逻辑处理,基于 gRPC 协议
        result = grpcClientProxy.subscribe(serviceName, groupName, clusters);
    }
    //serviceInfo 本地缓存处理
    serviceInfoHolder.processServiceInfo(result);
    return result;
}

订阅方法先开启定时任务，这个定时任务的主要作用就是用来定时同步服务的实例信息操作，但是如果是首次这里将会直接返回下一步进行
判断本地的缓存中是否有 serviceInfo 信息存在，如果存在直接返回。如果不存在，则会默认的采用 gRPC 协议进行订阅服务，并且返回 serviceInfo
grpcClientProxy 的 subscribe 订阅方法向服务器端发送一个订阅的请求，并将订阅的信息返回
最终是将 serviceInfo 本地缓存进行处理。在 processServiceInfo 方法中，将获取最新的 ServiceInfo 与本地内存中的 ServiceInfo 进行比较，完成更新，发布时间的变更，磁盘文件存储等操作。对于这一步的操作，其实在 订阅定时任务 中也进行了同样的处理。

以上就是关于客户端服务发现的大致流程。。。。