微服务组件-----Spring Cloud Alibaba 注册中心 Nacos源码（1.4.x版本）分 ...

 

核心功能点

【1】服务注册：Nacos Client会通过发送REST请求的方式向Nacos Server注册自己的服务，提供自身的元数据，比如ip地址、端口等信息。Nacos Server接收到注册请求后，就会把这些元数据信息存储在一个双层的内存Map中。
【2】服务心跳：在服务注册后，Nacos Client会维护一个定时心跳来持续通知Nacos Server，说明服务一直处于可用状态，防止被剔除。默认5s发送一次心跳。
【3】服务同步：Nacos Server集群之间会互相同步服务实例，用来保证服务信息的一致性。  
【4】服务发现：服务消费者（Nacos Client）在调用服务提供者的服务时，会发送一个REST请求给Nacos Server，获取上面注册的服务清单，并且缓存在Nacos Client本地，同时会在Nacos Client本地开启一个定时任务定时拉取服务端最新的注册表信息更新到本地缓存
【5】服务健康检查：Nacos Server会开启一个定时任务用来检查注册服务实例的健康情况，对于超过15s没有收到客户端心跳的实例会将它的healthy属性置为false(客户端服务发现时不会发现)，如果某个实例超过30秒没有收到心跳，直接剔除该实例(被剔除的实例如果恢复发送心跳则会重新注册)
 

源码精髓总结

【1】注册表的结构说明（这个仅是记录）：

1. //Map<namespaceId, Map<service_name, Service>【ConcurrentSkipListMap】>
2. private final Map<String, Map<String, Service>> serviceMap = new ConcurrentHashMap<>();
3. //再分析里面的Service，Map<clusterName, Cluster>
4. private Map<String, Cluster> clusterMap = new HashMap<>();
5. //再分析Cluster
6. private Set<Instance> persistentInstances = new HashSet<>();
7. private Set<Instance> ephemeralInstances = new HashSet<>();

复制代码

【2】分析注册表为何要这么设计

1. 1.注册表是基于第一层ConcurrentHashMap，第二层ConcurrentSkipListMap，第三层HashMap，然后定位到对应的Cluster。
2. 2.至于为什么要这样设计，一方面是将粒度划分的更细，通过源码分析可知，nacos更新注册表是进行小范围的更新，如定位到Cluster的临时列表ephemeralInstances或者持久列表persistentInstances【这两个都是set集合，所以排除了会有重复的数据】。因为粒度小所以更新速度会更快。
3. 3.其次采用的是 写时复制思想，也就是说，不会影响读取的效率，因为是新开一个副本，将新旧的数据合并到一个新数据里面，然后将引用指向新数据。
4. 4.其次是为了高扩展，对namespace进行划分【对开发环境隔离】，对service进行划分【对服务进行隔离】，对Cluster进行划分【多机房部署，加快访问速度】
5. 5.为了解决并发读写问题，采用的是ConcurrentHashMap与ConcurrentSkipListMap的分段锁，加上Cluster里面的写时复制。其次Cluster里面是不加锁的，因为是单线程进行修改，不存在冲突。
6. 6.虽说牺牲了，一定的实时性，但是大大提高了并发的性能。

复制代码

【3】分析AP架构下为什么高性能的原因

1. 1.因为采用的是异步任务加队列的形式来实现注册的，所以响应很快，然后任务是慢慢做的。
2. 2.Notifier 是在DistroConsistencyServiceImpl类中初始化，默认单线程，而且队列为ArrayBlockingQueue<>(1024 * 1024)。
3. 3.缩小了变更数据的粒度，单线程避免了线程安全问题【不用加锁】。
4. 4.这种方式毫无疑问是会存在问题的，就是响应了但是没有注册上。但是对于这个问题，在客户端里面做了心跳机制，如果检测不到会重新注册。

复制代码

 【4】分析Nacos为什么感知快的原因

1. 采用的是客户端定时进行一次拉取，兼服务端采用异步的形式使用UDP发送更新的数据到客户端；
2. 虽然UDP存在通知丢失的情况，但是每隔1s的拉取依旧能很好的保持数据的最终一致性。

复制代码

 
源码分析

验证服务端

【1】在启动的时候我们一般是调用shell脚本启动，查看startup.sh脚本
　　从以下看实际上是调用了java命令启动了个java的项目（-jar ${BASE_DIR}/target/${SERVER}.jar 将参数对应替换后 -jar ${BASE_DIR}/target/nacos-server.jar）
　　去寻找启动入口的时候会发现，它其实是SpringBoot搭建的一个WEB服务。

