一.NacosPropertySourceLocator

  1. 当前类实现了PropertySourceLocator，主要用于配置文件的拓展，从配置中心读取配置
  1. private void loadApplicationConfiguration(CompositePropertySource compositePropertySource, String dataIdPrefix, NacosConfigProperties properties, Environment environment) {
  2.         String fileExtension = properties.getFileExtension();
  3.         String nacosGroup = properties.getGroup();
  4.         //1.1读取配置
  5.         this.loadNacosDataIfPresent(compositePropertySource, dataIdPrefix, nacosGroup, fileExtension, true);
  6.         this.loadNacosDataIfPresent(compositePropertySource, dataIdPrefix + "." + fileExtension, nacosGroup, fileExtension, true);
  7.         String[] var7 = environment.getActiveProfiles();
  8.         int var8 = var7.length;
  10.         for(int var9 = 0; var9 < var8; ++var9) {
  11.             String profile = var7[var9];
  12.             String dataId = dataIdPrefix + "-" + profile + "." + fileExtension;
  13.             this.loadNacosDataIfPresent(compositePropertySource, dataId, nacosGroup, fileExtension, true);
  14.         }
  16.     }

二.RpcClient

  1. 在获取客户端配置的时候会启动Rpc客户端
  1. public final void start() throws NacosException {
  2.         //将当前客户端跟改为启动状态
  3.         boolean success = this.rpcClientStatus.compareAndSet(RpcClientStatus.INITIALIZED, RpcClientStatus.STARTING);
  4.         if (success) {
  5.             //创建定时任务线程池（包含两个守护线程）
  6.             this.clientEventExecutor = new ScheduledThreadPoolExecutor(2, new ThreadFactory() {
  7.                 public Thread newThread(Runnable r) {
  8.                     Thread t = new Thread(r);
  9.                     t.setName("com.alibaba.nacos.client.remote.worker");
  10.                     t.setDaemon(true);
  11.                     return t;
  12.                 }
  13.             });
  14.             this.clientEventExecutor.submit(new Runnable() {
  15.                 public void run() {
  16.                     while(!RpcClient.this.clientEventExecutor.isTerminated() && !RpcClient.this.clientEventExecutor.isShutdown()) {
  17.                         RpcClient.ConnectionEvent take = null;
  19.                         try {
  20.                             take = (RpcClient.ConnectionEvent)RpcClient.this.eventLinkedBlockingQueue.take();
  21.                             if (take.isConnected()) {
  22.                                 RpcClient.this.notifyConnected();
  23.                             } else if (take.isDisConnected()) {
  24.                                 RpcClient.this.notifyDisConnected();
  25.                             }
  26.                         } catch (Throwable var3) {
  27.                         }
  28.                     }
  30.                 }
  31.             });
  32.             this.clientEventExecutor.submit(new Runnable() {
  33.                 public void run() {
  34.                     while(true) {
  35.                         while(true) {
  36.                             while(true) {
  37.                                 while(true) {
  38.                                     while(true) {
  39.                                         try {
  40.                                             RpcClient.ReconnectContext reconnectContext;
  41.                                             boolean serverExist;
  42.                                             label69: {
  43.                                                 if (!RpcClient.this.isShutdown()) {
  44.                                                     reconnectContext = (RpcClient.ReconnectContext)RpcClient.this.reconnectionSignal.poll(RpcClient.this.keepAliveTime, TimeUnit.MILLISECONDS);
  45.                                                     if (reconnectContext != null) {
  46.                                                         break label69;
  47.                                                     }
  49.                                                     if (System.currentTimeMillis() - RpcClient.this.lastActiveTimeStamp < RpcClient.this.keepAliveTime) {
  50.                                                         continue;
  51.                                                     }
  53.                                                     serverExist = RpcClient.this.healthCheck();
  54.                                                     if (serverExist) {
  55.                                                         RpcClient.this.lastActiveTimeStamp = System.currentTimeMillis();
  56.                                                         continue;
  57.                                                     }
  59.                                                     if (RpcClient.this.currentConnection == null) {
  60.                                                         continue;
  61.                                                     }
  63.                                                     LoggerUtils.printIfInfoEnabled(RpcClient.LOGGER, "[{}]Server healthy check fail,currentConnection={}", new Object[]{RpcClient.this.name, RpcClient.this.currentConnection.getConnectionId()});
  64.                                                     RpcClientStatus rpcClientStatus = (RpcClientStatus)RpcClient.this.rpcClientStatus.get();
  65.                                                     if (!RpcClientStatus.SHUTDOWN.equals(rpcClientStatus)) {
  66.                                                         boolean success = RpcClient.this.rpcClientStatus.compareAndSet(rpcClientStatus, RpcClientStatus.UNHEALTHY);
  67.                                                         if (!success) {
  68.                                                             continue;
  69.                                                         }
  71.                                                         reconnectContext = RpcClient.this.new ReconnectContext((RpcClient.ServerInfo)null, false);
  72.                                                         break label69;
  73.                                                     }
  74.                                                 }
  76.                                                 return;
  77.                                             }
  79.                                             if (reconnectContext.serverInfo != null) {
  80.                                                 serverExist = false;
  81.                                                 Iterator var7 = RpcClient.this.getServerListFactory().getServerList().iterator();
  83.                                                 while(var7.hasNext()) {
  84.                                                     String server = (String)var7.next();
  85.                                                     RpcClient.ServerInfo serverInfo = RpcClient.this.resolveServerInfo(server);
  86.                                                     if (serverInfo.getServerIp().equals(reconnectContext.serverInfo.getServerIp())) {
  87.                                                         serverExist = true;
  88.                                                         reconnectContext.serverInfo.serverPort = serverInfo.serverPort;
  89.                                                         break;
  90.                                                     }
  91.                                                 }
  93.                                                 if (!serverExist) {
  94.                                                     LoggerUtils.printIfInfoEnabled(RpcClient.LOGGER, "[{}] Recommend server is not in server list ,ignore recommend server {}", new Object[]{RpcClient.this.name, reconnectContext.serverInfo.getAddress()});
  95.                                                     reconnectContext.serverInfo = null;
  96.                                                 }
  97.                                             }
  99.                                             RpcClient.this.reconnect(reconnectContext.serverInfo, reconnectContext.onRequestFail);
  100.                                         } catch (Throwable var6) {
  101.                                         }
  102.                                     }
  103.                                 }
  104.                             }
  105.                         }
  106.                     }
  107.                 }
  108.             });
  109.             Connection connectToServer = null;
  110.             this.rpcClientStatus.set(RpcClientStatus.STARTING);
  111.             //失败重试次数
  112.             int startUpRetryTimes = 3;
  114.             while(startUpRetryTimes > 0 && connectToServer == null) {
  115.                 try {
  116.                     --startUpRetryTimes;
  117.                     RpcClient.ServerInfo serverInfo = this.nextRpcServer();
  118.                     LoggerUtils.printIfInfoEnabled(LOGGER, "[{}] Try to connect to server on start up, server: {}", new Object[]{this.name, serverInfo});
  119.                     connectToServer = this.connectToServer(serverInfo);
  120.                 } catch (Throwable var5) {
  121.                     LoggerUtils.printIfWarnEnabled(LOGGER, "[{}]Fail to connect to server on start up, error message={}, start up retry times left: {}", new Object[]{this.name, var5.getMessage(), startUpRetryTimes});
  122.                 }
  123.             }
  125.             if (connectToServer != null) {
  126.                 LoggerUtils.printIfInfoEnabled(LOGGER, "[{}] Success to connect to server [{}] on start up,connectionId={}", new Object[]{this.name, connectToServer.serverInfo.getAddress(), connectToServer.getConnectionId()});
  127.                 this.currentConnection = connectToServer;
  128.                 this.rpcClientStatus.set(RpcClientStatus.RUNNING);
  129.                 this.eventLinkedBlockingQueue.offer(new RpcClient.ConnectionEvent(1));
  130.             } else {
  131.                 this.switchServerAsync();
  132.             }
  134.             this.registerServerRequestHandler(new RpcClient.ConnectResetRequestHandler());
  135.             this.registerServerRequestHandler(new ServerRequestHandler() {
  136.                 public Response requestReply(Request request) {
  137.                     return request instanceof ClientDetectionRequest ? new ClientDetectionResponse() : null;
  138.                 }
  139.             });
  140.         }
  141.     }

