Fork me on GitHub

framework/SpringCloud/eureka/Eureka服务注册续约注销

字数统计: 4.3k字   |   阅读时长≈ 23分

Eureka注册，续期，注销

单机版server的注册，续期，注销

注册：
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
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
InstanceRegistry
    
    
    @Override
	public void register(InstanceInfo info, int leaseDuration, boolean isReplication) {
    //发布注册事件
		handleRegistration(info, leaseDuration, isReplication);
		super.register(info, leaseDuration, isReplication);
	}
    
 public void register(InstanceInfo registrant, int leaseDuration, boolean isReplication) {
        try {
            read.lock();
            //获取注册表
            Map<String, Lease<InstanceInfo>> gMap = registry.get(registrant.getAppName());
            
            REGISTER.increment(isReplication);
            if (gMap == null) {
                
                //每个app信息在eurekaserver的数据结构是  appname --》 map（appid，instanceInfo ）
                final ConcurrentHashMap<String, Lease<InstanceInfo>> gNewMap = new ConcurrentHashMap<String, Lease<InstanceInfo>>();
                gMap = registry.putIfAbsent(registrant.getAppName(), gNewMap);
                if (gMap == null) {
                    gMap = gNewMap;
                }
            }
            Lease<InstanceInfo> existingLease = gMap.get(registrant.getId());
            // Retain the last dirty timestamp without overwriting it, if there is already a lease
            
            if (existingLease != null && (existingLease.getHolder() != null)) {
                //如果存在已注册的信息，比对过期时间
                Long existingLastDirtyTimestamp = existingLease.getHolder().getLastDirtyTimestamp();
                Long registrationLastDirtyTimestamp = registrant.getLastDirtyTimestamp();
                logger.debug("Existing lease found (existing={}, provided={}", existingLastDirtyTimestamp, registrationLastDirtyTimestamp);

                // this is a > instead of a >= because if the timestamps are equal, we still take the remote transmitted
                // InstanceInfo instead of the server local copy.
                if (existingLastDirtyTimestamp > registrationLastDirtyTimestamp) {
                    logger.warn("There is an existing lease and the existing lease's dirty timestamp {} is greater" +
                            " than the one that is being registered {}", existingLastDirtyTimestamp, registrationLastDirtyTimestamp);
                    logger.warn("Using the existing instanceInfo instead of the new instanceInfo as the registrant");
                    
                    //如果还没有过期，但是注册了相同Lease<InstanceInfo> 租约，则把新的丢弃
                    registrant = existingLease.getHolder();
                }
            } else {
                // The lease does not exist and hence it is a new registration
                synchronized (lock) {
                    if (this.expectedNumberOfClientsSendingRenews > 0) {
                        // Since the client wants to register it, increase the number of clients sending renews
                        this.expectedNumberOfClientsSendingRenews = this.expectedNumberOfClientsSendingRenews + 1;
                        updateRenewsPerMinThreshold();
                    }
                }
                logger.debug("No previous lease information found; it is new registration");
            }
            Lease<InstanceInfo> lease = new Lease<InstanceInfo>(registrant, leaseDuration);
            
            if (existingLease != null) {
                //如果是过期了的租约，想要重新注册，那么启动时间还是设置成原来的
                lease.setServiceUpTimestamp(existingLease.getServiceUpTimestamp());
            }
              //放进去
            gMap.put(registrant.getId(), lease);
          
            //给最近注册者队列 加载这个注册者的信息
            recentRegisteredQueue.add(new Pair<Long, String>(
                    System.currentTimeMillis(),
                    registrant.getAppName() + "(" + registrant.getId() + ")"));
            // This is where the initial state transfer of overridden status happens
            if (!InstanceStatus.UNKNOWN.equals(registrant.getOverriddenStatus())) {
             
                logger.debug("Found overridden status {} for instance {}. Checking to see if needs to be add to the "
                                + "overrides", registrant.getOverriddenStatus(), registrant.getId());
                if (!overriddenInstanceStatusMap.containsKey(registrant.getId())) {
                    logger.info("Not found overridden id {} and hence adding it", registrant.getId());
                    overriddenInstanceStatusMap.put(registrant.getId(), registrant.getOverriddenStatus());
                }
            }
            InstanceStatus overriddenStatusFromMap = overriddenInstanceStatusMap.get(registrant.getId());
            if (overriddenStatusFromMap != null) {
                
                logger.info("Storing overridden status {} from map", overriddenStatusFromMap);
                registrant.setOverriddenStatus(overriddenStatusFromMap);
            }
            
         ------------   //实例状态信息覆盖------------------------   
            

            // 基于重写的状态规则设置状态
            InstanceStatus overriddenInstanceStatus = getOverriddenInstanceStatus(registrant, existingLease, isReplication);
            registrant.setStatusWithoutDirty(overriddenInstanceStatus);

            // 如果租约是以UP状态注册的，请设置lease service UP timestamp
            if (InstanceStatus.UP.equals(registrant.getStatus())) {
                lease.serviceUp();
            }
            registrant.setActionType(ActionType.ADDED);
            
            
            //近期改变队列，加上这个租约信息
            recentlyChangedQueue.add(new RecentlyChangedItem(lease));
            registrant.setLastUpdatedTimestamp();
            
            //清除缓存
            invalidateCache(registrant.getAppName(), registrant.getVIPAddress(), registrant.getSecureVipAddress());
            logger.info("Registered instance {}/{} with status {} (replication={})",
                    registrant.getAppName(), registrant.getId(), registrant.getStatus(), isReplication);
        } finally {
            read.unlock();
        }
    }
