前言：nacos 玩过微服务的想必不会陌生，它是阿里对于springcloud孵化出来的产品，用来完成服务之间的注册发现和配置中心，其核心作用我就不废话了

大致流程：每个服务都会有一个nacos client，它用来和nacos server打交道 用来具体的服务注册 查询等操作，服务提供者在启动的时候会向nacos server注册自己，服务消费者在启动的时候订阅nacos server上的服务提供者

服务注册

首先需要引入spring-cloud-starter-alibaba-nacos-discovery包，本文的引入的版本是2.2.1

<dependency>

    <groupId>com.alibaba.cloud</groupId>

    <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>

    <version>2.2.1.RELEASE</version>

</dependency>

根据sprin.factories配置来完成相关类的自动注册

我们重点来看这几个类，看名称可猜到是用来服务注册的，NacosServiceRegistryAutoConfiguration用来注册管理这几个bean

NacosServiceRegistry：完成服务注册，实现ServiceRegistry

NacosRegistration：用来注册时存储nacos服务端的相关信息

NacosAutoServiceRegistration 继承spring中的AbstractAutoServiceRegistration，AbstractAutoServiceRegistration继承ApplicationListener<WebServerInitializedEvent>，通过事件监听来发起服务注册，到时候会调用NacosServiceRegistry.register(registration)

来看具体如何注册

/******************************************************NacosServiceRegistry******************************************************/

public void register(Registration registration) {

    if (StringUtils.isEmpty(registration.getServiceId())) {

        log.warn("No service to register for nacos client...");

    } else {

        String serviceId = registration.getServiceId();

        String group = this.nacosDiscoveryProperties.getGroup();

        Instance instance = this.getNacosInstanceFromRegistration(registration);

        try {

            this.namingService.registerInstance(serviceId, group, instance);

        }

    }

}

/******************************************************NacosNamingService******************************************************/

public void registerInstance(String serviceName, String groupName, Instance instance) throws NacosException {

    if (instance.isEphemeral()) {

        // 添加心跳检测

        beatReactor.addBeatInfo(NamingUtils.getGroupedName(serviceName, groupName), beatInfo);

    }

    // 完成服务注册

    serverProxy.registerService(NamingUtils.getGroupedName(serviceName, groupName), groupName, instance);

}

/******************************************************NacosNamingService******************************************************/

public void addBeatInfo(String serviceName, BeatInfo beatInfo) {

    String key = buildKey(serviceName, beatInfo.getIp(), beatInfo.getPort());

    // 发起一个心跳检测任务

    executorService.schedule(new BeatTask(beatInfo), beatInfo.getPeriod(), TimeUnit.MILLISECONDS);

    MetricsMonitor.getDom2BeatSizeMonitor().set(dom2Beat.size());

}

/******************************************************BeatTask******************************************************/

class BeatTask implements Runnable {

    @Override

    public void run() {

        if (beatInfo.isStopped()) {

            return;

        }

        long nextTime = beatInfo.getPeriod();

        try {

            // 向nacos服务发起心跳检测

            JSONObject result = serverProxy.sendBeat(beatInfo, BeatReactor.this.lightBeatEnabled);

            long interval = result.getIntValue("clientBeatInterval");

            boolean lightBeatEnabled = false;

            if (result.containsKey(CommonParams.LIGHT_BEAT_ENABLED)) {

                lightBeatEnabled = result.getBooleanValue(CommonParams.LIGHT_BEAT_ENABLED);

            }

            BeatReactor.this.lightBeatEnabled = lightBeatEnabled;

            if (interval > 0) {

                nextTime = interval;

            }

            int code = NamingResponseCode.OK;

            if (result.containsKey(CommonParams.CODE)) {

                code = result.getIntValue(CommonParams.CODE);

            }

            if (code == NamingResponseCode.RESOURCE_NOT_FOUND) {

                // 未注册 先完成注册

                try {

                    serverProxy.registerService(beatInfo.getServiceName(),

                        NamingUtils.getGroupName(beatInfo.getServiceName()), instance);

                }

            }

        }

        // 发起下一次心跳检测

        executorService.schedule(new BeatTask(beatInfo), nextTime, TimeUnit.MILLISECONDS);

    }

}