GrpcClient

  1. public Connection connectToServer(ServerInfo serverInfo) {
  2.         try {
  3.             int port;
  4.             if (this.grpcExecutor == null) {
  5.                 port = ThreadUtils.getSuitableThreadCount(8);
  6.                 this.grpcExecutor = new ThreadPoolExecutor(port, port, 10L, TimeUnit.SECONDS, new LinkedBlockingQueue(10000), (new ThreadFactoryBuilder()).setDaemon(true).setNameFormat("nacos-grpc-client-executor-%d").build());
  7.                 this.grpcExecutor.allowCoreThreadTimeOut(true);
  8.             }
  10.             port = serverInfo.getServerPort() + this.rpcPortOffset();
  11.             RequestFutureStub newChannelStubTemp = this.createNewChannelStub(serverInfo.getServerIp(), port);
  12.             if (newChannelStubTemp != null) {
  13.                 Response response = this.serverCheck(serverInfo.getServerIp(), port, newChannelStubTemp);
  14.                 if (response != null && response instanceof ServerCheckResponse) {
  15.                     BiRequestStreamStub biRequestStreamStub = BiRequestStreamGrpc.newStub(newChannelStubTemp.getChannel());
  16.                     GrpcConnection grpcConn = new GrpcConnection(serverInfo, this.grpcExecutor);
  17.                     grpcConn.setConnectionId(((ServerCheckResponse)response).getConnectionId());
  18.                     StreamObserver<Payload> payloadStreamObserver = this.bindRequestStream(biRequestStreamStub, grpcConn);
  19.                     grpcConn.setPayloadStreamObserver(payloadStreamObserver);
  20.                     grpcConn.setGrpcFutureServiceStub(newChannelStubTemp);
  21.                     grpcConn.setChannel((ManagedChannel)newChannelStubTemp.getChannel());
  22.                     ConnectionSetupRequest conSetupRequest = new ConnectionSetupRequest();
  23.                     conSetupRequest.setClientVersion(VersionUtils.getFullClientVersion());
  24.                     conSetupRequest.setLabels(super.getLabels());
  25.                     conSetupRequest.setAbilities(super.clientAbilities);
  26.                     conSetupRequest.setTenant(super.getTenant());
  27.                     grpcConn.sendRequest(conSetupRequest);
  28.                     Thread.sleep(100L);
  29.                     return grpcConn;
  30.                 } else {
  31.                     this.shuntDownChannel((ManagedChannel)newChannelStubTemp.getChannel());
  32.                     return null;
  33.                 }
  34.             } else {
  35.                 return null;
  36.             }
  37.         } catch (Exception var9) {
  38.             LOGGER.error("[{}]Fail to connect to server!,error={}", this.getName(), var9);
  39.             return null;
  40.         }
  41.     }