续约
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
public boolean renew(final String appName, final String serverId,
			boolean isReplication) {
		log("renew " + appName + " serverId " + serverId + ", isReplication {}"
				+ isReplication);
    //获取所有的 appname 和 appid信息 与传入的信息进行比对
		List<Application> applications = getSortedApplications();
		for (Application input : applications) {
			if (input.getName().equals(appName)) {
				InstanceInfo instance = null;
				for (InstanceInfo info : input.getInstances()) {
					if (info.getId().equals(serverId)) {
						instance = info;
						break;
					}
				}
                //找到发布续约时间
				publishEvent(new EurekaInstanceRenewedEvent(this, appName, serverId,
						instance, isReplication));
				break;
			}
		}
    
		return super.renew(appName, serverId, isReplication);
	}


 public boolean renew(final String appName, final String id, final boolean isReplication) {
        if (super.renew(appName, id, isReplication)) {
            replicateToPeers(Action.Heartbeat, appName, id, null, null, isReplication);
            return true;
        }
        return false;
    }


public boolean renew(String appName, String id, boolean isReplication) {
        RENEW.increment(isReplication);
    
    //通过appname获取 某一种服务的 实例集合
        Map<String, Lease<InstanceInfo>> gMap = registry.get(appName);
        Lease<InstanceInfo> leaseToRenew = null;
        if (gMap != null) {
            //通过appid来获取具体的一台实例
            leaseToRenew = gMap.get(id);
        }
        if (leaseToRenew == null) {
            RENEW_NOT_FOUND.increment(isReplication);
            logger.warn("DS: Registry: lease doesn't exist, registering resource: {} - {}", appName, id);
            return false;
        } else {
            //拿到实例
            InstanceInfo instanceInfo = leaseToRenew.getHolder();
            if (instanceInfo != null) {
                // touchASGCache(instanceInfo.getASGName());
                //获取实例状态
                InstanceStatus overriddenInstanceStatus = this.getOverriddenInstanceStatus(
                        instanceInfo, leaseToRenew, isReplication);
                
                if (overriddenInstanceStatus == InstanceStatus.UNKNOWN) {
                    logger.info("Instance status UNKNOWN possibly due to deleted override for instance {}"
                            + "; re-register required", instanceInfo.getId());
                    RENEW_NOT_FOUND.increment(isReplication);
                    return false;
                }
                if (!instanceInfo.getStatus().equals(overriddenInstanceStatus)) {
                    logger.info(
                            "The instance status {} is different from overridden instance status {} for instance {}. "
                                    + "Hence setting the status to overridden status", instanceInfo.getStatus().name(),
                                    instanceInfo.getOverriddenStatus().name(),
                                    instanceInfo.getId());
                    instanceInfo.setStatusWithoutDirty(overriddenInstanceStatus);

                }
            }
            renewsLastMin.increment();
            //这里进行续期，把当前时间加上过期时间
            leaseToRenew.renew();
            return true;
        }
    }
