Eureka应用注册与集群数据同步源码解析

绝地灬酷狼 2023-05-31 15:26 5阅读 0赞

在之前的[EurekaClient自动装配及启动流程解析][EurekaClient]一文中我们提到过，在构造`DiscoveryClient`类时，会把自身注册到服务端，本文就来分析一下这个注册流程

##### 客户端发起注册 #####

boolean register() throws Throwable {	
            logger.info(PREFIX + "{}: registering service...", appPathIdentifier);	
            EurekaHttpResponse<Void> httpResponse;	
            try {	
                httpResponse = eurekaTransport.registrationClient.register(instanceInfo);	
            } catch (Exception e) {	
                logger.warn(PREFIX + "{} - registration failed {}", appPathIdentifier, e.getMessage(), e);	
                throw e;	
            }	
            if (logger.isInfoEnabled()) {	
                logger.info(PREFIX + "{} - registration status: {}", appPathIdentifier, httpResponse.getStatusCode());	
            }	
            return httpResponse.getStatusCode() == 204;	
        }

这个方法中包含的`registrationClient`和`instanceInfo`两个对象在之前的文章中都已经单独拿出来了：[Eureka中重要的对象][Eureka]

##### 服务端接受注册 #####

服务端接受注册的Controller在`ApplicationResource`类中，这里需要注意的是普通客户端注册时其中参数`isReplication`为false，这个参数就是控制集群是否同步的表示

@POST	
        @Consumes({"application/json", "application/xml"})	
        public Response addInstance(InstanceInfo info,	
                                    @HeaderParam(PeerEurekaNode.HEADER_REPLICATION) String isReplication) {	
            logger.debug("Registering instance {} (replication={})", info.getId(), isReplication);	
            // 一系列的参数校验	
            if (isBlank(info.getId())) {	
                return Response.status(400).entity("Missing instanceId").build();	
            } else if (isBlank(info.getHostName())) {	
                return Response.status(400).entity("Missing hostname").build();	
            } else if (isBlank(info.getIPAddr())) {	
                return Response.status(400).entity("Missing ip address").build();	
            } else if (isBlank(info.getAppName())) {	
                return Response.status(400).entity("Missing appName").build();	
            } else if (!appName.equals(info.getAppName())) {	
                return Response.status(400).entity("Mismatched appName, expecting " + appName + " but was " + info.getAppName()).build();	
            } else if (info.getDataCenterInfo() == null) {	
                return Response.status(400).entity("Missing dataCenterInfo").build();	
            } else if (info.getDataCenterInfo().getName() == null) {	
                return Response.status(400).entity("Missing dataCenterInfo Name").build();	
            }	
    
    	
            // AWS的一些东西，不用细看	
            DataCenterInfo dataCenterInfo = info.getDataCenterInfo();	
            if (dataCenterInfo instanceof UniqueIdentifier) {	
                String dataCenterInfoId = ((UniqueIdentifier) dataCenterInfo).getId();	
                if (isBlank(dataCenterInfoId)) {	
                    boolean experimental = "true".equalsIgnoreCase(serverConfig.getExperimental("registration.validation.dataCenterInfoId"));	
                    if (experimental) {	
                        String entity = "DataCenterInfo of type " + dataCenterInfo.getClass() + " must contain a valid id";	
                        return Response.status(400).entity(entity).build();	
                    } else if (dataCenterInfo instanceof AmazonInfo) {	
                        AmazonInfo amazonInfo = (AmazonInfo) dataCenterInfo;	
                        String effectiveId = amazonInfo.get(AmazonInfo.MetaDataKey.instanceId);	
                        if (effectiveId == null) {	
                            amazonInfo.getMetadata().put(AmazonInfo.MetaDataKey.instanceId.getName(), info.getId());	
                        }	
                    } else {	
                        logger.warn("Registering DataCenterInfo of type {} without an appropriate id", dataCenterInfo.getClass());	
                    }	
                }	
            }	
        //注册	
            registry.register(info, "true".equals(isReplication));	
            return Response.status(204).build();  // 204 to be backwards compatible	
        }

接着往下看注册的处理逻辑

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);	
        }

`Lease`租约这个类之前也分析过了，不再展开了

###### 服务端保存注册信息 ######

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) {	
                    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());	
                // 当存在这个应用的注册信息时	
                if (existingLease != null && (existingLease.getHolder() != null)) {	
                    Long existingLastDirtyTimestamp = existingLease.getHolder().getLastDirtyTimestamp();	
                    Long registrationLastDirtyTimestamp = registrant.getLastDirtyTimestamp();	
                    logger.debug("Existing lease found (existing={}, provided={}", existingLastDirtyTimestamp, registrationLastDirtyTimestamp);	
    
    	
                    //使用注册时间长的一方的应用信息	
                    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");	
                        registrant = existingLease.getHolder();	
                    }	
                } else {	
    
    	
                    synchronized (lock) {	
                        if (this.expectedNumberOfRenewsPerMin > 0) {	
                            this.expectedNumberOfRenewsPerMin = this.expectedNumberOfRenewsPerMin + 2;	
                            this.numberOfRenewsPerMinThreshold =	
                                    (int) (this.expectedNumberOfRenewsPerMin * serverConfig.getRenewalPercentThreshold());	
                        }	
                    }	
                    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);	
                synchronized (recentRegisteredQueue) {	
                  //添加到最近注册队列	
                    recentRegisteredQueue.add(new Pair<Long, String>(	
                            System.currentTimeMillis(),	
                            registrant.getAppName() + "(" + registrant.getId() + ")"));	
                }	
                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);	
    
    	
                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.  实例信息是由一个map对象保存的，在这个map中，key是应用的appName，value是另外一个map，在这个map中key是应用的id，而value则是应用的租约信息，map对象如下：