NacosWatch

  1. SmartLifecycle 接口。当Spring容器加载所有bean并完成初始化之后，会接着回调实现该接口的类中对应的方法
  2. NacosWatch实现了该接口，主要作用试发布事件

NacosServiceRegistry

  1. #spring容器完成之后会发布事件,当前这个接口实现了服务注册接口AbstractAutoServiceRegistration
  2. public class NacosServiceRegistry implements ServiceRegistry<Registration> {
  3.     private static final Logger log = LoggerFactory.getLogger(NacosServiceRegistry.class);
  4.     private final NacosDiscoveryProperties nacosDiscoveryProperties;
  5.     private final NamingService namingService;
  7.     public NacosServiceRegistry(NacosDiscoveryProperties nacosDiscoveryProperties) {
  8.         this.nacosDiscoveryProperties = nacosDiscoveryProperties;
  9.         this.namingService = nacosDiscoveryProperties.namingServiceInstance();
  10.     }
  11.     #Registration在实例化的时候会把ip超时时间，心跳超时时间等一些数据放入metadata中
  12.     public void register(Registration registration) {
  13.         if (StringUtils.isEmpty(registration.getServiceId())) {
  14.             log.warn("No service to register for nacos client...");
  15.         } else {
  16.             String serviceId = registration.getServiceId();
  17.             Instance instance = this.getNacosInstanceFromRegistration(registration);
  19.             try {
  20.                 this.namingService.registerInstance(serviceId, instance);
  21.                 log.info("nacos registry, {} {}:{} register finished", new Object[]{serviceId, instance.getIp(), instance.getPort()});
  22.             } catch (Exception var5) {
  23.                 log.error("nacos registry, {} register failed...{},", new Object[]{serviceId, registration.toString(), var5});
  24.             }
  26.         }
  27.     }
  28. }