注销
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
@Override
	public boolean cancel(String appName, String serverId, boolean isReplication) {
        
        //发布注销事件
		handleCancelation(appName, serverId, isReplication);
		return super.cancel(appName, serverId, isReplication);
	}

 @Override
    public boolean cancel(final String appName, final String id,
                          final boolean isReplication) {
        if (super.cancel(appName, id, isReplication)) {
            replicateToPeers(Action.Cancel, appName, id, null, null, isReplication);

            return true;
        }
        return false;
    }

    @Override
    public boolean cancel(String appName, String id, boolean isReplication) {
        return internalCancel(appName, id, isReplication);
    }

 protected boolean internalCancel(String appName, String id, boolean isReplication) {
        try {
            read.lock();
            CANCEL.increment(isReplication);
            Map<String, Lease<InstanceInfo>> gMap = registry.get(appName);
            Lease<InstanceInfo> leaseToCancel = null;
            if (gMap != null) {
                leaseToCancel = gMap.remove(id);
            }
            
            ----以上是为了获取要注销的实例------
            
                //最近注销信息队列 添加 此实例注销信息
            recentCanceledQueue.add(new Pair<Long, String>(System.currentTimeMillis(), appName + "(" + id + ")"));
            
            //移除状态
            InstanceStatus instanceStatus = overriddenInstanceStatusMap.remove(id);
            if (instanceStatus != null) {
                logger.debug("Removed instance id {} from the overridden map which has value {}", id, instanceStatus.name());
            }
            if (leaseToCancel == null) {
                CANCEL_NOT_FOUND.increment(isReplication);
                logger.warn("DS: Registry: cancel failed because Lease is not registered for: {}/{}", appName, id);
                return false;
            } else {
                //记录最后活动的时间
                leaseToCancel.cancel();
                
                InstanceInfo instanceInfo = leaseToCancel.getHolder();
                String vip = null;
                String svip = null;
                if (instanceInfo != null) {
                    //把实例信息标记删除 
                    instanceInfo.setActionType(ActionType.DELETED);
                    //最近改变队列 添加实例
                    recentlyChangedQueue.add(new RecentlyChangedItem(leaseToCancel));
                    //设置最后的更新时间
                    instanceInfo.setLastUpdatedTimestamp();
                    vip = instanceInfo.getVIPAddress();
                    svip = instanceInfo.getSecureVipAddress();
                }
                //消除缓存
                invalidateCache(appName, vip, svip);
                logger.info("Cancelled instance {}/{} (replication={})", appName, id, isReplication);
            }
        } finally {
            read.unlock();
        }

        synchronized (lock) {
            //这里就是改变一下expectedNumberOfClientsSendingRenews
            if (this.expectedNumberOfClientsSendingRenews > 0) {
                // Since the client wants to cancel it, reduce the number of clients to send renews.
                this.expectedNumberOfClientsSendingRenews = this.expectedNumberOfClientsSendingRenews - 1;
                updateRenewsPerMinThreshold();
            }
        }

        return true;
    }
状态更新
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
89
90
91
92
93
94
95
96
97

一是为了续约，lease.lastUpdateTimestamp
    
    二是：实例状态改变   overriddenStatus status  lastDirtyTimestamp actionType
    lastUpdatedTimestamp 

public enum InstanceStatus {
        UP, // Ready to receive traffic
        DOWN, // Do not send traffic- healthcheck callback failed
        STARTING, // 关于启动：初始化要完成-不要发送流量
        OUT_OF_SERVICE, // Intentionally shutdown for traffic（有意的关闭 ）
        UNKNOWN;

        public static InstanceStatus toEnum(String s) {
            if (s != null) {
                try {
                    return InstanceStatus.valueOf(s.toUpperCase());
                } catch (IllegalArgumentException e) {
                    // ignore and fall through to unknown
                    logger.debug("illegal argument supplied to InstanceStatus.valueOf: {}, defaulting to {}", s, UNKNOWN);
                }
            }
            return UNKNOWN;
        }
    }

  public enum ActionType {
        ADDED, // Added in the discovery server
        MODIFIED, // Changed in the discovery server
        DELETED
        // Deleted from the discovery server
    }

