package org.springframework.cloud.netflix.eureka.server;
import com.netflix.appinfo.InstanceInfo; import com.netflix.eureka.cluster.PeerEurekaNode; import com.netflix.eureka.lease.Lease; import com.netflix.eureka.registry.PeerAwareInstanceRegistryImpl; import com.netflix.eureka.resources.; import com.netflix.eureka.registry.; import com.netflix.eureka.registry.AbstractInstanceRegistry.*;
import javax.ws.rs.core.UriInfo;
public class Doc { /** * * 服务启动通过集成spring的Import方式和springboot 的spring.factories文件通过spi的方式注册，不赘述 * * springcloud erueka server 也是mvc架构只不过web容器不同于其他而是使用jersery来接收和处理Http请求 * {@link EurekaServerAutoConfiguration#jerseyApplication} * * 传统的controller层在sc erueka server中以resource命名并且在Spring-cloud-eureka-core包中 * {@link com.netflix.eureka.resources}这个package下有一个{@link ApplicationResource} * 是提供服务注册的控制层类其中 * {@link ApplicationResource#addInstance(InstanceInfo, String)}方法就是服务注册的接收请求方法-> * {@link InstanceRegistry#register(InstanceInfo, boolean)} 开始注册 -> * {@link InstanceRegistry#handleRegistration) 继承spring的监听器机制，将注册事件和注册请求的信息发布 调用父类register方法 * {@link PeerAwareInstanceRegistryImpl#register(InstanceInfo, boolean) * 读取客户端配置服务剔除的时间替换本地剔除时间（最大续约时间周期） 调用父类register方法-> * {@link AbstractInstanceRegistry#register(InstanceInfo registrant, int leaseDuration, boolean isReplication)} * 方法参数解释: * registrant实例对象包含了微服务实例的注册信息， * leaseDuration服务剔除时间 * isReplication在自我保护机制，集群同步中使用 * * 方法的第一行代码 read.lock()这是一个读锁，写锁在{@link AbstractInstanceRegistry#getApplicationDeltasFromMultipleRegions } * 这个方法执行过程中，需要保证 recentlyChangedQueue 和 registry 共享变量的应用实例的状态一致， * 不然返回的增量应用实例集合的状态是不准确的。此时能够达到该效果，必须让这个方法和有读锁的方法互斥。 * * 接着判断 private final ConcurrentHashMap<String, Map<String, Lease>> registry; * Map<String, Lease> gMap = registry.get(registrant.getAppName()); * REGISTER.increment(isReplication); 信息收集用于自我保护机制 * if (gMap == null) * … * ps：参数连接 {@link AbstractInstanceRegistry#registry} 这个registry就是缓存微服务实例的map * registry的数据结构分析前先要知道springcloud中多个spring.appliction.name相同的微服务实例可以称为一个集群，(当前服务器名|代理域名|ip):port可以视为某个微服务的实例ID * 第一个string 表示service-user标识微服务集群 * 第二个string 表示集群中某个微服务实例的实例ID * {@link com.netflix.eureka.lease.Lease 这个对象中 * holder：具体微服务实例的注册信息对象 * evictionTimestamp：服务被剔除的时间 * registrationTimestamp：服务注册时间 * serviceUpTimestamp:服务为UP状态的时间 * lastUpdateTimestamp：最后操作时间（心跳续约、服务注册时间） * duration：最大可允许续约时间 * InstanceInfo就是微服务实例对象 * 代码中registry.get(registrant.getAppName())意思应该是通过微服务集群名称获得这个集群的所有已缓存的实例 * ps:如果在调试过程中这个缓存你是第一次进来，但是发现已经有实例缓存了，这表示客户端判断第一次注册请求时间过长，中断了注册并重新发起了一次 * 如果不存在就new一个内层final ConcurrentHashMap<String, Lease> gNewMap = new ConcurrentHashMap<String, Lease>(); * gMap = registry.putIfAbsent(registrant.getAppName(), gNewMap);这是gMap就有一个空的ConcurrentHashMap对象了 * 然后判断是否存在微服务实例对象存在 * Lease existingLease = gMap.get(registrant.getId()); * if (existingLease != null && (existingLease.getHolder() != null)) * 存在的话可以理解为实例ID注册冲突了，是一个重复注册的问题，需要处理：取得当前存在的最后操作时时间戳existingLastDirtyTimestamp * 与新注册上来的实例的时间戳registrationLastDirtyTimestamp做比较，获得最新的实例注册信息并覆盖 * 如果不存在进入else做自我保护阈值更新自我保护机制后面补齐,接着lease.setServiceUpTimestamp(existingLease.getServiceUpTimestamp()) * 将处理完的微服务实例信息和新的过期时间封装一个新的Lease，然后在existingLease存在的情况下将existingLease服务为UP状态的时间 * 接着gMap.put(registrant.getId(), lease);服务注册到这里就完成了 接下来的逻辑 集群同步 * * * * * 心跳续约从{@link InstanceResource#renewLease(String, String, String, String)方法为入口调用链路和注册差不多 } -> * {@link InstanceResource#renewLease(String, String, String, String) -> * {@link InstanceRegistry#renew(String, String, boolean) 发布实例心跳续约事件，调用super。renew}-> * {@link PeerAwareInstanceRegistryImpl#renew(String, String, boolean) 发布实例心跳续约事件，调用super。renew}-> * {@link AbstractInstanceRegistry#renew(String, String, boolean) * RENEW.increment(isReplication);收集数据 * 根据appname从{@link AbstractInstanceRegistry#registry}获取微服务集群组判断一下不存在则返回false， * {@link InstanceResource#renewLease 接收到发那返回值，封装一个Response.status(Status.NOT_FOUND)返回给客户端 客户端收到404后会把续约操作改为注册操作} * 不为空InstanceInfo instanceInfo = leaseToRenew.getHolder()或者租债器中的服务实例信息。判断一下当前实例在server中的状态UNKNOWN状态同样会返回404给客户端 * overriddenInstanceStatus状态red/black升级，将部分原服务先设置为OUT_OF_SERVICE，停止接收请求，即变为black，之后新部署的服务启动起来，即为red。如果新服务正常， * 就可以关闭旧服务了，假设新服务出现问题，则立马删除掉新服务，将原有服务的overriddenstatus删除掉，恢复UP，恢复接收流量。}-> * {@link Lease#renew() 更新最后操作时间LastUpdateTimestamp = System.currentTimeMillis() + duration以计算续约后的最终过期时间 * 最后操作时间 + 最新的续约时间 + 心跳续约间隔时长 = 续约后的最终过期时间 {@link Lease#isExpired(long)} 其中additionalLeaseMs是弥补集群时间差缺省0 * }; 这里算个小BUG 在计算续约后的最终过期时间时 lastUpdateTimestamp + duration 但是lastUpdateTimestamp在renew时已经+过一次duration了 * 这里又加了一次相当于(心跳续约间隔时长 x 2); 心跳续约完成，正常完成的情况下返回后走同步集群 * * * * 服务下架 主动下架：{@link InstanceResource#cancelLease(String) 注册、续约、下架三个接口的调用链路基本相同}-> * {@link InstanceResource#cancelLease(String) 发布实下架事件，调用super.cancel}} -> * {@link PeerAwareInstanceRegistryImpl#cancel(String, String, boolean) 调用super.cancel} -> * {@link PeerAwareInstanceRegistryImpl#cancel(String, String, boolean) 调用super.cancel} -> * {@link AbstractInstanceRegistry#cancel(String, String, boolean) 调用super.cancel} -> * @link AbstractInstanceRegistry#internalCancel(String, String, boolean) * 使用读锁，收集下架时间CANCEL.increment(isReplication);根据集群服务名从{@link AbstractInstanceRegistry#registry}获得服务集群组的Map * 在Map中根据服务实例ID remove * 服务下架正常完成的情况下返回后走同步集群 * @link AbstractInstanceRegistry#recentlyChangedQueue * * * * * 服务剔除 * {@link EurekaServerInitializerConfiguration -》 * {@link EurekaServerBootstrap#contextInitialized -> * {@link EurekaServerBootstrap#initEurekaServerContext 这还将实例缓存到服务发现的读写缓存中-> * {@link PeerAwareInstanceRegistryImpl#syncUp() -> * {@link PeerAwareInstanceRegistryImpl#openForTraffic * {@link PeerAwareInstanceRegistryImpl#expectedNumberOfRenewsPerMin 这个参数是预期更新次数，假设30秒更新一次 * 那么60秒内这个参数的值应该是需要更新实例的数量 * 2 * {@link PeerAwareInstanceRegistryImpl#numberOfRenewsPerMinThreshold = * (int) (this.expectedNumberOfRenewsPerMin * serverConfig.getRenewalPercentThreshold());自我保护阈值 预期更新次数*0.85 * 值初始化后完成后，更改是在服务注册和服务下架和15分钟自动检查时 * super.postInit() }-》 * {@link AbstractInstanceRegistry#postInit() 初始化服务剔除的线程} * } * {@link EvictionTask#run ,由定时调用Timer调用-> * {@link AbstractInstanceRegistry#evict -> {@link AbstractInstanceRegistry#evict(long)} * if (!isLeaseExpirationEnabled()) { * logger.debug(“DS: lease expiration is currently disabled.”); * return; * }这个判断如果开启了自我保护机制，会根据expectedNumberOfRenewsPerMin和renew时收集的数据判断是否需要进入自我保护状态, * return numberOfRenewsPerMinThreshold > 0 && getNumOfRenewsInLastMin() > numberOfRenewsPerMinThreshold; * * List<Lease> expiredLeases = new ArrayList<>(); * for (Entry<String, Map<String, Lease>> groupEntry : registry.entrySet()) { * Map<String, Lease> leaseMap = groupEntry.getValue(); * if (leaseMap != null) { * for (Entry<String, Lease> leaseEntry : leaseMap.entrySet()) { * Lease lease = leaseEntry.getValue(); * if (lease.isExpired(additionalLeaseMs) && lease.getHolder() != null) { * expiredLeases.add(lease); * } * } * } * }代码片段是从{@link AbstractInstanceRegistry#registry 中获得所有的组，在从组中获得服务实例信息，判断lease.isExpired(additionalLeaseMs) * 是否已达到过期是时间并且未续约的实例lease.isExpired(additionalLeaseMs)放入一个List集合中， * int registrySize = (int) getLocalRegistrySize(); * int registrySizeThreshold = (int) (registrySize * serverConfig.getRenewalPercentThreshold()); * int evictionLimit = registrySize - registrySizeThreshold; 每次计算的总剔除的数量不大于总注册数量的100%-85%=15% * 这三个变量表示registrySize注册服务的总数 serverConfig.getRenewalPercentThreshold()的阈值比例，registrySizeThreshold自我保护的实际阈值，evictionLimit自我保护的实际判断依据 * int toEvict = Math.min(expiredLeases.size(), evictionLimit);拿到更小的数量去随机剔除，在两个值不相等的情况下肯定定会留下部分过期未 * 剔除的服务，这就要留到下一次剔除服务或者从新被激活这样可以降低一组微服务被一次性剔除的可能性这算是高可用的一种体现~~~~吧？最后服务剔除不会集群同步 * } * * * * * 状态修改 * * * * * * * 集群同步{@link PeerAwareInstanceRegistryImpl#replicateToPeers(PeerAwareInstanceRegistryImpl.Action, String appName, String id, InstanceInfo, InstanceInfo.InstanceStatus, boolean)} * Action同步集群的事件枚举，包含Heartbeat, Register, Cancel, StatusUpdate, DeleteStatusOverride 集群同步消息有接收到事件的客户端去轮询service集群 * if (peerEurekaNodes == Collections.EMPTY_LIST || isReplication) { * return; * } * 这个判断当集群为空或者isRepliction为ture，本条消息是集群同步的情况下isRepliction为true，由于是1对多的方式去循环service-urls发送的所以其他接收事件消息的服务可以return， * 走下去，因为下面的代码是集群同步的请求 * {@link PeerAwareInstanceRegistryImpl#replicateInstanceActionsToPeers(PeerAwareInstanceRegistryImpl.Action, String, String, InstanceInfo, InstanceInfo.InstanceStatus, PeerEurekaNode)}-> * {@link PeerEurekaNode#register(InstanceInfo)} * deleteStatusOverride(String, String, InstanceInfo)} * statusUpdate(String, String, InstanceInfo.InstanceStatus, InstanceInfo)} * cancel(String, String)} * heartbeat(String, String, InstanceInfo, InstanceInfo.InstanceStatus, boolean)} }-> * * {@link com.netflix.eureka.transport.JerseyReplicationClient#regis+ter * #deleteStatusOverride * #statusUpdate * #cancel * #heartbeat }发送HTTP请求} -> * * * * * 服务发现 三层缓存机制（conurrentHashMap） 只读缓存 读写缓存(guava) 真实数据(conurrentHashMap) * 获取服务入口{@link ApplicationsResource#getContainers(String, String, String, String, UriInfo, String)} * * 只读缓存初始化更新入口{@link ResponseCacheImpl#getCacheUpdateTask() 1线程Timer方式每30s和读写缓存同步一次 2当请求服务发现时， * 在只读缓存内取到就从读写缓存中获取，并新增到啊只读缓存;在实际使用中客户端做服务发现时有延迟，在服务端服务发现的只读缓存有延时，ribbon缓存有延时， * 这三个延时全取默认值得情况下 30 + 30 + 60 =120秒延迟，这样的各层缓存提高了性能和可用性的提升，弊端也就是牺牲了cap的一致性 * * 读写缓存初始化入口{@link ResponseCacheImpl#ResponseCacheImpl : Value value = generatePayload * {@link ResponseCacheImpl#generatePayload 先判断一下更新类型}-> * {@link AbstractInstanceRegistry#getApplications()全量更新 * {@link AbstractInstanceRegistry#getApplicationDeltas增量更新 这里有写锁 * {@link AbstractInstanceRegistry#getApplicationDeltasFromMultipleRegions多区域的增量更新 这里有写锁 * 增量更新有写锁和服务注册，修改，下架时的读锁成对防止脏数据写入缓存 * 在获取已经注册上来的实例->{@link ResponseCacheImpl#getPayLoad}格式换数据 * * * * * *一个服务下架，剔除最大延时 在全部为缺省值的情况服务最后更新开始计算 90 + bug的90 + 只读缓存延时 30 + 客户端 30 + ribbon 30 = 270秒 * * * * * * * * * * */ }
