乐趣区

关于java:因为一次bug的教训我决定手撕Nacos源码先撕客户端源码

jiezi

Nacos 源码分析

Nacos 源码有很多值得咱们学习的中央，为了深刻了解 Nacos，咱们分析源码，剖析如下 2 个知识点：

1:Nacos 对注册核心的拜访原理

2:Nacos 注册服务解决流程

源码环境搭建

1、从官网我的项目上克隆下来，并且检出 1.4.1 版本，导入 idea。nacos 源码环境搭建起来比拟轻松，简直不会报什么谬误，导入后编译装置到本地环境即可

2、找到config 模块中找到 \resources\META-INF\nacos-db.sql，在本地 mysql 中创立数据库nacos-config，将该脚本导入执行创立表。

3、找到 console 模块下的配置文件application.properties，批改相干配置

#*************** Config Module Related Configurations ***************#
### If use MySQL as datasource:
spring.datasource.platform=mysql

### Count of DB:
db.num=1

### Connect URL of DB:
db.url.0=jdbc:mysql://127.0.0.1:3306/nacos-config?characterEncoding=utf8&connectTimeout=1000&socketTimeout=3000&autoReconnect=true&useUnicode=true&useSSL=false&serverTimezone=UTC
db.user.0=nacos
db.password.0=nacos

4、找到 console 模块下的启动类，启动 nacos 的服务端，启动时增加启动参数，指定启动模式为非集群启动

-Dnacos.standalone=true

5、拜访本地的 nacos：http://localhost:8848/nacos

至此，源码环境搭建胜利！

nacos 客户端

首先要搞清楚：nacos 的客户端其实在咱们本人的服务里，咱们引入了 nacos 的相干坐标依赖，nacos 客户端以 jar 包的模式在咱们的服务中工作

对于 nacos 的客户端，它都要帮咱们实现什么性能呢？

它的次要工作有：服务注册、服务发现、服务下线操作、服务订阅操作等相干操作。

客户端与注册核心服务端的交互，次要集中在服务注册、服务下线、服务发现、订阅某个服务，其实应用最多的就是服务注册和服务发现，上面我会从源码的角度剖析一下这四个性能。

在 Nacos 源码中 nacos-example 中com.alibaba.nacos.example.NamingExample类别离演示了这 4 个性能的操作，咱们能够把它当做入口，代码如下：

public class NamingExample {public static void main(String[] args) throws NacosException {Properties properties = new Properties();
        properties.setProperty("serverAddr", System.getProperty("serverAddr"));
        properties.setProperty("namespace", System.getProperty("namespace"));

        NamingService naming = NamingFactory.createNamingService(properties);
        // 服务注册
        naming.registerInstance("nacos.test.3", "11.11.11.11", 8888, "TEST1");
        naming.registerInstance("nacos.test.3", "2.2.2.2", 9999, "DEFAULT");
        // 服务发现
        System.out.println(naming.getAllInstances("nacos.test.3"));
        // 服务下线
        naming.deregisterInstance("nacos.test.3", "2.2.2.2", 9999, "DEFAULT");
        System.out.println(naming.getAllInstances("nacos.test.3"));
        // 服务订阅
        Executor executor = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(),
                new ThreadFactory() {
                    @Override
                    public Thread newThread(Runnable r) {Thread thread = new Thread(r);
                        thread.setName("test-thread");
                        return thread;
                    }
                });

        naming.subscribe("nacos.test.3", new AbstractEventListener() {

            //EventListener onEvent is sync to handle, If process too low in onEvent, maybe block other onEvent callback.
            //So you can override getExecutor() to async handle event.
            @Override
            public Executor getExecutor() {return executor;}

            @Override
            public void onEvent(Event event) {System.out.println(((NamingEvent) event).getServiceName());
                System.out.println(((NamingEvent) event).getInstances());
            }
        });
    }
}

1.1.1 服务注册

咱们沿着案例中的服务注册办法调用找到 nacos-api 中的 NamingService.registerInstance() 并找到它的实现类和办法com.alibaba.nacos.client.naming.NacosNamingService，代码如下：

/***
 * 服务注册
 * @param serviceName 服务名字
 * @param ip          服务 IP
 * @param port        服务端口
 * @param clusterName 集群名字
 * @throws NacosException
 */