@Override
    public boolean statusUpdate(String appName, String id,
                                InstanceStatus newStatus, String lastDirtyTimestamp,
                                boolean isReplication) {
        try {
            read.lock();
            STATUS_UPDATE.increment(isReplication);
            Map<String, Lease<InstanceInfo>> gMap = registry.get(appName);
            Lease<InstanceInfo> lease = null;
            if (gMap != null) {
                lease = gMap.get(id);
            }
            if (lease == null) {
                return false;
            } else {
                
                ---上面都是获取租约的lease的----
                    
                    //续约，就是当前系统时间加上过期时间
                lease.renew();
                InstanceInfo info = lease.getHolder();
                // Lease is always created with its instance info object.
                // This log statement is provided as a safeguard, in case this invariant is violated.
                if (info == null) {
                    logger.error("Found Lease without a holder for instance id {}", id);
                }
                if ((info != null) && !(info.getStatus().equals(newStatus))) {
                    // Mark service as UP if needed
                    //如果新状态时UP的
                  //那么设置启动时间
                    if (InstanceStatus.UP.equals(newStatus)) {
                        lease.serviceUp();
                    }
                    // This is NAC overriden status
                    //把新状态放进去
                    overriddenInstanceStatusMap.put(id, newStatus);
                    // Set it for transfer of overridden status to replica on replica start up
                    //设值 集群启动时可以复制该实例
                    info.setOverriddenStatus(newStatus);
                    long replicaDirtyTimestamp = 0;
                    //该实例信息设置新状态
                    info.setStatusWithoutDirty(newStatus);
                    if (lastDirtyTimestamp != null) {
                        replicaDirtyTimestamp = Long.valueOf(lastDirtyTimestamp);
                    }
                    // If the replication's dirty timestamp is more than the existing one, just update it to the replica's.
                    //如果复制品的脏时间戳大于现有的时间戳，只需将其更新为复制副本的。
                    if (replicaDirtyTimestamp > info.getLastDirtyTimestamp()) {
                        info.setLastDirtyTimestamp(replicaDirtyTimestamp);
                    }
                    //设置 实例为被修改类型
                    info.setActionType(ActionType.MODIFIED);
                    
                    //给队列添加最近修改的信息
                    recentlyChangedQueue.add(new RecentlyChangedItem(lease));
                    info.setLastUpdatedTimestamp();
                    invalidateCache(appName, info.getVIPAddress(), info.getSecureVipAddress());
                }
                return true;
            }
        } finally {
            read.unlock();
        }
    }
删除状态
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
public boolean deleteStatusOverride(String appName, String id,
                                       InstanceStatus newStatus,
                                       String lastDirtyTimestamp,
                                       boolean isReplication) {
       try {
           read.lock();
           STATUS_OVERRIDE_DELETE.increment(isReplication);
           Map<String, Lease<InstanceInfo>> gMap = registry.get(appName);
           Lease<InstanceInfo> lease = null;
           if (gMap != null) {
               lease = gMap.get(id);
           }
           if (lease == null) {
               return false;
           } else {
               
               //更新租约
               lease.renew();
               
               
               InstanceInfo info = lease.getHolder();

               // Lease is always created with its instance info object.
               // This log statement is provided as a safeguard, in case this invariant is violated.
               if (info == null) {
                   logger.error("Found Lease without a holder for instance id {}", id);
               }
			//从所有的实例中把 此id的状态清除
               InstanceStatus currentOverride = overriddenInstanceStatusMap.remove(id);
               if (currentOverride != null && info != null) {
                   //因为此id的实例没有状态信息存储，所以 实例的状态设为UNKNOWN
                   info.setOverriddenStatus(InstanceStatus.UNKNOWN);
                   
                   //设置最后的状态为新传进来的状态
                   info.setStatusWithoutDirty(newStatus);
                   
                   long replicaDirtyTimestamp = 0;
                   if (lastDirtyTimestamp != null) {
                       replicaDirtyTimestamp = Long.valueOf(lastDirtyTimestamp);
                   }
                   // If the replication's dirty timestamp is more than the existing one, just update
                   // it to the replica's.
                   if (replicaDirtyTimestamp > info.getLastDirtyTimestamp()) {
                       info.setLastDirtyTimestamp(replicaDirtyTimestamp);
                   }
                   
                   //设置实例信息被修改
                   info.setActionType(ActionType.MODIFIED);
                   recentlyChangedQueue.add(new RecentlyChangedItem(lease));
                   //最后设置此实例的最后更新时间
                   info.setLastUpdatedTimestamp();
                   invalidateCache(appName, info.getVIPAddress(), info.getSecureVipAddress());
               }
               return true;
           }
       } finally {
           read.unlock();
       }
   }
