Fork me on GitHub

framework/SpringCloud/Nacos/Nacos名词

字数统计: 3.1k字   |   阅读时长≈ 18分

Nacos名词

NamingFactory

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

Naming ： 提供 通过命名 来查找服务
那么这个NamingFactory就是创建Naming的
    
 /**
     * Create a new naming service.
     *
     * @param serverList server list
     * @return new naming service
     * @throws NacosException nacos exception
     */
    public static NamingService createNamingService(String serverList) throws NacosException {
        try {
            Class<?> driverImplClass = Class.forName("com.alibaba.nacos.client.naming.NacosNamingService");
            Constructor constructor = driverImplClass.getConstructor(String.class);
            NamingService vendorImpl = (NamingService) constructor.newInstance(serverList);
            return vendorImpl;
        } catch (Throwable e) {
            throw new NacosException(NacosException.CLIENT_INVALID_PARAM, e);
        }
    }
    
    /**
     * Create a new naming service.
     *
     * @param properties naming service properties
     * @return new naming service
     * @throws NacosException nacos exception
     */
    public static NamingService createNamingService(Properties properties) throws NacosException {
        try {
            Class<?> driverImplClass = Class.forName("com.alibaba.nacos.client.naming.NacosNamingService");
            Constructor constructor = driverImplClass.getConstructor(Properties.class);
            NamingService vendorImpl = (NamingService) constructor.newInstance(properties);
            return vendorImpl;
        } catch (Throwable e) {
            throw new NacosException(NacosException.CLIENT_INVALID_PARAM, e);
        }
    }

NacosNamingService

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
继承  

/**
     * 每个命名服务（注册中心）应该有不同的命名空间。
     */
    private String namespace;
    
    private String endpoint;
    
    private String serverList;
    
    private String cacheDir;
    
    private String logName;
    
    private HostReactor hostReactor;
    
    private BeatReactor beatReactor;
    
    private EventDispatcher eventDispatcher;
    
    private NamingProxy serverProxy;
    

private void init(Properties properties) throws NacosException {
        ValidatorUtils.checkInitParam(properties);
        this.namespace = InitUtils.initNamespaceForNaming(properties);
        InitUtils.initSerialization();
        initServerAddr(properties);
        InitUtils.initWebRootContext();
        initCacheDir();
        initLogName(properties);
        
        this.eventDispatcher = new EventDispatcher();
        this.serverProxy = new NamingProxy(this.namespace, this.endpoint, this.serverList, properties);
        this.beatReactor = new BeatReactor(this.serverProxy, initClientBeatThreadCount(properties));
        this.hostReactor = new HostReactor(this.eventDispatcher, this.serverProxy, beatReactor, this.cacheDir,
                isLoadCacheAtStart(properties), initPollingThreadCount(properties));
    }

initNamespaceForNaming

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

public static String initNamespaceForNaming(Properties properties) {
       String tmpNamespace = null;
       
       String isUseCloudNamespaceParsing = properties.getProperty(PropertyKeyConst.IS_USE_CLOUD_NAMESPACE_PARSING,
               System.getProperty(SystemPropertyKeyConst.IS_USE_CLOUD_NAMESPACE_PARSING,
                       String.valueOf(Constants.DEFAULT_USE_CLOUD_NAMESPACE_PARSING)));
       
       if (Boolean.parseBoolean(isUseCloudNamespaceParsing)) {
           
           tmpNamespace = TenantUtil.getUserTenantForAns();
           //如果tmpNamespace=null 则从call里面获取
           tmpNamespace = TemplateUtils.stringEmptyAndThenExecute(tmpNamespace, new Callable<String>() {
               @Override
               public String call() {
                   String namespace = System.getProperty(SystemPropertyKeyConst.ANS_NAMESPACE);
                   LogUtils.NAMING_LOGGER.info("initializer namespace from System Property :" + namespace);
                   return namespace;
               }
           });
           
           tmpNamespace = TemplateUtils.stringEmptyAndThenExecute(tmpNamespace, new Callable<String>() {
               @Override
               public String call() {
                   String namespace = System.getenv(PropertyKeyConst.SystemEnv.ALIBABA_ALIWARE_NAMESPACE);
                   LogUtils.NAMING_LOGGER.info("initializer namespace from System Environment :" + namespace);
                   return namespace;
               }
           });
       }
       
       tmpNamespace = TemplateUtils.stringEmptyAndThenExecute(tmpNamespace, new Callable<String>() {
           @Override
           public String call() {
               String namespace = System.getProperty(PropertyKeyConst.NAMESPACE);
               LogUtils.NAMING_LOGGER.info("initializer namespace from System Property :" + namespace);
               return namespace;
           }
       });
       
       if (StringUtils.isEmpty(tmpNamespace) && properties != null) {
           tmpNamespace = properties.getProperty(PropertyKeyConst.NAMESPACE);
       }
       
    
    //最后   public static final String DEFAULT_NAMESPACE_ID = "public";默认是public
       tmpNamespace = TemplateUtils.stringEmptyAndThenExecute(tmpNamespace, new Callable<String>() {
           @Override
           public String call() {
               return UtilAndComs.DEFAULT_NAMESPACE_ID;
           }
       });
       return tmpNamespace;
   }