@Override
public void registerInstance(String serviceName, String ip, int port, String clusterName) throws NacosException {registerInstance(serviceName, Constants.DEFAULT_GROUP, ip, port, clusterName);
}

@Override
public void registerInstance(String serviceName, String groupName, String ip, int port, String clusterName)
        throws NacosException {
    // 设置实例 IP：Port，默认权重为 1.0
    Instance instance = new Instance();
    instance.setIp(ip);
    instance.setPort(port);
    instance.setWeight(1.0);
    instance.setClusterName(clusterName);
    // 注册实例
    registerInstance(serviceName, groupName, instance);
}

@Override
public void registerInstance(String serviceName, Instance instance) throws NacosException {registerInstance(serviceName, Constants.DEFAULT_GROUP, instance);
}

/***
 * 实例注册
 * @param serviceName name of service
 * @param groupName   group of service
 * @param instance    instance to register
 * @throws NacosException
 */
@Override
public void registerInstance(String serviceName, String groupName, Instance instance) throws NacosException {NamingUtils.checkInstanceIsLegal(instance);
    String groupedServiceName = NamingUtils.getGroupedName(serviceName, groupName);
    // 该字段示意注册的实例是否是长期实例还是长久化实例。// 如果是长期实例，则不会在 Nacos 服务端长久化存储，须要通过上报心跳的形式进行包活，// 如果一段时间内没有上报心跳，则会被 Nacos 服务端摘除。if (instance.isEphemeral()) {
        // 为注册服务设置一个定时工作获取心跳信息，默认为 5s 汇报一次
        BeatInfo beatInfo = beatReactor.buildBeatInfo(groupedServiceName, instance);
        beatReactor.addBeatInfo(groupedServiceName, beatInfo);
    }
    // 注册到服务端
    serverProxy.registerService(groupedServiceName, groupName, instance);
}

注册次要做了两件事，第一件事：为注册的服务设置一个定时心跳工作。第二件事：将服务注册到服务端。

1: 启动一个定时心跳工作，工夫距离为 5s, 如果服务失常，不做解决，如果不失常，从新注册

2: 发送 http 申请给注册核心服务端，调用服务注册接口，注册服务

下面代码咱们能够看到定时工作增加，但并未齐全看到近程申请，serverProxy.registerService()办法如下，会先封装申请参数，接下来调用 reqApi() 而reqApi()最初会调用callServer()，代码如下：

public void registerService(String serviceName, String groupName, Instance instance) throws NacosException {NAMING_LOGGER.info("[REGISTER-SERVICE] {} registering service {} with instance: {}", namespaceId, serviceName,
            instance);
    // 封装 Http 申请参数
    final Map<String, String> params = new HashMap<String, String>(16);
    params.put(CommonParams.NAMESPACE_ID, namespaceId);
    params.put(CommonParams.SERVICE_NAME, serviceName);
    params.put(CommonParams.GROUP_NAME, groupName);
    params.put(CommonParams.CLUSTER_NAME, instance.getClusterName());
    params.put("ip", instance.getIp());
    params.put("port", String.valueOf(instance.getPort()));
    params.put("weight", String.valueOf(instance.getWeight()));
    params.put("enable", String.valueOf(instance.isEnabled()));
    params.put("healthy", String.valueOf(instance.isHealthy()));
    params.put("ephemeral", String.valueOf(instance.isEphemeral()));
    params.put("metadata", JacksonUtils.toJson(instance.getMetadata()));
    // 执行 Http 申请
    reqApi(UtilAndComs.nacosUrlInstance, params, HttpMethod.POST);
}