1. cygwin=false
2. darwin=false
3. os400=false
4. case "`uname`" in
5. CYGWIN*) cygwin=true;;
6. Darwin*) darwin=true;;
7. OS400*) os400=true;;
8. esac
9. error_exit ()
10. {
11.     echo "ERROR: $1 !!"
12.     exit 1
13. }
14. [ ! -e "$JAVA_HOME/bin/java" ] && JAVA_HOME=$HOME/jdk/java
15. [ ! -e "$JAVA_HOME/bin/java" ] && JAVA_HOME=/usr/java
16. [ ! -e "$JAVA_HOME/bin/java" ] && JAVA_HOME=/opt/taobao/java
17. [ ! -e "$JAVA_HOME/bin/java" ] && unset JAVA_HOME
19. if [ -z "$JAVA_HOME" ]; then
20.   if $darwin; then
22.     if [ -x '/usr/libexec/java_home' ] ; then
23.       export JAVA_HOME=`/usr/libexec/java_home`
25.     elif [ -d "/System/Library/Frameworks/JavaVM.framework/Versions/CurrentJDK/Home" ]; then
26.       export JAVA_HOME="/System/Library/Frameworks/JavaVM.framework/Versions/CurrentJDK/Home"
27.     fi
28.   else
29.     JAVA_PATH=`dirname $(readlink -f $(which javac))`
30.     if [ "x$JAVA_PATH" != "x" ]; then
31.       export JAVA_HOME=`dirname $JAVA_PATH 2>/dev/null`
32.     fi
33.   fi
34.   if [ -z "$JAVA_HOME" ]; then
35.         error_exit "Please set the JAVA_HOME variable in your environment, We need java(x64)! jdk8 or later is better!"
36.   fi
37. fi
39. export SERVER="nacos-server"
40. export MODE="cluster"
41. export FUNCTION_MODE="all"
42. export MEMBER_LIST=""
43. export EMBEDDED_STORAGE=""
44. while getopts ":m:f:s:c:p:" opt
45. do
46.     case $opt in
47.         m)
48.             MODE=$OPTARG;;
49.         f)
50.             FUNCTION_MODE=$OPTARG;;
51.         s)
52.             SERVER=$OPTARG;;
53.         c)
54.             MEMBER_LIST=$OPTARG;;
55.         p)
56.             EMBEDDED_STORAGE=$OPTARG;;
57.         ?)
58.         echo "Unknown parameter"
59.         exit 1;;
60.     esac
61. done
63. export JAVA_HOME
64. export JAVA="$JAVA_HOME/bin/java"
65. export BASE_DIR=`cd $(dirname $0)/..; pwd`
66. export CUSTOM_SEARCH_LOCATIONS=file:${BASE_DIR}/conf/
68. #===========================================================================================
69. # JVM Configuration
70. #===========================================================================================
71. if [[ "${MODE}" == "standalone" ]]; then
72.     JAVA_OPT="${JAVA_OPT} -Xms512m -Xmx512m -Xmn256m"
73.     JAVA_OPT="${JAVA_OPT} -Dnacos.standalone=true"
74. else
75.     if [[ "${EMBEDDED_STORAGE}" == "embedded" ]]; then
76.         JAVA_OPT="${JAVA_OPT} -DembeddedStorage=true"
77.     fi
78.     JAVA_OPT="${JAVA_OPT} -server -Xms2g -Xmx2g -Xmn1g -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"
79.     JAVA_OPT="${JAVA_OPT} -XX:-OmitStackTraceInFastThrow -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=${BASE_DIR}/logs/java_heapdump.hprof"
80.     JAVA_OPT="${JAVA_OPT} -XX:-UseLargePages"
82. fi
84. if [[ "${FUNCTION_MODE}" == "config" ]]; then
85.     JAVA_OPT="${JAVA_OPT} -Dnacos.functionMode=config"
86. elif [[ "${FUNCTION_MODE}" == "naming" ]]; then
87.     JAVA_OPT="${JAVA_OPT} -Dnacos.functionMode=naming"
88. fi
90. JAVA_OPT="${JAVA_OPT} -Dnacos.member.list=${MEMBER_LIST}"
92. JAVA_MAJOR_VERSION=$($JAVA -version 2>&1 | sed -E -n 's/.* version "([0-9]*).*$/\1/p')
93. if [[ "$JAVA_MAJOR_VERSION" -ge "9" ]] ; then
94.   JAVA_OPT="${JAVA_OPT} -Xlog:gc*:file=${BASE_DIR}/logs/nacos_gc.log:time,tags:filecount=10,filesize=102400"
95. else
96.   JAVA_OPT="${JAVA_OPT} -Djava.ext.dirs=${JAVA_HOME}/jre/lib/ext:${JAVA_HOME}/lib/ext"
97.   JAVA_OPT="${JAVA_OPT} -Xloggc:${BASE_DIR}/logs/nacos_gc.log -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:+PrintGCTimeStamps -XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=10 -XX:GCLogFileSize=100M"
98. fi
100. JAVA_OPT="${JAVA_OPT} -Dloader.path=${BASE_DIR}/plugins/health,${BASE_DIR}/plugins/cmdb"
101. JAVA_OPT="${JAVA_OPT} -Dnacos.home=${BASE_DIR}"
102. JAVA_OPT="${JAVA_OPT} -jar ${BASE_DIR}/target/${SERVER}.jar"
103. JAVA_OPT="${JAVA_OPT} ${JAVA_OPT_EXT}"
104. JAVA_OPT="${JAVA_OPT} --spring.config.additional-location=${CUSTOM_SEARCH_LOCATIONS}"
105. JAVA_OPT="${JAVA_OPT} --logging.config=${BASE_DIR}/conf/nacos-logback.xml"
106. JAVA_OPT="${JAVA_OPT} --server.max-http-header-size=524288"
108. if [ ! -d "${BASE_DIR}/logs" ]; then
109.   mkdir ${BASE_DIR}/logs
110. fi
112. echo "$JAVA ${JAVA_OPT}"
114. if [[ "${MODE}" == "standalone" ]]; then
115.     echo "nacos is starting with standalone"
116. else
117.     echo "nacos is starting with cluster"
118. fi
120. # check the start.out log output file
121. if [ ! -f "${BASE_DIR}/logs/start.out" ]; then
122.   touch "${BASE_DIR}/logs/start.out"
123. fi
124. # start
125. echo "$JAVA ${JAVA_OPT}" > ${BASE_DIR}/logs/start.out 2>&1 &
126. nohup $JAVA ${JAVA_OPT} nacos.nacos >> ${BASE_DIR}/logs/start.out 2>&1 &
127. echo "nacos is starting，you can check the ${BASE_DIR}/logs/start.out"