一、创建心跳信息信息，执行心跳定时任务

  1. public void registerInstance(String serviceName, String groupName, Instance instance) throws NacosException {
  2.         if (instance.isEphemeral()) {
  3.             BeatInfo beatInfo = new BeatInfo();
  4.             beatInfo.setServiceName(NamingUtils.getGroupedName(serviceName, groupName));
  5.             beatInfo.setIp(instance.getIp());
  6.             beatInfo.setPort(instance.getPort());
  7.             beatInfo.setCluster(instance.getClusterName());
  8.             beatInfo.setWeight(instance.getWeight());
  9.             beatInfo.setMetadata(instance.getMetadata());
  10.             beatInfo.setScheduled(false);
  11.             long instanceInterval = instance.getInstanceHeartBeatInterval();
  12.             // 设置心跳检测的时间周期，默认是5s
  13.             // 可以通过属性spring.cloud.nacos.discovery.heartBeatInterval配置
  14.             beatInfo.setPeriod(instanceInterval == 0L ? DEFAULT_HEART_BEAT_INTERVAL : instanceInterval);
  15.             //创建心跳定时任务
  16.             this.beatReactor.addBeatInfo(NamingUtils.getGroupedName(serviceName, groupName), beatInfo);
  17.         }
  19.         this.serverProxy.registerService(NamingUtils.getGroupedName(serviceName, groupName), groupName, instance);
  20.     }

Nacos源码2.03二、服务注册 - 图1
1.1创建心跳定时任务（默认为5秒）

  1. public void addBeatInfo(String serviceName, BeatInfo beatInfo) {
  2.         LogUtils.NAMING_LOGGER.info("[BEAT] adding beat: {} to beat map.", beatInfo);
  3.         //将服务名 ip 端口作为key 心跳信息作为value存入map
  4.         this.dom2Beat.put(this.buildKey(serviceName, beatInfo.getIp(), beatInfo.getPort()), beatInfo);
  5.         //创建定时任务
  6.         this.executorService.schedule(new BeatReactor.BeatTask(beatInfo), 0L, TimeUnit.MILLISECONDS);
  7.         #创建prometheus监视器监控心跳        
  8.         MetricsMonitor.getDom2BeatSizeMonitor().set((double)this.dom2Beat.size());
  9.     }

1.2线程池的创建

  1. #默认取namingClientBeatThreadCount数量如果没有根据下面算法得出
  2. #线程数的定义算法
  3. Runtime.getRuntime().availableProcessors() > 1 ? Runtime.getRuntime().availableProcessors() / 2 : 1
  6. #线程池的创建是在对象创建的构造器注入的,默认为8个（守护线程）
  7. public BeatReactor(NamingProxy serverProxy, int threadCount) {
  8.         this.dom2Beat = new ConcurrentHashMap();
  9.         this.serverProxy = serverProxy;
  10.         this.executorService = new ScheduledThreadPoolExecutor(threadCount, new ThreadFactory() {
  11.             public Thread newThread(Runnable r) {
  12.                 Thread thread = new Thread(r);
  13.                 thread.setDaemon(true);
  14.                 thread.setName("com.alibaba.nacos.naming.beat.sender");
  15.                 return thread;
  16.             }
  17.         });
  18.     }