initSerialization

1
2
3
4
5
6
7
8
9
10

* <p>
   * 子类实现类中的静态代码串中已经向Jackson进行了注册，但是由于classloader的原因，只有当 该子类被使用的时候，才会加载该类。这可能会导致Jackson先进性反序列化，再注册子类，从而导致 反序列化失败。
   * </p>
   */
  public static void initSerialization() {
      // TODO register in implementation class or remove subType
      JacksonUtils.registerSubtype(NoneSelector.class, SelectorType.none.name());
      JacksonUtils.registerSubtype(ExpressionSelector.class, SelectorType.label.name());
  }

initServerAddr

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

private void initServerAddr(Properties properties) {
       serverList = properties.getProperty(PropertyKeyConst.SERVER_ADDR);
       endpoint = InitUtils.initEndpoint(properties);
       if (StringUtils.isNotEmpty(endpoint)) {
           serverList = "";
       }
   }

 public static String initEndpoint(final Properties properties) {
       if (properties == null) {
           
           return "";
       }
       // Whether to enable domain name resolution rules
       String isUseEndpointRuleParsing = properties.getProperty(PropertyKeyConst.IS_USE_ENDPOINT_PARSING_RULE,
               System.getProperty(SystemPropertyKeyConst.IS_USE_ENDPOINT_PARSING_RULE,
                       String.valueOf(ParamUtil.USE_ENDPOINT_PARSING_RULE_DEFAULT_VALUE)));
       
       boolean isUseEndpointParsingRule = Boolean.parseBoolean(isUseEndpointRuleParsing);
       String endpointUrl;
       if (isUseEndpointParsingRule) {
           // Get the set domain name information
           endpointUrl = ParamUtil.parsingEndpointRule(properties.getProperty(PropertyKeyConst.ENDPOINT));
           if (StringUtils.isBlank(endpointUrl)) {
               return "";
           }
       } else {
           endpointUrl = properties.getProperty(PropertyKeyConst.ENDPOINT);
       }
       
       if (StringUtils.isBlank(endpointUrl)) {
           return "";
       }
       
       String endpointPort = TemplateUtils
               .stringEmptyAndThenExecute(System.getenv(PropertyKeyConst.SystemEnv.ALIBABA_ALIWARE_ENDPOINT_PORT),
                       new Callable<String>() {
                           @Override
                           public String call() {
                               
                               return properties.getProperty(PropertyKeyConst.ENDPOINT_PORT);
                           }
                       });
       
       endpointPort = TemplateUtils.stringEmptyAndThenExecute(endpointPort, new Callable<String>() {
           @Override
           public String call() {
               return "8080";
           }
       });
       
       return endpointUrl + ":" + endpointPort;
   }

initWebRootContext

1
2
3
4
5
6
7
8
9
10
11
12
13
14

public static void initWebRootContext() {
    // support the web context with ali-yun if the app deploy by EDAS
    final String webContext = System.getProperty(SystemPropertyKeyConst.NAMING_WEB_CONTEXT);
    TemplateUtils.stringNotEmptyAndThenExecute(webContext, new Runnable() {
        @Override
        public void run() {
            UtilAndComs.webContext = webContext.indexOf("/") > -1 ? webContext : "/" + webContext;
            
            UtilAndComs.nacosUrlBase = UtilAndComs.webContext + "/v1/ns";
            UtilAndComs.nacosUrlInstance = UtilAndComs.nacosUrlBase + "/instance";
        }
    });
}