/***
* 执行近程调用
**/
public String callServer(String api, Map<String, String> params, Map<String, String> body, String curServer,
        String method) throws NacosException {long start = System.currentTimeMillis();
    long end = 0;
    injectSecurityInfo(params);
    // 封装申请头部
    Header header = builderHeader();
    
    // 申请是 Http 还是 Https 协定
    String url;
    if (curServer.startsWith(UtilAndComs.HTTPS) || curServer.startsWith(UtilAndComs.HTTP)) {url = curServer + api;} else {if (!IPUtil.containsPort(curServer)) {curServer = curServer + IPUtil.IP_PORT_SPLITER + serverPort;}
        url = NamingHttpClientManager.getInstance().getPrefix() + curServer + api;
    }

    try {
        // 执行近程申请, 并获取后果集
        HttpRestResult<String> restResult = nacosRestTemplate
                .exchangeForm(url, header, Query.newInstance().initParams(params), body, method, String.class);
        end = System.currentTimeMillis();
        
        MetricsMonitor.getNamingRequestMonitor(method, url, String.valueOf(restResult.getCode()))
                .observe(end - start);
        // 后果集解析
        if (restResult.ok()) {return restResult.getData();
        }
        if (HttpStatus.SC_NOT_MODIFIED == restResult.getCode()) {return StringUtils.EMPTY;}
        throw new NacosException(restResult.getCode(), restResult.getMessage());
    } catch (Exception e) {NAMING_LOGGER.error("[NA] failed to request", e);
        throw new NacosException(NacosException.SERVER_ERROR, e);
    }
}

执行近程 Http 申请的对象是NacosRestTemplate，该对象就是封装了一般的 Http 申请，大家能够本人查阅一下。

1.1.2 服务发现

两个入口：

1、NamingService.getAllInstances(serviceName)

咱们沿着案例中的服务发现办法调用找到 nacos-api 中的 NamingService.getAllInstances() 并找到它的实现类和办法com.alibaba.nacos.client.naming.NacosNamingService.getAllInstances()，代码如下：

@Override
public List<Instance> getAllInstances(String serviceName, String groupName, List<String> clusters,
        boolean subscribe) throws NacosException {

    ServiceInfo serviceInfo;
    /* 默认 true-> 获取服务实例 */
    if (subscribe) {
        // 从本地缓存中获取，如果本地缓存不存在从服务端拉取
        // 本地缓存会存储在 HostReactor.serviceInfoMap 中，它是一个 Map 对象
        serviceInfo = hostReactor.getServiceInfo(NamingUtils.getGroupedName(serviceName, groupName),
                StringUtils.join(clusters, ","));
    } else {
        serviceInfo = hostReactor
                .getServiceInfoDirectlyFromServer(NamingUtils.getGroupedName(serviceName, groupName),
                        StringUtils.join(clusters, ","));
    }
    List<Instance> list;
    if (serviceInfo == null || CollectionUtils.isEmpty(list = serviceInfo.getHosts())) {return new ArrayList<Instance>();
    }
    return list;
}

下面的代码调用了 hostReactor.getServiceInfo() 办法，该办法会先调用 getServiceInfo0() 办法从本地缓存获取数据，缓存没有数据，就构建实例更新到 Nacos，并从 Nacos 中获取最新数据，getServiceInfo0()办法源码如下：

public ServiceInfo getServiceInfo(final String serviceName, final String clusters) {NAMING_LOGGER.debug("failover-mode:" + failoverReactor.isFailoverSwitch());
    String key = ServiceInfo.getKey(serviceName, clusters);
    if (failoverReactor.isFailoverSwitch()) {return failoverReactor.getService(key);
    }
    /*1。先从本地缓存中获取服务对象，因为启动是第一次进来，所以缓存暂不存在 */
    ServiceInfo serviceObj = getServiceInfo0(serviceName, clusters);

    if (null == serviceObj) {
        /* 构建服务实例 */
        serviceObj = new ServiceInfo(serviceName, clusters);
        /* 将服务实例寄存到缓存中 */
        serviceInfoMap.put(serviceObj.getKey(), serviceObj);
        /* 更新 nacos- 上的服务 */
        updatingMap.put(serviceName, new Object());
        /* 被动获取，并且更新到缓存本地，以及已过期的服务更新等 */
        updateServiceNow(serviceName, clusters);
        updatingMap.remove(serviceName);

    } else if (updatingMap.containsKey(serviceName)) {if (UPDATE_HOLD_INTERVAL > 0) {
            // hold a moment waiting for update finish
            synchronized (serviceObj) {
                try {serviceObj.wait(UPDATE_HOLD_INTERVAL);
                } catch (InterruptedException e) {
                    NAMING_LOGGER
                            .error("[getServiceInfo] serviceName:" + serviceName + ", clusters:" + clusters, e);
                }
            }
        }
    }
    /*2. 开启定时工作 */
    scheduleUpdateIfAbsent(serviceName, clusters);
    return serviceInfoMap.get(serviceObj.getKey());
}