复制代码

 
从客户端开始分析

【1】根据自动装配原理（寻找spring.factories文件配置）

1. org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
2.   com.alibaba.cloud.nacos.NacosDiscoveryAutoConfiguration,\
3.   com.alibaba.cloud.nacos.ribbon.RibbonNacosAutoConfiguration,\
4.   com.alibaba.cloud.nacos.endpoint.NacosDiscoveryEndpointAutoConfiguration,\
5.   com.alibaba.cloud.nacos.discovery.NacosDiscoveryClientAutoConfiguration,\
6.   com.alibaba.cloud.nacos.discovery.configclient.NacosConfigServerAutoConfiguration
7. org.springframework.cloud.bootstrap.BootstrapConfiguration=\
8.   com.alibaba.cloud.nacos.discovery.configclient.NacosDiscoveryClientConfigServiceBootstrapConfiguration

复制代码

【2】分析NacosDiscoveryAutoConfiguration类自动装配了什么

1. @Configuration
2. @EnableConfigurationProperties
3. @ConditionalOnNacosDiscoveryEnabled
4. @ConditionalOnProperty(value = "spring.cloud.service-registry.auto-registration.enabled", matchIfMissing = true)
5. @AutoConfigureAfter({ AutoServiceRegistrationConfiguration.class, AutoServiceRegistrationAutoConfiguration.class })
6. public class NacosDiscoveryAutoConfiguration {
8.     @Bean
9.     public NacosServiceRegistry nacosServiceRegistry(
10.             NacosDiscoveryProperties nacosDiscoveryProperties) {
11.         return new NacosServiceRegistry(nacosDiscoveryProperties);
12.     }
14.     @Bean
15.     @ConditionalOnBean(AutoServiceRegistrationProperties.class)
16.     public NacosRegistration nacosRegistration(
17.             NacosDiscoveryProperties nacosDiscoveryProperties,
18.             ApplicationContext context) {
19.         return new NacosRegistration(nacosDiscoveryProperties, context);
20.     }
22.     //可以看出是将上面两个Bean当做参数传入了这个Bean
23.     @Bean
24.     @ConditionalOnBean(AutoServiceRegistrationProperties.class)
25.     public NacosAutoServiceRegistration nacosAutoServiceRegistration(
26.             NacosServiceRegistry registry,
27.             AutoServiceRegistrationProperties autoServiceRegistrationProperties,
28.             NacosRegistration registration) {
29.         return new NacosAutoServiceRegistration(registry, autoServiceRegistrationProperties, registration);
30.     }
31. }