1.3线程任务的创建

  1. class BeatTask implements Runnable {
  2.         BeatInfo beatInfo;
  4.         public BeatTask(BeatInfo beatInfo) {
  5.             this.beatInfo = beatInfo;
  6.         }
  8.         public void run() {
  9.             if (!this.beatInfo.isStopped()) {
  10.                 //向客户端的心跳接口请求/instance/beat
  11.                 long result = BeatReactor.this.serverProxy.sendBeat(this.beatInfo);
  12.                 //获取下一次的请求时间
  13.                 long nextTime = result > 0L ? result : this.beatInfo.getPeriod();
  14.                 //将任务放入定时任务池中（守护线程）
  15.                 BeatReactor.this.executorService.schedule(BeatReactor.this.new BeatTask(this.beatInfo), nextTime, TimeUnit.MILLISECONDS);
  16.             }
  17.         }
  18.     }

2、创建监视器

  1.  private static Gauge nacosMonitor = (Gauge)((Builder)((Builder)((Builder)Gauge.build().name("nacos_monitor")).labelNames(new String[]{"module", "name"})).help("nacos_monitor")).register();
  3.     //设置监视器
  4.     public static Child getDom2BeatSizeMonitor() {
  5.         return (Child)nacosMonitor.labels(new String[]{"naming", "dom2BeatSize"});
  6.     }

二.注册服务调用

2.1参数封装

  1. public void registerService(String serviceName, String groupName, Instance instance) throws NacosException {
  2.         LogUtils.NAMING_LOGGER.info("[REGISTER-SERVICE] {} registering service {} with instance: {}", new Object[]{this.namespaceId, serviceName, instance});
  3.         Map<String, String> params = new HashMap(9);
  4.         params.put("namespaceId", this.namespaceId);
  5.         params.put("serviceName", serviceName);
  6.         params.put("groupName", groupName);
  7.         params.put("clusterName", instance.getClusterName());
  8.         params.put("ip", instance.getIp());
  9.         params.put("port", String.valueOf(instance.getPort()));
  10.         params.put("weight", String.valueOf(instance.getWeight()));
  11.         params.put("enable", String.valueOf(instance.isEnabled()));
  12.         params.put("healthy", String.valueOf(instance.isHealthy()));
  13.         params.put("ephemeral", String.valueOf(instance.isEphemeral()));
  14.         params.put("metadata", JSON.toJSONString(instance.getMetadata()));
  15.         this.reqAPI(UtilAndComs.NACOS_URL_INSTANCE, params, (String)"POST");
  16.     }

2.2采用故障转移算法，请求调用注册地址/nacos/v1/ns/instance

  1. public String reqAPI(String api, Map<String, String> params, List<String> servers, String method) {
  2.         params.put("namespaceId", this.getNamespaceId());
  3.         if (CollectionUtils.isEmpty(servers) && StringUtils.isEmpty(this.nacosDomain)) {
  4.             throw new IllegalArgumentException("no server available");
  5.         } else {
  6.             Exception exception = new Exception();
  7.             if (servers != null && !servers.isEmpty()) {
  8.                 Random random = new Random(System.currentTimeMillis());//根据当前时间创建随机数
  9.                 int index = random.nextInt(servers.size());//根据服务数量获取随机数
  11.                 for(int i = 0; i < servers.size(); ++i) {
  12.                     String server = (String)servers.get(index);//获取随机的注册地址
  14.                     try {
  15.                         return this.callServer(api, params, server, method);//调用nacos注册中心
  16.                     } catch (NacosException var11) {
  17.                         exception = var11;
  18.                         LogUtils.NAMING_LOGGER.error("request {} failed.", server, var11);
  19.                     } catch (Exception var12) {
  20.                         exception = var12;
  21.                         LogUtils.NAMING_LOGGER.error("request {} failed.", server, var12);
  22.                     }
  24.                     index = (index + 1) % servers.size();//出现异常请求下一个
  25.                 }
  27.                 throw new IllegalStateException("failed to req API:" + api + " after all servers(" + servers + ") tried: " + ((Exception)exception).getMessage());
  28.             } else {
  29.                 int i = 0;
  31.                 while(i < 3) {
  32.                     try {
  33.                         return this.callServer(api, params, this.nacosDomain);
  34.                     } catch (Exception var13) {
  35.                         exception = var13;
  36.                         LogUtils.NAMING_LOGGER.error("[NA] req api:" + api + " failed, server(" + this.nacosDomain, var13);
  37.                         ++i;
  38.                     }
  39.                 }
  41.                 throw new IllegalStateException("failed to req API:/api/" + api + " after all servers(" + servers + ") tried: " + ((Exception)exception).getMessage());
  42.             }
  43.         }
  44.     }
  1. public String callServer(String api, Map<String, String> params, String curServer, String method) throws NacosException {
  2.         long start = System.currentTimeMillis();
  3.         long end = 0L;
  4.         this.checkSignature(params);
  5.         List<String> headers = this.builderHeaders();
  6.         String url;//请求路径为：/nacos/v1/ns/instance
  7.         if (!curServer.startsWith("https://") && !curServer.startsWith("http://")) {
  8.             if (!curServer.contains(":")) {
  9.                 curServer = curServer + ":" + this.serverPort;
  10.             }
  12.             url = HttpClient.getPrefix() + curServer + api;
  13.         } else {
  14.             url = curServer + api;
  15.         }
  17.         HttpResult result = HttpClient.request(url, headers, params, "UTF-8", method);
  18.         end = System.currentTimeMillis();
  19.         MetricsMonitor.getNamingRequestMonitor(method, url, String.valueOf(result.code)).observe((double)(end - start));
  20.         if (200 == result.code) {
  21.             return result.content;
  22.         } else if (304 == result.code) {
  23.             return "";
  24.         } else {
  25.             throw new NacosException(500, "failed to req API:" + curServer + api + ". code:" + result.code + " msg: " + result.content);
  26.         }
  27.     }