逐出
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
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
 public void evict() {
       evict(0l);
   }

   public void evict(long additionalLeaseMs) {
       logger.debug("Running the evict task");

       if (!isLeaseExpirationEnabled()) {
           logger.debug("DS: lease expiration is currently disabled.");
           return;
       }

       // We collect first all expired items, to evict them in random order. For large eviction sets,
       // if we do not that, we might wipe out whole apps before self preservation kicks in. By randomizing it,
       // the impact should be evenly distributed across all applications.
       List<Lease<InstanceInfo>> expiredLeases = new ArrayList<>();
       for (Entry<String, Map<String, Lease<InstanceInfo>>> groupEntry : registry.entrySet()) {
           Map<String, Lease<InstanceInfo>> leaseMap = groupEntry.getValue();
           if (leaseMap != null) {
               for (Entry<String, Lease<InstanceInfo>> leaseEntry : leaseMap.entrySet()) {
                   Lease<InstanceInfo> lease = leaseEntry.getValue();
                   if (lease.isExpired(additionalLeaseMs) && lease.getHolder() != null) {
                       expiredLeases.add(lease);
                   }
               }
           }
       }
       ---上面把所有过期的实例全部放到一个集合----------

       // To compensate for GC pauses or drifting local time, we need to use current registry size as a base for
       // triggering self-preservation. Without that we would wipe out full registry.
           
           
           //下面这里是通过一个 【0~1】的阈值乘上 总共的实例，然后与过期的实例数比较，选一个最小的数 ，这个数就是注册中心要删除的数（他不会全删）
       int registrySize = (int) getLocalRegistrySize();
       int registrySizeThreshold = (int) (registrySize * serverConfig.getRenewalPercentThreshold());
       int evictionLimit = registrySize - registrySizeThreshold;

       int toEvict = Math.min(expiredLeases.size(), evictionLimit);
       if (toEvict > 0) {
           logger.info("Evicting {} items (expired={}, evictionLimit={})", toEvict, expiredLeases.size(), evictionLimit);

           Random random = new Random(System.currentTimeMillis());
           for (int i = 0; i < toEvict; i++) {
               // Pick a random item (Knuth shuffle algorithm)
               int next = i + random.nextInt(expiredLeases.size() - i);
               Collections.swap(expiredLeases, i, next);
               Lease<InstanceInfo> lease = expiredLeases.get(i);

               String appName = lease.getHolder().getAppName();
               String id = lease.getHolder().getId();
               EXPIRED.increment();
               logger.warn("DS: Registry: expired lease for {}/{}", appName, id);
               internalCancel(appName, id, false);
           }
       }
   }


protected boolean internalCancel(String appName, String id, boolean isReplication) {
       try {
           read.lock();
           CANCEL.increment(isReplication);
           Map<String, Lease<InstanceInfo>> gMap = registry.get(appName);
           Lease<InstanceInfo> leaseToCancel = null;
           if (gMap != null) {
               leaseToCancel = gMap.remove(id);
           }
           //注销队列加上注销的信息
           recentCanceledQueue.add(new Pair<Long, String>(System.currentTimeMillis(), appName + "(" + id + ")"));
           InstanceStatus instanceStatus = overriddenInstanceStatusMap.remove(id);
           if (instanceStatus != null) {
               logger.debug("Removed instance id {} from the overridden map which has value {}", id, instanceStatus.name());
           }
           if (leaseToCancel == null) {
               CANCEL_NOT_FOUND.increment(isReplication);
               logger.warn("DS: Registry: cancel failed because Lease is not registered for: {}/{}", appName, id);
               return false;
           } else {
               //设置过期时间
               leaseToCancel.cancel();
               InstanceInfo instanceInfo = leaseToCancel.getHolder();
               String vip = null;
               String svip = null;
               if (instanceInfo != null) {
                   instanceInfo.setActionType(ActionType.DELETED);
                   //最近的改变 队列 放入该失效并且移除的租约
                   recentlyChangedQueue.add(new RecentlyChangedItem(leaseToCancel));
                   instanceInfo.setLastUpdatedTimestamp();
                   vip = instanceInfo.getVIPAddress();
                   svip = instanceInfo.getSecureVipAddress();
               }
               invalidateCache(appName, vip, svip);
               logger.info("Cancelled instance {}/{} (replication={})", appName, id, isReplication);
           }
       } finally {
           read.unlock();
       }

       synchronized (lock) {
           //更新阈值
           if (this.expectedNumberOfClientsSendingRenews > 0) {
               // Since the client wants to cancel it, reduce the number of clients to send renews.
               this.expectedNumberOfClientsSendingRenews = this.expectedNumberOfClientsSendingRenews - 1;
               updateRenewsPerMinThreshold();
           }
       }

       return true;
   }

eureka集群间的注册，续期，和注销