服务提供者向nacos server发起服务注册前，先向nacos server建立起心跳检测机制，nacos server那边也有一个心跳检测，服务提供者不停的向nacos server发起心跳检测 告知自己的健康状态，nacos serve发现该服务心跳检测时间超时会发布超时事件来告知服务消费者

服务发现

服务发现由NacosWatch完成，它实现了Spring的Lifecycle接口，容器启动和销毁时会调用对应的start（）和stop（）方法

来看对应源码

@Override

public void start() {

    // cas设置运行状态为true

    if (this.running.compareAndSet(false, true)) {

        // 延时执行一个服务发现任务

        this.watchFuture = this.taskScheduler.scheduleWithFixedDelay(

                this::nacosServicesWatch, this.properties.getWatchDelay());

    }

}

@Override

public void stop() {

    // 设置运行状态为false 然后取消正在执行的任务

    if (this.running.compareAndSet(true, false) && this.watchFuture != null) {

        this.watchFuture.cancel(true);

    }

}

public void nacosServicesWatch() {

    try {

        boolean changed = false;

        NamingService namingService = properties.namingServiceInstance();

        // 获取nacos server上最新的服务提供者们

        ListView<String> listView = properties.namingServiceInstance()

                .getServicesOfServer(1, Integer.MAX_VALUE);

        List<String> serviceList = listView.getData();

        // 有新的订阅产生 订阅完后发布事件

        Set<String> currentServices = new HashSet<>(serviceList);

        currentServices.removeAll(cacheServices);

        if (currentServices.size() > 0) {

            changed = true;

        }

        // 取消已经下线的服务订阅，发起取消订阅操作并删除订阅监听

        if (cacheServices.removeAll(new HashSet<>(serviceList))

                && cacheServices.size() > 0) {

            changed = true;

            for (String serviceName : cacheServices) {

                namingService.unsubscribe(serviceName,

                        subscribeListeners.get(serviceName));

                subscribeListeners.remove(serviceName);

            }

        }

        cacheServices = new HashSet<>(serviceList);

        // 订阅服务 并对每个服务都添加一个心跳检测监听

        for (String serviceName : cacheServices) {

            if (!subscribeListeners.containsKey(serviceName)) {

                EventListener eventListener = event -> NacosWatch.this.publisher

                        .publishEvent(new HeartbeatEvent(NacosWatch.this,

                                nacosWatchIndex.getAndIncrement()));

                subscribeListeners.put(serviceName, eventListener);

                namingService.subscribe(serviceName, eventListener);

            }

        }

        // 有服务变化 发布事件

        if (changed) {

            this.publisher.publishEvent(

                    new HeartbeatEvent(this, nacosWatchIndex.getAndIncrement()));

        }

    }

    catch (Exception e) {

        log.error("Error watching Nacos Service change", e);

    }

}

大致流程：nacos client这边在spring容器启动后执行一个服务订阅操作的延时任务，这个任务执行时先拉取nacos server那边最新的服务列表，然后与本地缓存的服务列表进行比较，取消订阅下线的服务，然后向nacos server发起订阅操作，订阅所有服务

那么服务消费者如何实时感知服务提供者的状态信息呢

1、服务消费者订阅后会执行一个轮询任务（每10s执行一次）用来拉取最新的服务提供者信息并实时更新,实现在HostReactor中的UpdateTask完成，下面来看代码

public class UpdateTask implements Runnable {

    long lastRefTime = Long.MAX_VALUE;

    private String clusters;

    private String serviceName;

    public UpdateTask(String serviceName, String clusters) {

        this.serviceName = serviceName;

        this.clusters = clusters;

    }

    @Override