3.Nacos Server端接收/nacos/v1/ns/instance请求

  1.     @CanDistro
  2.     @PostMapping
  3.     @Secured(parser = NamingResourceParser.class, action = ActionTypes.WRITE)
  4.     public String register(HttpServletRequest request) throws Exception {
  6.         final String namespaceId = WebUtils//根据namespaceId获取命名空间,如果没有则为默认public
  7.                 .optional(request, CommonParams.NAMESPACE_ID, Constants.DEFAULT_NAMESPACE_ID);
  8.         final String serviceName = WebUtils.required(request, CommonParams.SERVICE_NAME);//根据serviceName获取服务名
  9.         NamingUtils.checkServiceNameFormat(serviceName);//服务名格式为DEFAULT_GROUP@@nacos-test
  11.         final Instance instance = HttpRequestInstanceBuilder.newBuilder()//根据当前请求创建临时实例
  12.                 .setDefaultInstanceEphemeral(switchDomain.isDefaultInstanceEphemeral()).setRequest(request).build();
  14.         getInstanceOperator().registerInstance(namespaceId, serviceName, instance);//注册实例
  15.         return "ok";
  16.     }

这里是客户端请求会进入InstanceOperatorClientImpl

  1.         boolean ephemeral = instance.isEphemeral();//判断是否是临时实例
  2.         String clientId = IpPortBasedClient.getClientId(instance.toInetAddr(), ephemeral);//根据ip获取clientId——》127.0.0.1:8181#true
  3.         createIpPortClientIfAbsent(clientId);//如果容器已有当前ip实例建立连接
  4.         Service service = getService(namespaceId, serviceName, ephemeral);//创建服务实例
  5.         clientOperationService.registerInstance(service, instance, clientId);//注册客户端实例
  1.     @Override
  2.     public void registerInstance(Service service, Instance instance, String clientId) {
  3.         Service singleton = ServiceManager.getInstance().getSingleton(service);//获取服务单实例
  4.         Client client = clientManager.getClient(clientId);
  5.         if (!clientIsLegal(client, clientId)) {//判断连接是否已经断开或者不是临时连接
  6.             return;
  7.         }
  8.         InstancePublishInfo instanceInfo = getPublishInfo(instance);//创建事件实例
  9.         client.addServiceInstance(singleton, instanceInfo);
  10.         client.setLastUpdatedTime();
  11.         NotifyCenter.publishEvent(new ClientOperationEvent.ClientRegisterServiceEvent(singleton, clientId));
  12.         NotifyCenter
  13.                 .publishEvent(new MetadataEvent.InstanceMetadataEvent(singleton, instanceInfo.getMetadataId(), false));//发布事件
  14.     }