注册，续期，和注销：
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
89
90
91
92
93
94
95
96
97
98
99
主要的注册逻辑与 单机版的一样，
    但是 多了集群间的通讯操作
    
    
      public void register(final InstanceInfo info, final boolean isReplication) {
        int leaseDuration = Lease.DEFAULT_DURATION_IN_SECS;
        if (info.getLeaseInfo() != null && info.getLeaseInfo().getDurationInSecs() > 0) {
            leaseDuration = info.getLeaseInfo().getDurationInSecs();
        }
        super.register(info, leaseDuration, isReplication);
    ----上面其实就是单机版的操作-------
    
        replicateToPeers(Action.Register, info.getAppName(), info.getId(), info, null, isReplication);
    } 
    

private void replicateToPeers(Action action, String appName, String id,
                                  InstanceInfo info /* optional */,
                                  InstanceStatus newStatus /* optional */, boolean isReplication) {
        Stopwatch tracer = action.getTimer().start();
        try {
            if (isReplication) {
                numberOfReplicationsLastMin.increment();
            }
            // If it is a replication already, do not replicate again as this will create a poison replication
            if (peerEurekaNodes == Collections.EMPTY_LIST || isReplication) {
                return;
            }
---------------上面   如果集群eureka节点是空的或者不需要进行集群间通信就直接返回--------
    
    
    
            for (final PeerEurekaNode node : peerEurekaNodes.getPeerEurekaNodes()) {
                // If the url represents this host, do not replicate to yourself.
                
                if (peerEurekaNodes.isThisMyUrl(node.getServiceUrl())) {
                    //如果此集群节点是自己，那么跳过
                    continue;
                }
                replicateInstanceActionsToPeers(action, appName, id, info, newStatus, node);
            }
        } finally {
            tracer.stop();
        }
    }



private void replicateInstanceActionsToPeers(Action action, String appName,
                                                 String id, InstanceInfo info, InstanceStatus newStatus,
                                                 PeerEurekaNode node) {
        try {
            InstanceInfo infoFromRegistry;
            CurrentRequestVersion.set(Version.V2);
            switch (action) {
                case Cancel:
                    node.cancel(appName, id);
                    break;
                case Heartbeat:
                    InstanceStatus overriddenStatus = overriddenInstanceStatusMap.get(id);
                    infoFromRegistry = getInstanceByAppAndId(appName, id, false);
                    node.heartbeat(appName, id, infoFromRegistry, overriddenStatus, false);
                    break;
                case Register:
                    node.register(info);
                    break;
                case StatusUpdate:
                    infoFromRegistry = getInstanceByAppAndId(appName, id, false);
                    node.statusUpdate(appName, id, newStatus, infoFromRegistry);
                    break;
                case DeleteStatusOverride:
                    infoFromRegistry = getInstanceByAppAndId(appName, id, false);
                    node.deleteStatusOverride(appName, id, infoFromRegistry);
                    break;
            }
        } catch (Throwable t) {
            logger.error("Cannot replicate information to {} for action {}", node.getServiceUrl(), action.name(), t);
        } finally {
            CurrentRequestVersion.remove();
        }
    }




注册：
        public void register(final InstanceInfo info) throws Exception {
        long expiryTime = System.currentTimeMillis() + getLeaseRenewalOf(info);
        batchingDispatcher.process(
                taskId("register", info),
                new InstanceReplicationTask(targetHost, Action.Register, info, null, true) {
                    public EurekaHttpResponse<Void> execute() {
                        //这里就是一个EurekaHttpClient实现去发送 eureka实例内部消息
                        return replicationClient.register(info);
                    }
                },
                expiryTime
        );
    }

获取服务应用

获取指定服务
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

public Application getApplication(String appName) {
       boolean disableTransparentFallback = serverConfig.disableTransparentFallbackToOtherRegion();
       return this.getApplication(appName, !disableTransparentFallback);
   }

public Application getApplication(String appName, boolean includeRemoteRegion) {
       Application app = null;

       Map<String, Lease<InstanceInfo>> leaseMap = registry.get(appName);

       if (leaseMap != null && leaseMap.size() > 0) {
           for (Entry<String, Lease<InstanceInfo>> entry : leaseMap.entrySet()) {
               if (app == null) {
                   app = new Application(appName);
               }
               //把所有此 appname服务的 实例
               app.addInstance(decorateInstanceInfo(entry.getValue()));
           }
       } else if (includeRemoteRegion) {
           //是否获取集群注册的app
           for (RemoteRegionRegistry remoteRegistry : this.regionNameVSRemoteRegistry.values()) {
               Application application = remoteRegistry.getApplication(appName);
               if (application != null) {
                   return application;
               }
           }
       }
       return app;
   }