复制代码

 
 【3】分析NacosAutoServiceRegistration类有什么重要性

　　利用监听机制，达到注册服务的目的。监听WebServer初始化事件

1. //class NacosAutoServiceRegistration extends AbstractAutoServiceRegistration<Registration>
2. //abstract class AbstractAutoServiceRegistration<R extends Registration> implements AutoServiceRegistration, ApplicationContextAware, ApplicationListener<WebServerInitializedEvent>
3. //因为继承了ApplicationListener，必然会有监听方法
4. public void onApplicationEvent(WebServerInitializedEvent event) {
5.     <strong>bind</strong>(event);
6. }
8. @Deprecated
9. public void bind(WebServerInitializedEvent event) {
10.     ApplicationContext context = event.getApplicationContext();
11.     if (context instanceof ConfigurableWebServerApplicationContext) {
12.         if ("management".equals(((ConfigurableWebServerApplicationContext) context).getServerNamespace())) {
13.             return;
14.         }
15.     }
16.     this.port.compareAndSet(0, event.getWebServer().getPort());
17.     this.<strong>start</strong>();
18. }
20. public void start() {
21.     if (!isEnabled()) {return;
22.     }
24.     // only initialize if nonSecurePort is greater than 0 and it isn't already running
25.     // because of containerPortInitializer below
26.     if (!this.running.get()) {
27.         this.context.publishEvent(new InstancePreRegisteredEvent(this, getRegistration()));
28.         <strong>register</strong>();
29.         if (shouldRegisterManagement()) {
30.             registerManagement();
31.         }
32.         this.context.publishEvent(new InstanceRegisteredEvent<>(this, getConfiguration()));
33.         this.running.compareAndSet(false, true);
34.     }
36. }
38. protected void register() {
39.     this.serviceRegistry.<strong>register</strong>(getRegistration());
40. }
42. @Override
43. public void register(Registration registration) {
44.     if (StringUtils.isEmpty(registration.getServiceId())) {
45.         return;
46.     }
48.     NamingService namingService = namingService();
49.     String serviceId = registration.getServiceId();
50.     String group = nacosDiscoveryProperties.getGroup();
52.     Instance instance = getNacosInstanceFromRegistration(registration);
54.     try {
55.         <strong>namingService.registerInstance</strong>(serviceId, group, instance);
56.     }
57.     catch (Exception e) {
58.         // rethrow a RuntimeException if the registration is failed.
59.         // issue : https://github.com/alibaba/spring-cloud-alibaba/issues/1132
60.         rethrowRuntimeException(e);
61.     }
62. }

复制代码

 
【4】分析如何注册的【服务注册】

