SpringCloud之Eureka的定时任务详解(Client)

迈不过友情╰ 2022-05-21 10:53 382阅读 0赞

1.  **EurekaServer集群节点更新任务：**

public class AsyncResolver<T extends EurekaEndpoint> implements ClosableResolver<T> { 
    
    	/** * 每隔5分钟执行一次重新解析EndPoint 集群任务, 更新服务器集群 */
        private final Runnable updateTask = new Runnable() { 
            @Override
            public void run() { 
                try { 
                    List<T> newList = delegate.getClusterEndpoints(); // 调用 委托的解析器 解析 EndPoint 集群
                    if (newList != null) { 
                        resultsRef.getAndSet(newList);
                        lastLoadTimestamp = System.currentTimeMillis();
                    } else { 
                        logger.warn("Delegate returned null list of cluster endpoints");
                    }
                    logger.debug("Resolved to {}", newList);
                } catch (Exception e) { 
                    logger.warn("Failed to retrieve cluster endpoints from the delegate", e);
                }
            }
        };
    
    }
    
    
    public class ConfigClusterResolver implements ClusterResolver<AwsEndpoint> { 
    	
    	/** * 根据配置来生成EurekaServer集群 * 先获取"availability-zones" 里的当前region对应的zones, 然后获取第一个zone为myZone * 然后从serviceUrl里先获取myZone的urls,放进集合第一位; * 然后从myZone的顺序之后,依次获取zone对应的urls,存入集合 * 然后返回 Map<Zone, List<ServiceUrl>> serviceUrls * * @return */
        private List<AwsEndpoint> getClusterEndpointsFromConfig() { 
            // 获得 可用区,默认值 "defaultZone", 在获取zones时,只会获取到同一个Region下的zone, 也就是availZones的Region是同一个,都是当前应用的Region
            String[] availZones = clientConfig.getAvailabilityZones(clientConfig.getRegion());
            // 获取 应用实例自己 的 可用区, 当有多个可用区是返回第一个; 若未配置返回 default
            String myZone = InstanceInfo.getZone(availZones, myInstanceInfo);
            // 获得 可用区与 serviceUrls 的映射, 返回的结果集是一个LinkedHashMap
            Map<String, List<String>> serviceUrls = EndpointUtils.getServiceUrlsMapFromConfig(clientConfig, myZone, clientConfig.shouldPreferSameZoneEureka());
            // 拼装 EndPoint 集群结果, 相同zone的位于List前端
            List<AwsEndpoint> endpoints = new ArrayList<>();
            for (String zone : serviceUrls.keySet()) { 
                for (String url : serviceUrls.get(zone)) { 
                    try { 
                        //实例化亚马逊云节点, 也就是Eureka Server 节点
                        endpoints.add(new AwsEndpoint(url, getRegion(), zone));
                    } catch (Exception ignore) { 
                        logger.warn("Invalid eureka server URI: {}; removing from the server pool", url);
                    }
                }
            }
    		......
            return endpoints;
        }
    }

每隔5分钟执行一次，根据配置信息生成EurekaServer集群，也就是可以拉取注册信息的服务发现集群；

EurekaClient的配置信息可能如下：

eureka:
      client:
        serviceUrl:
          # 指定Zone和Server地址的映射; 当有多个zone符合时, 默认选择第一个; 一个zone可以有多个应用,也就是多个Server
          defaultZone: http://192.168.157.117:8761/eureka/, http://192.168.157.162:8761/eureka/
          hangzhou: http://192.168.220.122:8761/eureka/
          eu-east-1: http://192.168.223.112:8761/eureka/
    	  
    	# 指定给当前Client属于哪个Region
        region: china
        # 指定当前应用有哪些可用的Zone
        availability-zones:
          china: defaultZone,hangzhou
          eu: eu-east-1,eu-west-1
    
    
    server:
      port: 8763
    
    spring:
      application:
        name: service-eureka

Zone可以理解为空间，Region可以理解为地区，Region是Zone的上一级。Region可为欧洲，美国东部，美国西部，中国等；而Zone则可以理解为城市，比如中国-上海，美国东部-纽约等。  
Netflix广泛应用于AWS，而云服务有很多服务器集群，相同的Zone一般意味着服务器的链路距离较近。

在生成Server集群时，只会选择当前Client的Region下的所有Zone,根据Region,Zone，serviceUrl生成Server节点，将相同Zone的节点排列在结果集的前列，不同的排之后，这样能保证优先选择哪些Zone相同的Server节点来交互。

应用的配置信息可能发生变化，所以需要定时的刷新集群节点的，替换Server集群信息。

2.**EurekaClient内可用的服务信息(InstanceInfo)更新任务**

