二、仍然拷贝一份介绍
Eureka是一项基于REST（代表性状态转移）的服务，主要在AWS云中用于定位服务，以实现负载均衡和中间层服务器的故障转移。
在Netflix，Eureka除了在中间层负载平衡中发挥关键作用外，还用于以下目的。
-
为了协助Netflix Asgard，这是一项开放源代码服务，可简化云部署，
- 在发生问题的情况下快速回滚版本，避免重新启动100个实例，这可能需要很长时间。
- 在滚动推送中，以避免出现问题时将新版本传播到所有实例。
-
对于我们的cassandra（一种分布式key-value数据库）部署，可以使实例流量减少以进行维护。
-
为我们的memcached（分布式的高速内存型缓存系统）缓存服务标识环中的节点列表。
-
用于出于各种其他原因承载有关服务的其他其他特定于应用程序的元数据。
三、如何开始应用
- 启动Eureka-Server：文档参考
@SpringBootApplication @EnableEurekaServer public class Application { public static void main(String[] args) { new SpringApplicationBuilder(Application.class).web(true).run(args); } }
- 配置文件：文档参考
spring: application: # 应用名称，服务中心注册的是这个，同名被认为是同一个服务 name: eureka-server eureka: client: # 是否注册到服务中心，一般服务中心本身不注册到服务中心是可以的 register-with-eureka: false # 设置成true的时候，启动的时候会同步服务中心的注册信息（这个是客户端的操作） fetch-registry: false --- spring: # 可以被认为用于区分不同的配置环境用 profiles: peer1 eureka: client: # 配置服务中心地址 serviceUrl: defaultZone: http://peer1:8761/eureka/,http://peer2:8762/eureka/,http://peer3:8763/eureka/ instance: # 域名地址，不过个人感觉应该生产环境用集群内的高并发会比较多，应该会用内网IP吧 hostname: peer1 server: port: 8761 --- spring: profiles: peer2 server: port: 8762 eureka: client: serviceUrl: defaultZone: http://peer1:8761/eureka/,http://peer2:8762/eureka/,http://peer3:8763/eureka/ instance: hostname: peer2 --- spring: profiles: peer3 server: port: 8763 eureka: client: serviceUrl: defaultZone: http://peer1:8761/eureka/,http://peer2:8762/eureka/,http://peer3:8763/eureka/ instance: hostname: peer3
四、源码阅读：
- 入口：@EnableEurekaServer->@Import EurekaServerMarkerConfiguration->bean Marker->EurekaServerAutoConfiguration(@ConditionalOnBean(EurekaServerMarkerConfiguration.Marker.class))
-
EurekaServerAutoConfiguration
- DefaultEurekaServerContext
@Bean @ConditionalOnMissingBean public EurekaServerContext eurekaServerContext(ServerCodecs serverCodecs, PeerAwareInstanceRegistry registry, PeerEurekaNodes peerEurekaNodes) { return new DefaultEurekaServerContext(this.eurekaServerConfig, serverCodecs, registry, peerEurekaNodes, this.applicationInfoManager); }
- initialize->peerEurekaNodes.start(); // 同步和开启定时同步节点信息
public void start() { taskExecutor = Executors.newSingleThreadScheduledExecutor( new ThreadFactory() { @Override public Thread newThread(Runnable r) { Thread thread = new Thread(r, "Eureka-PeerNodesUpdater"); thread.setDaemon(true); return thread; } } ); try { updatePeerEurekaNodes(resolvePeerUrls()); Runnable peersUpdateTask = new Runnable() { @Override public void run() { try { updatePeerEurekaNodes(resolvePeerUrls()); } catch (Throwable e) { logger.error("Cannot update the replica Nodes", e); } } }; taskExecutor.scheduleWithFixedDelay( peersUpdateTask, serverConfig.getPeerEurekaNodesUpdateIntervalMs(), serverConfig.getPeerEurekaNodesUpdateIntervalMs(), TimeUnit.MILLISECONDS ); } catch (Exception e) { throw new IllegalStateException(e); } for (PeerEurekaNode node : peerEurekaNodes) { logger.info("Replica node URL: {}", node.getServiceUrl()); } }
-
initialize->registry.init(peerEurekaNodes); // PeerAwareInstanceRegistryImpl ，很大一部分的逻辑就在这里了
@Override public void init(PeerEurekaNodes peerEurekaNodes) throws Exception { // 最后重复的次数,这边用的是cas的方式，start里面启动一个定时器，不断将当前状态（同步的次数）同步到最后一次的状态（最后一次同步了几次） this.numberOfReplicationsLastMin.start(); // 普通设值 this.peerEurekaNodes = peerEurekaNodes; // 初始化response 缓存，这边有个方法shouldUseReadOnlyResponseCache（从useReadOnlyResponseCache属性读），如果设其属性为false，则不会启动定时同步write缓存到read缓存的任务，且读取的时候都从write缓存读，实现较高的一致性，这边源码不是很复杂，建议自己看一下吧 initializedResponseCache(); // 开启定时更新续约阈值的任务，会通过计算当前可注册的应用总数来和公式计算结果进行比较，大的话，重新设置expectedNumberOfClientsSendingRenews为可注册的应用总数，来保证稳定性；这块还没弄的很清楚 scheduleRenewalThresholdUpdateTask(); // 估摸着是把远程的区域注册进来 initRemoteRegionRegistry(); try { // 注册到监控中 Monitors.registerObject(this); } catch (Throwable e) { logger.warn("Cannot register the JMX monitor for the InstanceRegistry :", e); } }
- initialize->peerEurekaNodes.start(); // 同步和开启定时同步节点信息
- @Import(EurekaServerInitializerConfiguration.class)
- start()->eurekaServerBootstrap.contextInitialized(EurekaServerInitializerConfiguration.this.servletContext);
- 从peer拉取注册信息
// 从邻近的节点中拷贝注册表信息 int registryCount = this.registry.syncUp();
- 开启剔除任务（其中包含自我保护相关功能）
-
自我保护功能，看下在哪
-
看下上面这个截图，点到里面可以发现自我保护功能在剔除服务的定时任务的前置校验中，具体看下代码(不过我看源码好像有个默认90秒的值会加到过期判断里面，至少要经过90秒之后才会过期。。。不知道是我用错了方式，还是怎样，反正按这个理解的话，不管开关自我保护，都要经过一定的时间长度才会剔除；如果这个成立的话，也说明了为什么快速下线需要主动检查和设置缓存过期)
@Override public boolean isLeaseExpirationEnabled() { // 如果没有开启自我保护，直接返回true进行服务剔除任务执行 if (!isSelfPreservationModeEnabled()) { // The self preservation mode is disabled, hence allowing the instances to expire. return true; } // 否则需要判断每分钟需要更新的阈值次数是否大于0，而且最后一次每分钟续约的次数是否大于每分钟需要更新的阈值次数 return numberOfRenewsPerMinThreshold > 0 && getNumOfRenewsInLastMin() > numberOfRenewsPerMinThreshold; }
- 从peer拉取注册信息
- start()->eurekaServerBootstrap.contextInitialized(EurekaServerInitializerConfiguration.this.servletContext);
- DefaultEurekaServerContext
五、其他比较核心的源码
1、AbstractInstanceRegistry类
1.1、注册
/** * 注册一个具有给定持续时间的新实例。 * * @see com.netflix.eureka.lease.LeaseManager#register(java.lang.Object, int, boolean) */ public void register(InstanceInfo registrant, int leaseDuration, boolean isReplication) { try { // 单机并发锁 read.lock(); // 获取此应用的注册信息Map<registerId, 租约<实例信息>> Map<String, Lease<InstanceInfo>> gMap = registry.get(registrant.getAppName()); // CAS原子自增,注册次数++，如果不是复制的话，则自己区域的注册次数也自增 REGISTER.increment(isReplication); // 如果注册信息为空的话，第一次初始化 if (gMap == null) { final ConcurrentHashMap<String, Lease<InstanceInfo>> gNewMap = new ConcurrentHashMap<String, Lease<InstanceInfo>>(); // 这是一个并发容器，putIfAbsent代表的是如果没有这个key的话就设置进去，因为是并发场景，所以也有可能返回空， // 和DCL（双重检查锁一样的道理） gMap = registry.putIfAbsent(registrant.getAppName(), gNewMap); // 如果不存在映射的情况下，则会返回null，这边手动设置（这块代码还不是很清楚） 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); // 这是>而不是>=，因为如果时间戳相等，我们仍然采用远程传输InstanceInfo而不是服务器本地副本 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.expectedNumberOfClientsSendingRenews > 0) { // 由于客户端想要注册它，增加发送更新的客户端数量 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() + ")")); // 这是覆盖状态的初始状态转移发生的地方 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状态注册，则设置租约服务启动时间戳 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、
/** * 给定应用程序名称的给定实例标记为已更新，并标记它是否来自复制。 * * @see com.netflix.eureka.lease.LeaseManager#renew(java.lang.String, java.lang.String, boolean) */ public boolean renew(String appName, String id, boolean isReplication) { // 更新续约的统计信息，不是复制的话，也要统计到私有的信息中 RENEW.increment(isReplication); // 获取注册表信息 Map<String, Lease<InstanceInfo>> gMap = registry.get(appName); Lease<InstanceInfo> leaseToRenew = null; if (gMap != null) { 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.3、取消。。略
六、优化点
- 自我保护（在剔除的逻辑中）-根据实际节点数量和网络情况进行调整
- 缓存：（开关三级缓存-read only response）
- 打乱server-url的顺序，不要全部配置一样的
- 缩短客户端拉取注册表的时间
- 停服并手动下线
七、CAP
- C一致性：
- 一致性会比较差，因为节点之间的同步是定时的
- 有些信息是在启动的时候才拉取一次
- 增量同步模式更是不能保证一致性
- A可用性：
- 服务剔除，不可用的搞掉
- 租约模式，解耦，减少续约、过期等操作的效率
- 自我保护机制，保护网络抖动而未实际挂掉的服务（滥用也可能导致服务的可用性受到影响，一般节点少的情况下关闭，节点多的情况下开启）等
- P分区容错
- 自我保护机制可以使得即使服务挂掉，仍然可以调用，但会影响一致性