1. //NacosNamingService类的registerInstance方法
2. @Override
3. public void registerInstance(String serviceName, Instance instance) throws NacosException {
4.     <strong>registerInstance</strong>(serviceName, Constants.DEFAULT_GROUP, instance);
5. }
7. //NacosNamingService类#registerInstance方法
8. @Override
9. public void registerInstance(String serviceName, String groupName, Instance instance) throws NacosException {
10.     NamingUtils.checkInstanceIsLegal(instance);
11.     String groupedServiceName = NamingUtils.getGroupedName(serviceName, groupName);
12.     if (instance.isEphemeral()) {
13.         BeatInfo beatInfo = beatReactor.buildBeatInfo(groupedServiceName, instance);<br>　　　　　<strong>//添加一个延时执行的心跳任务
14.         beatReactor.addBeatInfo</strong>(groupedServiceName, beatInfo);
15.     }<br>　　 <strong>//进行服务注册</strong>
16.     <strong>serverProxy.registerService</strong>(groupedServiceName, groupName, instance);
17. }
19. //NamingProxy类#registerService方法
20. public void registerService(String serviceName, String groupName, Instance instance) throws NacosException {
22.     //构建注册参数
23.     final Map<String, String> params = new HashMap<String, String>(16);
24.     params.put(CommonParams.NAMESPACE_ID, namespaceId);
25.     params.put(CommonParams.SERVICE_NAME, serviceName);
26.     params.put(CommonParams.GROUP_NAME, groupName);
27.     params.put(CommonParams.CLUSTER_NAME, instance.getClusterName());
28.     params.put("ip", instance.getIp());
29.     params.put("port", String.valueOf(instance.getPort()));
30.     params.put("weight", String.valueOf(instance.getWeight()));
31.     params.put("enable", String.valueOf(instance.isEnabled()));
32.     params.put("healthy", String.valueOf(instance.isHealthy()));
33.     params.put("ephemeral", String.valueOf(instance.isEphemeral()));
34.     params.put("metadata", JacksonUtils.toJson(instance.getMetadata()));
35.     //向服务端发送请求
36.     //UtilAndComs.nacosUrlInstance=/nacos/v1/ns/instance  也就是官网所示的注册接口地址
37. <strong>    reqApi</strong>(UtilAndComs.nacosUrlInstance, params, HttpMethod.POST);
38.    
39. }
41. public String reqApi(String api, Map<String, String> params, String method) throws NacosException {
42.     return reqApi(api, params, Collections.EMPTY_MAP, method);
43. }
44. public String reqApi(String api, Map<String, String> params, Map<String, String> body, String method) throws NacosException {
45.     return reqApi(api, params, body, getServerList(), method);
46. }
48. public String reqApi(String api, Map<String, String> params, Map<String, String> body, List<String> servers, String method) throws NacosException {
49.    
50.     params.put(CommonParams.NAMESPACE_ID, getNamespaceId());
51.    
52.     if (CollectionUtils.isEmpty(servers) && StringUtils.isBlank(nacosDomain)) {
53.         throw new NacosException(...);
54.     }
55.    
56.     NacosException exception = new NacosException();
57.    
58.     if (StringUtils.isNotBlank(nacosDomain)) {
59.         for (int i = 0; i < maxRetry; i++) {
60.             try {
61.                 return<strong> callServer</strong>(api, params, body, nacosDomain, method);
62.             } catch (NacosException e) {
63.                 exception = e;
64.             }
65.         }
66.     } else {
67.         Random random = new Random(System.currentTimeMillis());
68.         int index = random.nextInt(servers.size());
69.         
70.         for (int i = 0; i < servers.size(); i++) {
71.             String server = servers.get(index);
72.             try {
73.                 return<strong> callServer</strong>(api, params, body, server, method);
74.             } catch (NacosException e) {
75.                 exception = e;
76.             }
77.             index = (index + 1) % servers.size();
78.         }
79.     }
81.     throw new NacosException(...);
82. }
84. public String callServer(String api, Map<String, String> params, Map<String, String> body, String curServer, String method) throws NacosException {
85.     long start = System.currentTimeMillis();
86.     long end = 0;
87.     injectSecurityInfo(params);
88.     Header header = builderHeader();
89.    
90.     String url;
91.     if (curServer.startsWith(UtilAndComs.HTTPS) || curServer.startsWith(UtilAndComs.HTTP)) {
92.         url = curServer + api;
93.     } else {
94.         if (!IPUtil.containsPort(curServer)) {
95.             curServer = curServer + IPUtil.IP_PORT_SPLITER + serverPort;
96.         }
97.         url = NamingHttpClientManager.getInstance().getPrefix() + curServer + api;
98.     }
99.    
100.     try {
101.         //真正远程调用
102.         HttpRestResult<String> restResult = nacosRestTemplate
103.                 .exchangeForm(url, header, Query.newInstance().initParams(params), body, method, String.class);
104.         end = System.currentTimeMillis();
105.         
106.         MetricsMonitor.getNamingRequestMonitor(method, url, String.valueOf(restResult.getCode())).observe(end - start);
107.         
108.         if (restResult.ok()) {
109.             return restResult.getData();
110.         }
111.         if (HttpStatus.SC_NOT_MODIFIED == restResult.getCode()) {
112.             return StringUtils.EMPTY;
113.         }
114.         throw new NacosException(restResult.getCode(), restResult.getMessage());
115.     } catch (Exception e) {
116.         throw new NacosException(NacosException.SERVER_ERROR, e);
117.     }
118. }
120. public <T> HttpRestResult<T> exchangeForm(String url, Header header, Query query, Map<String, String> bodyValues, String httpMethod, Type responseType) throws Exception {
121.     RequestHttpEntity requestHttpEntity = new RequestHttpEntity( header.setContentType(MediaType.APPLICATION_FORM_URLENCODED), query, bodyValues);
122.     return execute(url, httpMethod, requestHttpEntity, responseType);
123. }
125. private <T> HttpRestResult<T> execute(String url, String httpMethod, RequestHttpEntity requestEntity, Type responseType) throws Exception {
126.     URI uri = HttpUtils.buildUri(url, requestEntity.getQuery());
127.    
128.     ResponseHandler<T> responseHandler = super.selectResponseHandler(responseType);
129.     HttpClientResponse response = null;
130.     try {
131.         //使用JdkHttpClientRequest去发起请求
132.         response = this.requestClient().execute(uri, httpMethod, requestEntity);
133.         return responseHandler.handle(response);
134.     } finally {
135.         if (response != null) {
136.             response.close();
137.         }
138.     }
139. }
141. //JdkHttpClientRequest类#execute方法
142. @Override
143. public HttpClientResponse execute(URI uri, String httpMethod, RequestHttpEntity requestHttpEntity) throws Exception {
144.     final Object body = requestHttpEntity.getBody();
145.     final Header headers = requestHttpEntity.getHeaders();
146.     replaceDefaultConfig(requestHttpEntity.getHttpClientConfig());
147.    
148.     HttpURLConnection conn = (HttpURLConnection) uri.toURL().openConnection();
149.     Map<String, String> headerMap = headers.getHeader();
150.     if (headerMap != null && headerMap.size() > 0) {
151.         for (Map.Entry<String, String> entry : headerMap.entrySet()) {
152.             conn.setRequestProperty(entry.getKey(), entry.getValue());
153.         }
154.     }
155.    
156.     conn.setConnectTimeout(this.httpClientConfig.getConTimeOutMillis());
157.     conn.setReadTimeout(this.httpClientConfig.getReadTimeOutMillis());
158.     conn.setRequestMethod(httpMethod);
159.     if (body != null && !"".equals(body)) {
160.         String contentType = headers.getValue(HttpHeaderConsts.CONTENT_TYPE);
161.         String bodyStr = JacksonUtils.toJson(body);
162.         if (MediaType.APPLICATION_FORM_URLENCODED.equals(contentType)) {
163.             Map<String, String> map = JacksonUtils.toObj(bodyStr, HashMap.class);
164.             bodyStr = HttpUtils.encodingParams(map, headers.getCharset());
165.         }
166.         if (bodyStr != null) {
167.             conn.setDoOutput(true);
168.             byte[] b = bodyStr.getBytes();
169.             conn.setRequestProperty("Content-Length", String.valueOf(b.length));
170.             OutputStream outputStream = conn.getOutputStream();
171.             outputStream.write(b, 0, b.length);
172.             outputStream.flush();
173.             IoUtils.closeQuietly(outputStream);
174.         }
175.     }
176.     <strong>conn.connect()</strong>;
177.     return new JdkHttpClientResponse(conn);
178. }