initCacheDir

1
2
3
4
5
6
7

private void initCacheDir() {
        cacheDir = System.getProperty("com.alibaba.nacos.naming.cache.dir");
        if (StringUtils.isEmpty(cacheDir)) {
            cacheDir = System.getProperty("user.home") + "/nacos/naming/" + namespace;
        }
    }

initLogName

1
2
3
4
5
6
7
8
9
10
11
12
13

private void initLogName(Properties properties) {
       logName = System.getProperty(UtilAndComs.NACOS_NAMING_LOG_NAME);
       if (StringUtils.isEmpty(logName)) {
           
           if (properties != null && StringUtils
                   .isNotEmpty(properties.getProperty(UtilAndComs.NACOS_NAMING_LOG_NAME))) {
               logName = properties.getProperty(UtilAndComs.NACOS_NAMING_LOG_NAME);
           } else {
               logName = "naming.log";
           }
       }
   }

EventDispatcher

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
// 事件分发器,那几可以分发，可以注册事件，此分发器 之分发 服务信息 的事件
EventDispatcher
 private ExecutorService executor = null;
    
    private final BlockingQueue<ServiceInfo> changedServices = new LinkedBlockingQueue<ServiceInfo>();
    
    private final ConcurrentMap<String, List<EventListener>> observerMap = new ConcurrentHashMap<String, List<EventListener>>();
    
    private volatile boolean closed = false;

public EventDispatcher() {
        
        //单线程
        this.executor = Executors.newSingleThreadExecutor(new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r, "com.alibaba.nacos.naming.client.listener");
                thread.setDaemon(true);
                
                return thread;
            }
        });
        
        this.executor.execute(new Notifier());
    }


 private class Notifier implements Runnable {
        
        @Override
        public void run() {
            
            // 循环调度
            while (!closed) {
                
                ServiceInfo serviceInfo = null;
                try {
                    //弹出一个  （服务信息） 
                    serviceInfo = changedServices.poll(5, TimeUnit.MINUTES);
                } catch (Exception ignore) {
                }
                
                if (serviceInfo == null) {
                    continue;
                }
                
                try {
                    List<EventListener> listeners = observerMap.get(serviceInfo.getKey());
                    
                    if (!CollectionUtils.isEmpty(listeners)) {
                        for (EventListener listener : listeners) {
                            // 这是一个不能修改的集合，已修改就报错
                            List<Instance> hosts = Collections.unmodifiableList(serviceInfo.getHosts());
                            
                            listener.onEvent(new NamingEvent(serviceInfo.getName(), serviceInfo.getGroupName(),
                                    serviceInfo.getClusters(), hosts));
                        }
                    }
                    
                } catch (Exception e) {
                    NAMING_LOGGER.error("[NA] notify error for service: " + serviceInfo.getName() + ", clusters: "
                            + serviceInfo.getClusters(), e);
                }
            }
        }
    }

NamingProxy

注册中心的代理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
private final NacosRestTemplate nacosRestTemplate = NamingHttpClientManager.getInstance().getNacosRestTemplate();
   
   private static final int DEFAULT_SERVER_PORT = 8848;
   
   private int serverPort = DEFAULT_SERVER_PORT;
   
   private final String namespaceId;
   
   private final String endpoint;
   
   private String nacosDomain;
   
   private List<String> serverList;
   
   private List<String> serversFromEndpoint = new ArrayList<String>();
   
   private final SecurityProxy securityProxy;
   
   private long lastSrvRefTime = 0L;
   
   private final long vipSrvRefInterMillis = TimeUnit.SECONDS.toMillis(30);
   
   private final long securityInfoRefreshIntervalMills = TimeUnit.SECONDS.toMillis(5);
   
   private Properties properties;
   
   private ScheduledExecutorService executorService;
   