/** * 注册信息缓存更新线程 * The task that fetches the registry information at specified intervals. */
    class CacheRefreshThread implements Runnable { 
    
    	@Override
    	public void run() { 
    		refreshRegistry();
    	}
    }
    
    public class DiscoveryClient implements EurekaClient { 
    
    	void refreshRegistry() { 
    		......
    		boolean success = fetchRegistry(remoteRegionsModified);
    		......
    	}
    	
    	private boolean fetchRegistry(boolean forceFullRegistryFetch) { 
    		......
    		if(...){ 
    			// 执行 全量获取
                getAndStoreFullRegistry();
    		}else{ 
    			// 执行 增量获取
                getAndUpdateDelta(applications);
    		}
    		......
    	}
    	
    }

EurekaClient在初始化时，从上一步生成的EurekaServer集群中按顺序选择一个来执行全量拉取服务信息；之后正常情况下每隔30S向之前的那个EurekaServer拉取服务的增量信息，当然如果Server不可用，则按顺序向后顺延。

服务的增量信息，指的是服务的改变状态记录，EurekaServer会记录每个Client的注册，取消，状态变更事件，存入一个叫 recentlyChangedQueue 的集合中。当Client获取增量信息时，将此集合内的数据返回给Client,Client再根据每隔元素的ActionType的值，以及其内部的服务租约信息来对应的更新自己的服务信息，以保证服务的时效性。

3.**应用服务InstanceInfo的定时更新任务**

class InstanceInfoReplicator implements Runnable { 
    	
    	/** * 在后台更新, 当应用状态发生改变时触发 * * @return */
        public boolean onDemandUpdate() { 
            if (rateLimiter.acquire(burstSize, allowedRatePerMinute)) {  // 限流相关，跳过
                scheduler.submit(new Runnable() { 
                    @Override
                    public void run() { 
                        logger.debug("Executing on-demand update of local InstanceInfo");
                        // 取消任务
                        Future latestPeriodic = scheduledPeriodicRef.get();
                        if (latestPeriodic != null && !latestPeriodic.isDone()) { 
                            logger.debug("Canceling the latest scheduled update, it will be rescheduled at the end of on demand update");
                            latestPeriodic.cancel(false);
                        }
                        // 再次调用
                        InstanceInfoReplicator.this.run();
                    }
                });
                return true;
            } else { 
                logger.warn("Ignoring onDemand update due to rate limiter");
                return false;
            }
        }
    	
    	/** * 40S后执行应用备份 * * @param initialDelayMs */
        public void start(int initialDelayMs) { 
            if (started.compareAndSet(false, true)) { 
                // 设置脏状态
                instanceInfo.setIsDirty();  // for initial register
                // 提交任务，并设置该任务的 Future
                Future next = scheduler.schedule(this, initialDelayMs, TimeUnit.SECONDS);
                scheduledPeriodicRef.set(next);
            }
        }
    	
    	/** * 如果应用配置发生改变，最长需要30S才会将配置信息更新至InstanceInfo, 然后同步至Eureka Server */
        @Override
        public void run() { 
            try { 
                // 刷新 应用实例信息, 包括hostName,ip,续约频率，续约过期时间,InstanceInfo状态等;
                // 当他们改变时，更新InstanceInfo里的相应数据, 同时设置 isInstanceInfoDirty = true
                discoveryClient.refreshInstanceInfo();
                // 判断 应用实例信息 是否脏了
                Long dirtyTimestamp = instanceInfo.isDirtyWithTime();
                // 如果数据不一致(脏了), 那么重新发起注册, 同步最新数据至Server
                if (dirtyTimestamp != null) { 
                    // 发起注册
                    discoveryClient.register();
                    // 如果注册期间数据未改变,那么还原为 非脏 状态
                    instanceInfo.unsetIsDirty(dirtyTimestamp);
                }
            } catch (Throwable t) { 
                logger.warn("There was a problem with the instance info replicator", t);
            } finally { 
                // 设置下次备份任务, 指定延迟(30S)后再次执行
                Future next = scheduler.schedule(this, replicationIntervalSeconds, TimeUnit.SECONDS);
                scheduledPeriodicRef.set(next);
            }
        }
    }

InstanceInfoReplicator，顾名思义，InstanceInfo复制器，也就是服务信息复制器，它的作用就是复制应用配置信息，重新生成InstanceInfo。

InstanceInfo就是服务信息，比如appName,host,ipAddr,id等等，其他服务能够通过这些信息来发现本服务。

InstanceInfoReplicator会监控当前应用的状态，当状态发生变化时，比如上线UP，下线DOWN，暂停服务OUT\_OF\_SERVICE等，触发onDemandUpdate()方法，立即更新InstanceInfo信息。  
这样当应用上线后，会触发InstanceInfoReplicator的第一次复制生成InstanceInfo，然后向EurekaServer注册自身的InstanceInfo。