获取所有的服务
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
public Applications getApplications() {
       boolean disableTransparentFallback = serverConfig.disableTransparentFallbackToOtherRegion();
       if (disableTransparentFallback) {
           //只从本地获取
           return getApplicationsFromLocalRegionOnly();
       } else {
           //从远程获取
           return getApplicationsFromAllRemoteRegions();  // Behavior of falling back to remote region can be disabled.
       }
   }

public Applications getApplicationsFromLocalRegionOnly() {
       return getApplicationsFromMultipleRegions(EMPTY_STR_ARRAY);
   }


 public Applications getApplicationsFromAllRemoteRegions() {
       return getApplicationsFromMultipleRegions(allKnownRemoteRegions);
   }
拉取本地和远程服务
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

public Applications getApplicationsFromMultipleRegions(String[] remoteRegions) {

       boolean includeRemoteRegion = null != remoteRegions && remoteRegions.length != 0;

       logger.debug("Fetching applications registry with remote regions: {}, Regions argument {}",
               includeRemoteRegion, remoteRegions);

       if (includeRemoteRegion) {
           GET_ALL_WITH_REMOTE_REGIONS_CACHE_MISS.increment();
       } else {
           GET_ALL_CACHE_MISS.increment();
       }
       Applications apps = new Applications();
       apps.setVersion(1L);
       for (Entry<String, Map<String, Lease<InstanceInfo>>> entry : registry.entrySet()) {
           Application app = null;

           if (entry.getValue() != null) {
               for (Entry<String, Lease<InstanceInfo>> stringLeaseEntry : entry.getValue().entrySet()) {
                   Lease<InstanceInfo> lease = stringLeaseEntry.getValue();
                   if (app == null) {
                       app = new Application(lease.getHolder().getAppName());
                   }
                   app.addInstance(decorateInstanceInfo(lease));
               }
           }
           if (app != null) {
               apps.addApplication(app);
           }
       }
    ---上面是获取本地所有的------
    
    ---下面是从已经缓存好的 远程机注册信息中提取注册信息-----
       if (includeRemoteRegion) {
           for (String remoteRegion : remoteRegions) {
               RemoteRegionRegistry remoteRegistry = regionNameVSRemoteRegistry.get(remoteRegion);
               if (null != remoteRegistry) {
                   Applications remoteApps = remoteRegistry.getApplications();
                   for (Application application : remoteApps.getRegisteredApplications()) {
                       if (shouldFetchFromRemoteRegistry(application.getName(), remoteRegion)) {
                           logger.info("Application {}  fetched from the remote region {}",
                                   application.getName(), remoteRegion);

                           Application appInstanceTillNow = apps.getRegisteredApplications(application.getName());
                           if (appInstanceTillNow == null) {
                               appInstanceTillNow = new Application(application.getName());
                               apps.addApplication(appInstanceTillNow);
                           }
                           for (InstanceInfo instanceInfo : application.getInstances()) {
                               appInstanceTillNow.addInstance(instanceInfo);
                           }
                       } else {
                           logger.debug("Application {} not fetched from the remote region {} as there exists a "
                                           + "whitelist and this app is not in the whitelist.",
                                   application.getName(), remoteRegion);
                       }
                   }
               } else {
                   logger.warn("No remote registry available for the remote region {}", remoteRegion);
               }
           }
       }
       apps.setAppsHashCode(apps.getReconcileHashCode());
       return apps;
   }


private boolean shouldFetchFromRemoteRegistry(String appName, String remoteRegion) {
       Set<String> whiteList = serverConfig.getRemoteRegionAppWhitelist(remoteRegion);
       if (null == whiteList) {
           whiteList = serverConfig.getRemoteRegionAppWhitelist(null); // see global whitelist.
       }
       return null == whiteList || whiteList.contains(appName);
   }
通过id获取指定实例
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

public InstanceInfo getInstanceByAppAndId(String appName, String id) {
       return this.getInstanceByAppAndId(appName, id, true);
   }


public InstanceInfo getInstanceByAppAndId(String appName, String id, boolean includeRemoteRegions) {
       Map<String, Lease<InstanceInfo>> leaseMap = registry.get(appName);
       Lease<InstanceInfo> lease = null;
       if (leaseMap != null) {
           lease = leaseMap.get(id);
       }
       if (lease != null
               && (!isLeaseExpirationEnabled() || !lease.isExpired())) {
           //查看是否过期，这里有一个续约阈值和 该实例当前续约次数的比较
           return decorateInstanceInfo(lease);
       } else if (includeRemoteRegions) {
           for (RemoteRegionRegistry remoteRegistry : this.regionNameVSRemoteRegistry.values()) {
               Application application = remoteRegistry.getApplication(appName);
               if (application != null) {
                   return application.getByInstanceId(id);
               }
           }
       }
       return null;
   }