public NamingProxy(String namespaceId, String endpoint, String serverList, Properties properties) {
       
       this.securityProxy = new SecurityProxy(properties, nacosRestTemplate);
       this.properties = properties;
       this.setServerPort(DEFAULT_SERVER_PORT);
       this.namespaceId = namespaceId;
       this.endpoint = endpoint;
       if (StringUtils.isNotEmpty(serverList)) {
           this.serverList = Arrays.asList(serverList.split(","));
           if (this.serverList.size() == 1) {
               this.nacosDomain = serverList;
           }
       }
       this.initRefreshTask();
   }

initRefreshTask

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88

private void initRefreshTask() {
       
       this.executorService = new ScheduledThreadPoolExecutor(2, new ThreadFactory() {
           @Override
           public Thread newThread(Runnable r) {
               Thread t = new Thread(r);
               t.setName("com.alibaba.nacos.client.naming.updater");
               t.setDaemon(true);
               return t;
           }
       });
       
       this.executorService.scheduleWithFixedDelay(new Runnable() {
           @Override
           public void run() {
               refreshSrvIfNeed();
           }
       }, 0, vipSrvRefInterMillis, TimeUnit.MILLISECONDS);
       
       this.executorService.scheduleWithFixedDelay(new Runnable() {
           @Override
           public void run() {
               securityProxy.login(getServerList());
           }
       }, 0, securityInfoRefreshIntervalMills, TimeUnit.MILLISECONDS);
       
       refreshSrvIfNeed();
       this.securityProxy.login(getServerList());
   }



private void refreshSrvIfNeed() {
       try {
           
           //serverList==null 就返回
           if (!CollectionUtils.isEmpty(serverList)) {
               NAMING_LOGGER.debug("server list provided by user: " + serverList);
               return;
           }
           
           if (System.currentTimeMillis() - lastSrvRefTime < vipSrvRefInterMillis) {
               
               //当前刷新时间间隔 少于 vipSrvRefInterMillis
               return;
           }
           
           List<String> list = getServerListFromEndpoint();
           
           if (CollectionUtils.isEmpty(list)) {
               throw new Exception("Can not acquire Nacos list");
           }
           
           if (!CollectionUtils.isEqualCollection(list, serversFromEndpoint)) {
               NAMING_LOGGER.info("[SERVER-LIST] server list is updated: " + list);
           }
           
           serversFromEndpoint = list;
           lastSrvRefTime = System.currentTimeMillis();
       } catch (Throwable e) {
           NAMING_LOGGER.warn("failed to update server list", e);
       }
   }


public boolean login(List<String> servers) {
       
       try {
           //TTL是 Time To Live的缩写，该字段指定IP包被路由器丢弃之前允许通过的最大网段数量。TTL是IPv4包头的一个8 bit字段。
           // tokenTtl：token的生存时间值，tokenRefreshWindow 刷新安全信息的时间窗口（秒）。
           if ((System.currentTimeMillis() - lastRefreshTime) < TimeUnit.SECONDS
                   .toMillis(tokenTtl - tokenRefreshWindow)) {
               //防止反复刷新的？
               return true;
           }
           
           for (String server : servers) {
               if (login(server)) {
                   lastRefreshTime = System.currentTimeMillis();
                   return true;
               }
           }
       } catch (Throwable ignore) {
       }
       
       return false;
   }
getServerListFromEndpoint
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

public List<String> getServerListFromEndpoint() {
       
       try {
           String urlString = "http://" + endpoint + "/nacos/serverlist";
           Header header = builderHeader();
           HttpRestResult<String> restResult = nacosRestTemplate.get(urlString, header, Query.EMPTY, String.class);
           if (!restResult.ok()) {
               throw new IOException(
                       "Error while requesting: " + urlString + "'. Server returned: " + restResult.getCode());
           }
           
           String content = restResult.getData();
           List<String> list = new ArrayList<String>();
           for (String line : IoUtils.readLines(new StringReader(content))) {
               if (!line.trim().isEmpty()) {
                   list.add(line.trim());
               }
           }
           
           return list;
           
       } catch (Exception e) {
           e.printStackTrace();
       }
       
       return null;
   }
login
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