    public void run() {

        try {

            // 拿到当前的服务信息

            ServiceInfo serviceObj = serviceInfoMap.get(ServiceInfo.getKey(serviceName, clusters));

            // 为空 拉取最新的服务列表随后更新

            if (serviceObj == null) {

                updateServiceNow(serviceName, clusters);

                // 继续轮询

                executor.schedule(this, DEFAULT_DELAY, TimeUnit.MILLISECONDS);

                return;

            }

            if (serviceObj.getLastRefTime() <= lastRefTime) {

                // 当前服务未及时更新 进行更新操作

                updateServiceNow(serviceName, clusters);

                serviceObj = serviceInfoMap.get(ServiceInfo.getKey(serviceName, clusters));

            } else {

                // if serviceName already updated by push, we should not override it

                // since the push data may be different from pull through force push

                refreshOnly(serviceName, clusters);

            }

            // 设置服务最新的更新时间

            lastRefTime = serviceObj.getLastRefTime();

            // 订阅被取消

            if (!eventDispatcher.isSubscribed(serviceName, clusters) &&

                !futureMap.containsKey(ServiceInfo.getKey(serviceName, clusters))) {

                // abort the update task:

                NAMING_LOGGER.info("update task is stopped, service:" + serviceName + ", clusters:" + clusters);

                return;

            }

            // 继续下一次轮询

            executor.schedule(this, serviceObj.getCacheMillis(), TimeUnit.MILLISECONDS);

        } catch (Throwable e) {

            NAMING_LOGGER.warn("[NA] failed to update serviceName: " + serviceName, e);

        }

    }

}

2、上面服务注册时我们说过，服务提供者注册时nacos服务端也有一个相应的心跳检测，当心跳检测超时也就是未及时收到服务提供者的心跳包，nacos server判定该服务状态异常 随后通过UDP推送服务信息用来告知对应服务消费者，服务消费者通过PushReceiver来处理udp协议，HostReactor.processServiceJson(String json)来更新本地服务列表

/********************************PushReceiver*****************************/

public void run() {

    while (true) {

        try {

            // byte[] is initialized with 0 full filled by default

            byte[] buffer = new byte[UDP_MSS];

            DatagramPacket packet = new DatagramPacket(buffer, buffer.length);

            udpSocket.receive(packet);

            String json = new String(IoUtils.tryDecompress(packet.getData()), "UTF-8").trim();

            NAMING_LOGGER.info("received push data: " + json + " from " + packet.getAddress().toString());

            PushPacket pushPacket = JSON.parseObject(json, PushPacket.class);

            String ack;

            if ("dom".equals(pushPacket.type) || "service".equals(pushPacket.type)) {

                // 处理变更信息

                hostReactor.processServiceJSON(pushPacket.data);

                // send ack to server

                ack = "{\"type\": \"push-ack\""

                    + ", \"lastRefTime\":\"" + pushPacket.lastRefTime

                    + "\", \"data\":" + "\"\"}";

            } else if ("dump".equals(pushPacket.type)) {

                // dump data to server

                ack = "{\"type\": \"dump-ack\""

                    + ", \"lastRefTime\": \"" + pushPacket.lastRefTime

                    + "\", \"data\":" + "\""

                    + StringUtils.escapeJavaScript(JSON.toJSONString(hostReactor.getServiceInfoMap()))

                    + "\"}";

            } else {

                // do nothing send ack only

                ack = "{\"type\": \"unknown-ack\""

                    + ", \"lastRefTime\":\"" + pushPacket.lastRefTime

                    + "\", \"data\":" + "\"\"}";

            }

            udpSocket.send(new DatagramPacket(ack.getBytes(Charset.forName("UTF-8")),

                ack.getBytes(Charset.forName("UTF-8")).length, packet.getSocketAddress()));

        } catch (Exception e) {

            NAMING_LOGGER.error("[NA] error while receiving push data", e);

        }

    }

}

服务注册和订阅我只讲解了主要流程，nacos server那边处理源码太多就不一一贴出来了，根据对应的api接口进去一看便知，nacos源码比较好理解，没有什么特别难读懂的地方，这边只是提供给大家一个看源码的思路，具体详细流程还需要读者自己去细读

下面通过代码来模拟nacos服务注册和订阅