在应用启动时，设置40S后正常执行InstanceInfo复制刷新操作，之后每隔30S执行一次。如果在复制刷新的过程中，发现最新的InstanceInfo信息和之前的服务信息不一致时，设置InstanceInfo的状态为dirty, 赋值dirtyTimestamp, 然后重新向EurekaServer注册自身的服务信息，覆盖之前的服务，保证Server上的服务有效性。

4.**定时续约任务，也就是向EurekaServer发送心跳心跳信息**

public class DiscoveryClient implements EurekaClient { 
    	
    	/** * 续租, 如果在续租失败, 重新注册期间, InstanceInfo 被修改过, 也就是说注册的数据是脏的, 设置{@link InstanceInfo#isInstanceInfoDirty} == true * {@link AbstractJerseyEurekaHttpClient#sendHeartBeat(String, String, InstanceInfo, InstanceStatus)} * Renew with the eureka service by making the appropriate REST call */
        boolean renew() { 
            EurekaHttpResponse<InstanceInfo> httpResponse;
            try { 
                httpResponse = eurekaTransport.registrationClient.sendHeartBeat(instanceInfo.getAppName(), instanceInfo.getId(), instanceInfo, null);
                logger.debug("{} - Heartbeat status: {}", PREFIX + appPathIdentifier, httpResponse.getStatusCode());
    			// 如果返回的是404，那么有可能是服务的lastDirtyTimestamp时间比EurekaServer上的对应时间晚，也就是说当前应用的服务信息已经发生了变化
                if (httpResponse.getStatusCode() == 404) { 
                    REREGISTER_COUNTER.increment();
                    logger.info("{} - Re-registering apps/{}", PREFIX + appPathIdentifier, instanceInfo.getAppName());
                    // 设置脏时间 instanceInfo.lastDirtyTimestamp
                    long timestamp = instanceInfo.setIsDirtyWithTime();
                    // 发起注册
                    boolean success = register();
                    if (success) { 
                        // 如果 timestamp >= instanceInfo内的的 lastDirtyTimestamp , 那么说明 instanceInfo 在注册期间未被修改过, 那么重置脏状态
                        // 如果 instanceInfo 在注册期间被修改过, 那么 lastDirtyTimestamp > timestamp, 此时instanceInfo应该是脏状态
                        instanceInfo.unsetIsDirty(timestamp);
                    }
                    return success;
                }
                return httpResponse.getStatusCode() == 200;
            } catch (Throwable e) { 
                logger.error("{} - was unable to send heartbeat!", PREFIX + appPathIdentifier, e);
                return false;
            }
        }
    }
    
    /** * 发送心跳信息 */
    public abstract class AbstractJerseyEurekaHttpClient implements EurekaHttpClient { 
    		
    	public EurekaHttpResponse<InstanceInfo> sendHeartBeat(String appName, String id, InstanceInfo info, InstanceStatus overriddenStatus) { 
    		......
    		WebResource webResource = jerseyClient.resource(serviceUrl)
                    .path(urlPath)
                    .queryParam("status", info.getStatus().toString())
                    .queryParam("lastDirtyTimestamp", info.getLastDirtyTimestamp().toString());
    		......
    	}
    }
    
    @Produces({ "application/xml", "application/json"})
    public class InstanceResource { 
    	
    	@PUT
        public Response renewLease(
            @HeaderParam(PeerEurekaNode.HEADER_REPLICATION) String isReplication,
            @QueryParam("overriddenstatus") String overriddenStatus,
            @QueryParam("status") String status,
            @QueryParam("lastDirtyTimestamp") String lastDirtyTimestamp) { 
    			......
    			if (lastDirtyTimestamp != null && serverConfig.shouldSyncWhenTimestampDiffers()) { 
    				// 如果Client的InstanceInfo发生改变, isDirty, 且最新的currentDirtyTime > lastDirtyTime , 返回404, 需要Client重新注册
    				response = this.validateDirtyTimestamp(Long.valueOf(lastDirtyTimestamp), isFromReplicaNode);
    				......
    			}
    			......
    	}
    		
    	private Response validateDirtyTimestamp(Long lastDirtyTimestamp, boolean isReplication) { 
    		// 请求 的 较大
    		if (lastDirtyTimestamp > appInfo.getLastDirtyTimestamp()) { 
    			logger.debug("Time to sync, since the last dirty timestamp differs -"
    				+ " ReplicationInstance id : {},Registry : {} Incoming: {} Replication: {}", args);
    			// 需要Client重新注册, 因为Client的InstanceInfo已经发生改变了, 设置成Dirty了
    			return Response.status(Status.NOT_FOUND).build();
    			// Server 的 较大
    		} 
    		......
    	}
    }

心跳信息发送到EurekaServer后，其会对比传过来的最新的和现存的lastDirtyTimestamp，如果新的较大，也就是说Client的服务信息发生类变化，此时返回404，需要Client重新注册。  
当lastDirtyTimestamp未发生变化时，更新租约信息的最后更新时间。