private InstanceInfo decorateInstanceInfo(Lease<InstanceInfo> lease) {
       InstanceInfo info = lease.getHolder();

       // client app settings
       int renewalInterval = LeaseInfo.DEFAULT_LEASE_RENEWAL_INTERVAL;
       int leaseDuration = LeaseInfo.DEFAULT_LEASE_DURATION;

       // TODO: clean this up
       if (info.getLeaseInfo() != null) {
           renewalInterval = info.getLeaseInfo().getRenewalIntervalInSecs();
           leaseDuration = info.getLeaseInfo().getDurationInSecs();
       }

       info.setLeaseInfo(LeaseInfo.Builder.newBuilder()
               .setRegistrationTimestamp(lease.getRegistrationTimestamp())
               .setRenewalTimestamp(lease.getLastRenewalTimestamp())
               .setServiceUpTimestamp(lease.getServiceUpTimestamp())
               .setRenewalIntervalInSecs(renewalInterval)
               .setDurationInSecs(leaseDuration)
               .setEvictionTimestamp(lease.getEvictionTimestamp()).build());

       info.setIsCoordinatingDiscoveryServer();
       return info;
   }
获取最近改变的节点
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
  
//干了一件事，是本地客户端和远程机缓存获取 最近队列修改的消息

public Applications getApplicationDeltasFromMultipleRegions(String[] remoteRegions) {
        if (null == remoteRegions) {
            remoteRegions = allKnownRemoteRegions; // null means all remote regions.
        }

        boolean includeRemoteRegion = remoteRegions.length != 0;

        if (includeRemoteRegion) {
            GET_ALL_WITH_REMOTE_REGIONS_CACHE_MISS_DELTA.increment();
        } else {
            GET_ALL_CACHE_MISS_DELTA.increment();
        }

        Applications apps = new Applications();
        apps.setVersion(responseCache.getVersionDeltaWithRegions().get());
        Map<String, Application> applicationInstancesMap = new HashMap<String, Application>();
        try {
            write.lock();
            Iterator<RecentlyChangedItem> iter = this.recentlyChangedQueue.iterator();
            logger.debug("The number of elements in the delta queue is :{}", this.recentlyChangedQueue.size());
            while (iter.hasNext()) {
                Lease<InstanceInfo> lease = iter.next().getLeaseInfo();
                InstanceInfo instanceInfo = lease.getHolder();
                logger.debug("The instance id {} is found with status {} and actiontype {}",
                        instanceInfo.getId(), instanceInfo.getStatus().name(), instanceInfo.getActionType().name());
                Application app = applicationInstancesMap.get(instanceInfo.getAppName());
                if (app == null) {
                    app = new Application(instanceInfo.getAppName());
                    applicationInstancesMap.put(instanceInfo.getAppName(), app);
                    apps.addApplication(app);
                }
                app.addInstance(new InstanceInfo(decorateInstanceInfo(lease)));
            }

            if (includeRemoteRegion) {
                for (String remoteRegion : remoteRegions) {
                    RemoteRegionRegistry remoteRegistry = regionNameVSRemoteRegistry.get(remoteRegion);
                    if (null != remoteRegistry) {
                        Applications remoteAppsDelta = remoteRegistry.getApplicationDeltas();
                        if (null != remoteAppsDelta) {
                            for (Application application : remoteAppsDelta.getRegisteredApplications()) {
                                if (shouldFetchFromRemoteRegistry(application.getName(), remoteRegion)) {
                                    Application appInstanceTillNow =
                                            apps.getRegisteredApplications(application.getName());
                                    if (appInstanceTillNow == null) {
                                        appInstanceTillNow = new Application(application.getName());
                                        apps.addApplication(appInstanceTillNow);
                                    }
                                    for (InstanceInfo instanceInfo : application.getInstances()) {
                                        appInstanceTillNow.addInstance(new InstanceInfo(instanceInfo));
                                    }
                                }
                            }
                        }
                    }
                }
            }

            Applications allApps = getApplicationsFromMultipleRegions(remoteRegions);
            apps.setAppsHashCode(allApps.getReconcileHashCode());
            return apps;
        } finally {
            write.unlock();
        }
    }
赏

谢谢你请我吃糖果

支付宝
微信

扫一扫，分享到微信

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

很惭愧

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