复制代码

 
【5】beatReactor.addBeatInfo  心跳任务的流程【服务心跳】

1. //BeatReactor类#构造方法
2. public BeatReactor(NamingProxy serverProxy, int threadCount) {
3.     this.serverProxy = serverProxy;
4.     //定义延迟的线程池
5.     this.executorService = new ScheduledThreadPoolExecutor(threadCount, new ThreadFactory() {
6.         @Override
7.         public Thread newThread(Runnable r) {
8.             Thread thread = new Thread(r);
9.             thread.setDaemon(true);
10.             thread.setName("com.alibaba.nacos.naming.beat.sender");
11.             return thread;
12.         }
13.     });
14. }
16. //添加任务方法
17. public void addBeatInfo(String serviceName, BeatInfo beatInfo) {
18.     NAMING_LOGGER.info("[BEAT] adding beat: {} to beat map.", beatInfo);
19.     String key = buildKey(serviceName, beatInfo.getIp(), beatInfo.getPort());
20.     BeatInfo existBeat = null;
21.     //fix #1733
22.     if ((existBeat = dom2Beat.remove(key)) != null) {
23.         existBeat.setStopped(true);
24.     }
25.     dom2Beat.put(key, beatInfo);
26.     //实际上就是往延迟的线程池添加任务
27.     executorService.schedule(new BeatTask(beatInfo), beatInfo.getPeriod(), TimeUnit.MILLISECONDS);
28.     MetricsMonitor.getDom2BeatSizeMonitor().set(dom2Beat.size());
29. }
31. //分析心跳任务类，主要都是run方法
32. //这种调用方式eureka中也是
33. class BeatTask implements Runnable {
34.    
35.     BeatInfo beatInfo;
36.    
37.     public BeatTask(BeatInfo beatInfo) {
38.         this.beatInfo = beatInfo;
39.     }
40.    
41.     @Override
42.     public void run() {
43.         if (beatInfo.isStopped()) {
44.             return;
45.         }
46.         long nextTime = beatInfo.getPeriod();
47.         try {
48.             //调用server代理实例发送心跳接口
49.             JsonNode result = serverProxy.sendBeat(beatInfo, BeatReactor.this.lightBeatEnabled);
50.             long interval = result.get("clientBeatInterval").asLong();
51.             boolean lightBeatEnabled = false;
52.             if (result.has(CommonParams.LIGHT_BEAT_ENABLED)) {
53.                 lightBeatEnabled = result.get(CommonParams.LIGHT_BEAT_ENABLED).asBoolean();
54.             }
55.             BeatReactor.this.lightBeatEnabled = lightBeatEnabled;
56.             if (interval > 0) {
57.                 nextTime = interval;
58.             }
59.             int code = NamingResponseCode.OK;
60.             if (result.has(CommonParams.CODE)) {
61.                 code = result.get(CommonParams.CODE).asInt();
62.             }
63.             //服务返回没有，则再次注册
64.             if (code == NamingResponseCode.RESOURCE_NOT_FOUND) {
65.                 Instance instance = new Instance();
66.                 instance.setPort(beatInfo.getPort());
67.                 instance.setIp(beatInfo.getIp());
68.                 instance.setWeight(beatInfo.getWeight());
69.                 instance.setMetadata(beatInfo.getMetadata());
70.                 instance.setClusterName(beatInfo.getCluster());
71.                 instance.setServiceName(beatInfo.getServiceName());
72.                 instance.setInstanceId(instance.getInstanceId());
73.                 instance.setEphemeral(true);
74.                 try {
75.                     //又是一个注册方法的调用
76.                     serverProxy.registerService(beatInfo.getServiceName(),
77.                             NamingUtils.getGroupName(beatInfo.getServiceName()), instance);
78.                 } catch (Exception ignore) {
79.                 }
80.             }
81.         } catch (NacosException ex) {...}
82.         //方法内再次将任务塞入，形成循环调用
83.         executorService.schedule(new BeatTask(beatInfo), nextTime, TimeUnit.MILLISECONDS);
84.     }
85. }
87. public JsonNode sendBeat(BeatInfo beatInfo, boolean lightBeatEnabled) throws NacosException {
89.     Map<String, String> params = new HashMap<String, String>(8);
90.     Map<String, String> bodyMap = new HashMap<String, String>(2);
91.     if (!lightBeatEnabled) {
92.         bodyMap.put("beat", JacksonUtils.toJson(beatInfo));
93.     }
94.     params.put(CommonParams.NAMESPACE_ID, namespaceId);
95.     params.put(CommonParams.SERVICE_NAME, beatInfo.getServiceName());
96.     params.put(CommonParams.CLUSTER_NAME, beatInfo.getCluster());
97.     params.put("ip", beatInfo.getIp());
98.     params.put("port", String.valueOf(beatInfo.getPort()));
99.     //地址为/nacos/v1/ns/instance/beat
100.     String result = reqApi(UtilAndComs.nacosUrlBase + "/instance/beat", params, bodyMap, HttpMethod.PUT);
101.     return JacksonUtils.toObj(result);
102. }