public boolean login(String server) {
      
      if (StringUtils.isNotBlank(username)) {
          Map<String, String> params = new HashMap<String, String>(2);
          Map<String, String> bodyMap = new HashMap<String, String>(2);
          params.put("username", username);
          bodyMap.put("password", password);
          String url = "http://" + server + contextPath + LOGIN_URL;
          
          if (server.contains(Constants.HTTP_PREFIX)) {
              url = server + contextPath + LOGIN_URL;
          }
          try {
              HttpRestResult<String> restResult = nacosRestTemplate
                      .postForm(url, Header.EMPTY, params, bodyMap, String.class);
              if (!restResult.ok()) {
                  SECURITY_LOGGER.error("login failed: {}", JacksonUtils.toJson(restResult));
                  return false;
              }
              JsonNode obj = JacksonUtils.toObj(restResult.getData());
              if (obj.has(Constants.ACCESS_TOKEN)) {
                  accessToken = obj.get(Constants.ACCESS_TOKEN).asText();
                  tokenTtl = obj.get(Constants.TOKEN_TTL).asInt();
                  tokenRefreshWindow = tokenTtl / 10;
              }
          } catch (Exception e) {
              SECURITY_LOGGER.error("[SecurityProxy] login http request failed"
                      + " url: {}, params: {}, bodyMap: {}, errorMsg: {}", url, params, bodyMap, e.getMessage());
              return false;
          }
      }
      return true;
  }

SecurityProxy

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

SecurityProxy
    更新安全信息的安全代理。
    
   private static final Logger SECURITY_LOGGER = LoggerFactory.getLogger(SecurityProxy.class);
    
    private static final String LOGIN_URL = "/v1/auth/users/login";
    
    private final NacosRestTemplate nacosRestTemplate;
    
    private String contextPath;
    
    /**
     * User's name.
     */
    private final String username;
    
    /**
     * User's password.
     */
    private final String password;
    
    /**
     * A token to take with when sending request to Nacos server.
     */
    private String accessToken;
    
    /**
     * TTL of token in seconds.
     */
    private long tokenTtl;
    
    /**
     * Last timestamp refresh security info from server.
     */
    private long lastRefreshTime;
    
    /**
     * time window to refresh security info in seconds.
     */
    private long tokenRefreshWindow;   
    


    public SecurityProxy(Properties properties, NacosRestTemplate nacosRestTemplate) {
        username = properties.getProperty(PropertyKeyConst.USERNAME, StringUtils.EMPTY);
        password = properties.getProperty(PropertyKeyConst.PASSWORD, StringUtils.EMPTY);
        contextPath = properties.getProperty(PropertyKeyConst.CONTEXT_PATH, "/nacos");
        contextPath = contextPath.startsWith("/") ? contextPath : "/" + contextPath;
        this.nacosRestTemplate = nacosRestTemplate;
    }

BeatReactor

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
心跳反应堆。


private final ScheduledExecutorService executorService;
    
    private final NamingProxy serverProxy;
    
    private boolean lightBeatEnabled = false;
    
    public final Map<String, BeatInfo> dom2Beat = new ConcurrentHashMap<String, BeatInfo>();

    public BeatReactor(NamingProxy serverProxy, int threadCount) {
        this.serverProxy = serverProxy;
        this.executorService = new ScheduledThreadPoolExecutor(threadCount, new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setDaemon(true);
                thread.setName("com.alibaba.nacos.naming.beat.sender");
                return thread;
            }
        });
    }

HostReactor

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
//主反应器。

  private static final long DEFAULT_DELAY = 1000L;
    
    private static final long UPDATE_HOLD_INTERVAL = 5000L;
    
    private final Map<String, ScheduledFuture<?>> futureMap = new HashMap<String, ScheduledFuture<?>>();
    
    private final Map<String, ServiceInfo> serviceInfoMap;
    
    private final Map<String, Object> updatingMap;
    
    private final PushReceiver pushReceiver;
    
    private final EventDispatcher eventDispatcher;
    
    private final BeatReactor beatReactor;
    
    private final NamingProxy serverProxy;
    
    private final FailoverReactor failoverReactor;
    
    private final String cacheDir;
    
    private final ScheduledExecutorService executor;