先启动一个nacos server，然后打开控制台，添加一个命名空间

服务注册:分别注册两个服务，其中一个服务有两个实例

public class ServerRegister {

    public static void main(String[] args) throws Exception {

        Properties properties = new Properties();

        properties.setProperty("serverAddr", "http://localhost:8848");

        properties.setProperty("namespace", "c7981cfd-ccb8-4a9f-8e80-cd1f9633ecec");

        NamingService namingService = NacosFactory.createNamingService(properties);

        // 同一个服务注册两个实例

        namingService.registerInstance("serverProvider_1", "127.0.0.1", 8080);

        namingService.registerInstance("serverProvider_1", "127.0.0.1", 8081);

        namingService.registerInstance("serverProvider_2", "127.0.0.1", 7070);
　　　　　// 获取服务名为serverProvider_1的实例信息

        List<Instance> serverProvider = namingService.getAllInstances("serverProvider_1");

        System.out.println(JSONArray.toJSONString(serverProvider, SerializerFeature.PrettyFormat, SerializerFeature.WriteMapNullValue,

                SerializerFeature.WriteDateUseDateFormat));

        System.in.read();

    }

}

服务订阅：获取所有的服务提供者，然后进行订阅 并添加一个事件用来监听订阅成功后的实例

public class ServerCustomer {

    public static void main(String[] args) throws Exception {

        Properties properties = new Properties();

        properties.setProperty("serverAddr", "http://localhost:8848");

        properties.setProperty("namespace", "c7981cfd-ccb8-4a9f-8e80-cd1f9633ecec");

        NamingService namingService = NacosFactory.createNamingService(properties);

        List<String> serverList = namingService.getServicesOfServer(1, Integer.MAX_VALUE).getData();

        System.out.println("得到服务提供者列表:" + JSONArray.toJSONString(serverList));

        for (String server : serverList) {

            // 订阅serverProvider服务 并添加一个监听器用来监听服务状态

            namingService.subscribe(server, event -> {

                NamingEvent namingEvent = (NamingEvent) event;

                System.out.println(JSONObject.toJSONString(namingEvent, SerializerFeature.PrettyFormat, SerializerFeature.WriteMapNullValue,

                        SerializerFeature.WriteDateUseDateFormat));

            });

        }

        System.in.read();

    }

}

打印信息：

nacos控制台也能看到相应服务信息

随后把服务提供者下线 再来看服务消费者那边的输出，可以看到服务实例已全部下线

nacos控制台

最新文章

  1. android Sqlite select * from myDatabase没有内容的问题
  2. 关于flume配置加载（二）
  3. 完成卸载vs2010后再安装
  4. jdk动态代理学习
  5. Codeforces Round #261 (Div. 2)
  6. 从零教你如何获取hadoop2.4源码并使用eclipse关联hadoop2.4源码
  7. android重要的对象
  8. [置顶] Firefox OS 学习——简单了解知识
  9. HDU 4513 哥几个系列故事——形成完善II manacher求最长回文
  10. Jenkins 发布后自动创建git tag
  11. Java 第六周总结
  12. Python/ MySQL练习题（一）
  13. BZOJ_4004_[JLOI2015]装备购买_线性基
  14. IO多路复用注解
  15. [MicroPython]TPYBoardv102超全DIY案例一览
  16. windows10不能获取有效IP的问题
  17. struts2：标签库图示，控制标签
  18. 树形DP 复习
  19. H2内嵌数据库使用步骤
  20. python强大的数据类型转换

热门文章

  1. 派大星的烦恼MISC
  2. leetcode——（四）2020.06.08
  3. 《Eroico》关卡与操作设计
  4. 从HBase底层原理解析HBASE列族不能设计太多的原因？
  5. uni-app中组件的使用
  6. STL——容器（deque） deque 的赋值 assign() operator=() swap()
  7. 【GIT】随笔
  8. 纯HTML + CSS制作个人资料卡
  9. oracle ADG启动顺序
  10. JavaSE13-常用API&amp;异常