复制代码

 
【6】分析如何引入服务的【服务发现】

1. //NacosNamingService类#getAllInstances方法
2. @Override
3. public List<Instance> getAllInstances(String serviceName, String groupName, List<String> clusters, boolean subscribe) throws NacosException {
4.    
5.     ServiceInfo serviceInfo;
6.     // 是否是订阅模式，默认是true
7.     if (subscribe) {
8.         // 先从客户端缓存获取服务信息
9.         serviceInfo = hostReactor.getServiceInfo(NamingUtils.getGroupedName(serviceName, groupName), StringUtils.join(clusters, ","));
10.     } else {
11.         // 如果本地缓存不存在服务信息，则进行订阅
12.         serviceInfo = hostReactor.getServiceInfoDirectlyFromServer(NamingUtils.getGroupedName(serviceName, groupName), StringUtils.join(clusters, ","));
13.     }
14.     List<Instance> list;
15.     // 从服务信息中获取实例列表
16.     if (serviceInfo == null || CollectionUtils.isEmpty(list = serviceInfo.getHosts())) {
17.         return new ArrayList<Instance>();
18.     }
19.     return list;
20. }

复制代码

 

【6.1】分析先从缓存中拿的hostReactor.getServiceInfo方法

1. //获取服务信息
2. public ServiceInfo getServiceInfo(final String serviceName, final String clusters) {
3.    
4.     NAMING_LOGGER.debug("failover-mode: " + failoverReactor.isFailoverSwitch());
5.     String key = ServiceInfo.getKey(serviceName, clusters);
6.     if (failoverReactor.isFailoverSwitch()) {
7.         return failoverReactor.getService(key);
8.     }
9.     //获取服务的信息
10.     ServiceInfo serviceObj = getServiceInfo0(serviceName, clusters);
11.     //客户端第一次获取这个注册表信息为空
12.     if (null == serviceObj) {
13.         serviceObj = new ServiceInfo(serviceName, clusters);
14.         serviceInfoMap.put(serviceObj.getKey(), serviceObj);
15.         updatingMap.put(serviceName, new Object());
17.         //会去拉取这个注册中心里面的注册表信息
18.         updateServiceNow(serviceName, clusters);
19.         updatingMap.remove(serviceName);
20.         
21.     }
22.     //如果本地缓存里面已有这个注册表信息
23.     else if (updatingMap.containsKey(serviceName)) {
24.         
25.         if (UPDATE_HOLD_INTERVAL > 0) {
26.             // hold a moment waiting for update finish
27.             synchronized (serviceObj) {
28.                 try {
29.                     serviceObj.wait(UPDATE_HOLD_INTERVAL);
30.                 } catch (InterruptedException e) {...}
31.             }
32.         }
33.     }
34.     //客户端会开启一个定时任务，每隔几秒会去拉取注册中心里面的全部实例的信息
35.     scheduleUpdateIfAbsent(serviceName, clusters);
36.    
37.     return serviceInfoMap.get(serviceObj.getKey());
38. }
40. //HostReactor类# Map<String, ServiceInfo> serviceInfoMap属性【这个便是客户端保存实例数据的缓存所在】
41. //实际上是先从serviceInfoMap属性里面拿的
42. private ServiceInfo getServiceInfo0(String serviceName, String clusters) {
43.     String key = ServiceInfo.getKey(serviceName, clusters);
44.     return serviceInfoMap.get(key);
45. }