private final ConcurrentHashMap<String, Map<String, Lease<InstanceInfo>>> 	
    registry        = new ConcurrentHashMap<String, Map<String, Lease<InstanceInfo>>>();

1.  大体流程为，先查看是否存在该appName的注册信息，如不存在则创建。
2.  接着查看是否存在该id的注册信息，如果存在判断客户端的最后修改时间，记录最后修改的实例，如果不存在则设置自我保护模式的几个参数
3.  根据实例创建对应的租约信息，然后添加到map对象中
4.  添加到最近注册队列
5.  添加到应用实例覆盖状态映射
6.  设置应用实例最终状态，添加最近租约变更队列，设置缓存等

###### 集群数据同步 ######

当`isReplication`属性为true的时候，就会牵扯到集群信息同步了

private void replicateToPeers(Action action, String appName, String id,	
                                      InstanceInfo info ,	
                                      InstanceStatus newStatus , boolean isReplication) {	
            Stopwatch tracer = action.getTimer().start();	
            try {	
                if (isReplication) {	
                    numberOfReplicationsLastMin.increment();	
                }	
                if (peerEurekaNodes == Collections.EMPTY_LIST || isReplication) {	
                    return;	
                }	
    
    	
                for (final PeerEurekaNode node : peerEurekaNodes.getPeerEurekaNodes()) {	
                    if (peerEurekaNodes.isThisMyUrl(node.getServiceUrl())) {	
                        continue;	
                    }	
                    replicateInstanceActionsToPeers(action, appName, id, info, newStatus, node);	
                }	
            } finally {	
                tracer.stop();	
            }	
        }

关于`PeerEurekaNode`在之前的文章也提到了，保存了集群节点信息，这里可以看到是循环所有的集群节点，然后排除本身的信息之后调用了`replicateInstanceActionsToPeers`方法

private void replicateInstanceActionsToPeers(Action action, String appName,	
                                                     String id, InstanceInfo info, InstanceStatus newStatus,	
                                                     PeerEurekaNode node) {	
            try {	
                InstanceInfo infoFromRegistry = null;	
                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);	
            }	
        }

这里重点关注Register分支

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() {	
                            return replicationClient.register(info);	
                        }	
                    },	
                    expiryTime	
            );	
        }

在不关心Eureka自有的调度处理相关的前提下，核心代码是`replicationClient.register(info)`，也就是说，如果当客户端注册时指定需要同步集群信息时，Eureka会把这个客户端再注册到它所在的集群其它的节点上

##### 服务端发起集群数据同步 #####

在之前得[EurekaServer自动装配及启动流程解析][EurekaServer]一文中，我们提到过在初始化服务端的时候会从EurekaServer集群中同步数据，也就是下面这段代码：

public class EurekaServerBootstrap {   	
        protected void initEurekaServerContext() throws Exception {	
      //xxxx	
    
    	
                    EurekaServerContextHolder.initialize(this.serverContext);	
    
    	
                    log.info("Initialized server context");	
    
    	
                    // 从其他 Eureka-Server 拉取注册信息	
                    int registryCount = this.registry.syncUp();	
                    this.registry.openForTraffic(this.applicationInfoManager, registryCount);	
    
    	
                    // Register all monitoring statistics.	
                    EurekaMonitors.registerAllStats();	
            }

数据同步的代码在`syncUp`中

public int syncUp() {	
            int count = 0;	
    
    	
            for (int i = 0; ((i < serverConfig.getRegistrySyncRetries()) && (count == 0)); i++) {	
                if (i > 0) {	
                  //未读取到注册信息则开始等待	
                    try {	
                        Thread.sleep(serverConfig.getRegistrySyncRetryWaitMs());	
                    } catch (InterruptedException e) {	
                        logger.warn("Interrupted during registry transfer..");	
                        break;	
                    }	
                }	
               // 获取注册信息	
                Applications apps = eurekaClient.getApplications();	
                for (Application app : apps.getRegisteredApplications()) {	
                    for (InstanceInfo instance : app.getInstances()) {	
                        try {	
                          //判断是否可以注册，这里基于AWS环境的判断，如果不是AWS环境直接返回true，故不需要关心这个问题	
                            if (isRegisterable(instance)) {	
                              //注册	
                                register(instance, instance.getLeaseInfo().getDurationInSecs(), true);	
                                count++;	
                            }	
                        } catch (Throwable t) {	
                            logger.error("During DS init copy", t);	
                        }	
                    }	
                }	
            }	
            return count;	
        }

这里的处理流程就是当前EurekaServer获取到客户端的注册信息之后，就会再次调用上边咱们提到的服务端Controller调用的`register`方法，当然这次调用时`isReplication`参数就变为true了

往期好文

![640?wx\_fmt=png][640_wx_fmt_png]

**如果文章对您有所帮助，收藏、转发、在看安排一下！！！**

[EurekaClient]: https://mp.weixin.qq.com/s?__biz=MzU5MDgzOTYzMw==&mid=2247484711&idx=1&sn=b10645a5926b56d9ca0bd058c0a6e919&scene=21#wechat_redirect
[Eureka]: https://mp.weixin.qq.com/s?__biz=MzU5MDgzOTYzMw==&mid=2247484712&idx=1&sn=f9ec8d4451b5bd58f6a71fbb281a0257&scene=21#wechat_redirect
[EurekaServer]: https://mp.weixin.qq.com/s?__biz=MzU5MDgzOTYzMw==&mid=2247484710&idx=1&sn=d705ade677238be9027ede095de437c0&scene=21#wechat_redirect
[640_wx_fmt_png]: /images/20230531/8ef613184a1b4180aa6f81db536c4a2b.png