public HostReactor(EventDispatcher eventDispatcher, NamingProxy serverProxy, BeatReactor beatReactor,
            String cacheDir, boolean loadCacheAtStart, int pollingThreadCount) {
        // init executorService
        this.executor = new ScheduledThreadPoolExecutor(pollingThreadCount, new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setDaemon(true);
                thread.setName("com.alibaba.nacos.client.naming.updater");
                return thread;
            }
        });
        this.eventDispatcher = eventDispatcher;
        this.beatReactor = beatReactor;
        this.serverProxy = serverProxy;
        this.cacheDir = cacheDir;
        if (loadCacheAtStart) {
            this.serviceInfoMap = new ConcurrentHashMap<String, ServiceInfo>(DiskCache.read(this.cacheDir));
        } else {
            this.serviceInfoMap = new ConcurrentHashMap<String, ServiceInfo>(16);
        }
        
        this.updatingMap = new ConcurrentHashMap<String, Object>();
        this.failoverReactor = new FailoverReactor(this, cacheDir);
        this.pushReceiver = new PushReceiver(this);
    }

PushReceiver

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
推动接收器。
  private static final Charset UTF_8 = Charset.forName("UTF-8");
    
    private static final int UDP_MSS = 64 * 1024;
    
    private ScheduledExecutorService executorService;
    
    private DatagramSocket udpSocket;
    
    private HostReactor hostReactor;
    
    private volatile boolean closed = false;
    
    public PushReceiver(HostReactor hostReactor) {
        try {
            this.hostReactor = hostReactor;
            this.udpSocket = new DatagramSocket();
            this.executorService = new ScheduledThreadPoolExecutor(1, new ThreadFactory() {
                @Override
                public Thread newThread(Runnable r) {
                    Thread thread = new Thread(r);
                    thread.setDaemon(true);
                    thread.setName("com.alibaba.nacos.naming.push.receiver");
                    return thread;
                }
            });
            
            this.executorService.execute(this);
        } catch (Exception e) {
            NAMING_LOGGER.error("[NA] init udp socket failed", e);
        }
    }

//发布 服务信息修改 信息

   @Override
    public void run() {
        while (!closed) {
            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 = JacksonUtils.toObj(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(JacksonUtils.toJson(hostReactor.getServiceInfoMap()))
                            + "\"}";
                } else {
                    // do nothing send ack only
                    ack = "{\"type\": \"unknown-ack\"" + ", \"lastRefTime\":\"" + pushPacket.lastRefTime
                            + "\", \"data\":" + "\"\"}";
                }
                
                udpSocket.send(new DatagramPacket(ack.getBytes(UTF_8), ack.getBytes(UTF_8).length,
                        packet.getSocketAddress()));
            } catch (Exception e) {
                NAMING_LOGGER.error("[NA] error while receiving push data", e);
            }
        }
    }

FailoverReactor

故障转移反应器。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55


private final String failoverDir;
   
   private final HostReactor hostReactor;
   
   private final ScheduledExecutorService executorService;
   
   public FailoverReactor(HostReactor hostReactor, String cacheDir) {
       this.hostReactor = hostReactor;
       this.failoverDir = cacheDir + "/failover";
       // init executorService
       this.executorService = new ScheduledThreadPoolExecutor(1, new ThreadFactory() {
           @Override
           public Thread newThread(Runnable r) {
               Thread thread = new Thread(r);
               thread.setDaemon(true);
               thread.setName("com.alibaba.nacos.naming.failover");
               return thread;
           }
       });
       this.init();
   }

public void init() {
       
    //这个好像是文件的
       executorService.scheduleWithFixedDelay(new SwitchRefresher(), 0L, 5000L, TimeUnit.MILLISECONDS);
       
    
    //写文件
       executorService.scheduleWithFixedDelay(new DiskFileWriter(), 30, DAY_PERIOD_MINUTES, TimeUnit.MINUTES);
       
       // backup file on startup if failover directory is empty.
       executorService.schedule(new Runnable() {
           @Override
           public void run() {
               try {
                   File cacheDir = new File(failoverDir);
                   
                   if (!cacheDir.exists() && !cacheDir.mkdirs()) {
                       throw new IllegalStateException("failed to create cache dir: " + failoverDir);
                   }
                   
                   File[] files = cacheDir.listFiles();
                   if (files == null || files.length <= 0) {
                       new DiskFileWriter().run();
                   }
               } catch (Throwable e) {
                   NAMING_LOGGER.error("[NA] failed to backup file on startup.", e);
               }
               
           }
       }, 10000L, TimeUnit.MILLISECONDS);
   }