复制代码

 

【6.1.1】分析远程拉取流程updateServiceNow方法

1. private void updateServiceNow(String serviceName, String clusters) {
2.     try {
3.         updateService(serviceName, clusters);
4.     } catch (NacosException e) {...}
5. }
7. public void updateService(String serviceName, String clusters) throws NacosException {
8.     ServiceInfo oldService = getServiceInfo0(serviceName, clusters);
9.     try {
10.         //远程调用
11.         String result = serverProxy.queryList(serviceName, clusters, pushReceiver.getUdpPort(), false);
12.         
13.         if (StringUtils.isNotEmpty(result)) {
14.             //处理并塞入serviceInfoMap，还会发送一个InstancesChangeEvent事件
15.             processServiceJson(result);
16.         }
17.     } finally {
18.         if (oldService != null) {
19.             synchronized (oldService) {
20.                 oldService.notifyAll();
21.             }
22.         }
23.     }
24. }
26. public String queryList(String serviceName, String clusters, int udpPort, boolean healthyOnly) throws NacosException {
27.    
28.     final Map<String, String> params = new HashMap<String, String>(8);
29.     params.put(CommonParams.NAMESPACE_ID, namespaceId);
30.     params.put(CommonParams.SERVICE_NAME, serviceName);
31.     params.put("clusters", clusters);
32.     params.put("udpPort", String.valueOf(udpPort));
33.     params.put("clientIP", NetUtils.localIP());
34.     params.put("healthyOnly", String.valueOf(healthyOnly));
35.     //调用服务的API，获取服务注册中心里面的全部实例
36.     return reqApi(UtilAndComs.nacosUrlBase + "/instance/list", params, HttpMethod.GET);
37. }

复制代码

 
【6.1.1.1】分析定时任务scheduleUpdateIfAbsent方法做了什么

[code]public void scheduleUpdateIfAbsent(String serviceName, String clusters) {    if (futureMap.get(ServiceInfo.getKey(serviceName, clusters)) != null) {        return;    }        synchronized (futureMap) {        if (futureMap.get(ServiceInfo.getKey(serviceName, clusters)) != null) {            return;        }                ScheduledFuture future = addTask(new UpdateTask(serviceName, clusters));        futureMap.put(ServiceInfo.getKey(serviceName, clusters), future);    }}//DEFAULT_DELAY = 1000L，也就是说是1spublic synchronized ScheduledFuture addTask(UpdateTask task) {    return executor.schedule(task, DEFAULT_DELAY, TimeUnit.MILLISECONDS);}//分析UpdateTask类的run方法@Overridepublic void run() {    long delayTime = DEFAULT_DELAY;        try {        // 根据serviceName获取到当前服务的信息，包括服务器地址列表        ServiceInfo serviceObj = serviceInfoMap.get(ServiceInfo.getKey(serviceName, clusters));        // 如果为空，则重新拉取最新的服务列表        if (serviceObj == null) {            updateService(serviceName, clusters);            return;        }        // 如果时间戳