updateServiceNow(serviceName, clusters);主从从近程服务器获取更新数据，最终会调用 updateService() 办法，在该办法中实现近程申请和数据处理，源码如下：

public void updateService(String serviceName, String clusters) throws NacosException {
    /* 获取本地缓存列表中所存在的服务 */
    ServiceInfo oldService = getServiceInfo0(serviceName, clusters);
    try {
        /* 获取服务以及提供者端口信息，端口等 */
        String result = serverProxy.queryList(serviceName, clusters, pushReceiver.getUdpPort(), false);

        if (StringUtils.isNotEmpty(result)) {
            /* 对后果进行解决 */
            processServiceJson(result);
        }
    } finally {if (oldService != null) {synchronized (oldService) {oldService.notifyAll();
            }
        }
    }
}

1.1.3 服务下线

咱们沿着案例中的服务下线办法调用找到 nacos-api 中的 NamingService.deregisterInstance() 并找到它的实现类和办法NacosNamingService.deregisterInstance()，代码如下：

@Override
public void deregisterInstance(String serviceName, String groupName, String ip, int port, String clusterName)
        throws NacosException {
    // 构建实例信息
    Instance instance = new Instance();
    instance.setIp(ip);
    instance.setPort(port);
    instance.setClusterName(clusterName);
    // 服务下线操作
    deregisterInstance(serviceName, groupName, instance);
}

@Override
public void deregisterInstance(String serviceName, String groupName, Instance instance) throws NacosException {if (instance.isEphemeral()) {
        // 移除心跳信息监测的定时工作
        beatReactor.removeBeatInfo(NamingUtils.getGroupedName(serviceName, groupName), instance.getIp(),
                instance.getPort());
    }
    // 发送近程申请执行服务下线销毁操作
    serverProxy.deregisterService(NamingUtils.getGroupedName(serviceName, groupName), instance);
}

服务下线办法比较简单，和服务注册做的事件正好相同，也做了两件事，第一件事：不在进行心跳检测。第二件事：申请服务端服务下线接口。

1.1.4 服务订阅

咱们能够查看订阅服务的案例，会先创立一个线程池，接下来会把线程池封装到监听器中，而监听器中能够监听指定实例信息，代码如下：

// 服务订阅
Executor executor = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(),
        new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {Thread thread = new Thread(r);
                thread.setName("test-thread");
                return thread;
            }
        });

naming.subscribe("nacos.test.3", new AbstractEventListener() {

    //EventListener onEvent is sync to handle, If process too low in onEvent, maybe block other onEvent callback.
    //So you can override getExecutor() to async handle event.
    @Override
    public Executor getExecutor() {return executor;}
    // 读取监听到的服务实例
    @Override
    public void onEvent(Event event) {System.out.println(((NamingEvent) event).getServiceName());
        System.out.println(((NamingEvent) event).getInstances());
    }
});

咱们沿着案例中的服务订阅办法调用找到 nacos-api 中的 NamingService.subscribe() 并找到它的实现类和办法NacosNamingService.deregisterInstance()，代码如下：

public void subscribe(String serviceName, String clusters, EventListener eventListener) {
    // 注册监听
    notifier.registerListener(serviceName, clusters, eventListener);
    // 获取并更新服务实例
    getServiceInfo(serviceName, clusters);
}

此时会注册监听，注册监听，注册监听就是将以后的监听对象信息注入到 listenerMap 汇合中，在监听对象的指定办法 onEvent 中能够读取实例信息，代码如下：

public void registerListener(String serviceName, String clusters, EventListener listener) {String key = ServiceInfo.getKey(serviceName, clusters);
    ConcurrentHashSet<EventListener> eventListeners = listenerMap.get(key);
    if (eventListeners == null) {synchronized (lock) {eventListeners = listenerMap.get(key);
            if (eventListeners == null) {eventListeners = new ConcurrentHashSet<EventListener>();
                listenerMap.put(key, eventListeners);
            }
        }
    }
    // 将以后监听对象放入到汇合中，在监听对象的 onEvent 中能够读出对应的实例对象
    eventListeners.add(listener);
}

本文由传智教育博学谷 – 狂野架构师教研团队公布，转载请注明出处！

如果本文对您有帮忙，欢送关注和点赞；如果您有任何倡议也可留言评论或私信，您的反对是我保持创作的能源

退出移动版