SwitchRefresher

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

class SwitchRefresher implements Runnable {
        
        long lastModifiedMillis = 0L;
        
        @Override
        public void run() {
            try {
                File switchFile = new File(failoverDir + UtilAndComs.FAILOVER_SWITCH);
                if (!switchFile.exists()) {
                    switchParams.put("failover-mode", "false");
                    NAMING_LOGGER.debug("failover switch is not found, " + switchFile.getName());
                    return;
                }
                
                long modified = switchFile.lastModified();
                
                if (lastModifiedMillis < modified) {
                    lastModifiedMillis = modified;
                    String failover = ConcurrentDiskUtil.getFileContent(failoverDir + UtilAndComs.FAILOVER_SWITCH,
                            Charset.defaultCharset().toString());
                    if (!StringUtils.isEmpty(failover)) {
                        String[] lines = failover.split(DiskCache.getLineSeparator());
                        
                        for (String line : lines) {
                            String line1 = line.trim();
                            if ("1".equals(line1)) {
                                switchParams.put("failover-mode", "true");
                                NAMING_LOGGER.info("failover-mode is on");
                                new FailoverFileReader().run();
                            } else if ("0".equals(line1)) {
                                switchParams.put("failover-mode", "false");
                                NAMING_LOGGER.info("failover-mode is off");
                            }
                        }
                    } else {
                        switchParams.put("failover-mode", "false");
                    }
                }
                
            } catch (Throwable e) {
                NAMING_LOGGER.error("[NA] failed to read failover switch.", e);
            }
        }
    }

DiskFileWriter

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

class DiskFileWriter extends TimerTask {
        
        @Override
        public void run() {
            Map<String, ServiceInfo> map = hostReactor.getServiceInfoMap();
            for (Map.Entry<String, ServiceInfo> entry : map.entrySet()) {
                ServiceInfo serviceInfo = entry.getValue();
                if (StringUtils.equals(serviceInfo.getKey(), UtilAndComs.ALL_IPS) || StringUtils
                        .equals(serviceInfo.getName(), UtilAndComs.ENV_LIST_KEY) || StringUtils
                        .equals(serviceInfo.getName(), "00-00---000-ENV_CONFIGS-000---00-00") || StringUtils
                        .equals(serviceInfo.getName(), "vipclient.properties") || StringUtils
                        .equals(serviceInfo.getName(), "00-00---000-ALL_HOSTS-000---00-00")) {
                    continue;
                }
                
                DiskCache.write(serviceInfo, failoverDir);
            }
        }
    }

NamingService

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

这个接口更像一个注册中心，即提供注册，有提供获取，所以namingService的概念就有了，他就是一个注册中心

    
    //注册
void registerInstance(String serviceName, String ip, int port) throws NacosException;

   //从服务中注销实例。
  void deregisterInstance(String serviceName, String ip, int port) throws NacosException;

//获取指定服务名的所有实例
  List<Instance> getAllInstances(String serviceName) throws NacosException;

//选择健康的实例
List<Instance> selectInstances(String serviceName, boolean healthy) throws NacosException;


Instance selectOneHealthyInstance(String serviceName) throws NacosException
    
    
    //订阅，订阅服务接收实例更改事件。
  void subscribe(String serviceName, EventListener listener) throws NacosException;    


    void unsubscribe(String serviceName, EventListener listener) throws NacosException;

//从服务器获取所有服务名称。
ListView<String> getServicesOfServer(int pageNo, int pageSize) throws NacosException;

//获取当前客户端的所有订阅服务。
  List<ServiceInfo> getSubscribeServices() throws NacosException;

  /**
     * get server health status.
     *
     * @return is server healthy
     */
    String getServerStatus();
    
    /**
     * Shutdown the resource service.
     *
     * @throws NacosException exception.
     */
    void shutDown() throws NacosException;

服务端

ServiceManager

赏

谢谢你请我吃糖果

支付宝
微信

扫一扫，分享到微信

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia-plus根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

主要涉及技术：
Java后端开发
开源爱好者、Linux

联系QQ:2960571342

很惭愧

只做了一点微小的工作
谢谢大家