系列目录：
- Spring Cloud Eureka 全解 （1） - 总览篇
- Spring Cloud Eureka 全解 （2） - 整体流程篇
- Spring Cloud Eureka 全解 （3） - 核心流程-服务注册与取消详解
- Spring Cloud Eureka 全解 （4） - 核心流程-服务与实例列表获取详解
- Spring Cloud Eureka 全解 （5） - 自我保护机制
- Spring Cloud Eureka 全解 （6） - 一些热门QA
- Spring Cloud Eureka 全解 （7） - 生产配置最佳实践
- Spring Cloud Eureka 全解 （8） - 安全配置
关于服务注册
开启/关闭服务注册配置：eureka.client.register-with-eureka = true （默认）
什么时候注册？
- 应用第一次启动时，初始化EurekaClient时，应用状态改变：从STARTING变为UP会触发这个Listener，调用instanceInfoReplicator.onDemandUpdate();
可以推测出，实例状态改变时，也会通过注册接口更新实例状态信息
statusChangeListener = new ApplicationInfoManager.StatusChangeListener() { @Override public String getId() { return "statusChangeListener"; } @Override public void notify(StatusChangeEvent statusChangeEvent) { if (InstanceStatus.DOWN == statusChangeEvent.getStatus() || InstanceStatus.DOWN == statusChangeEvent.getPreviousStatus()) { // log at warn level if DOWN was involved logger.warn("Saw local status change event {}", statusChangeEvent); } else { logger.info("Saw local status change event {}", statusChangeEvent); } instanceInfoReplicator.onDemandUpdate(); } };
- 定时任务，如果InstanceInfo发生改变，也会通过注册接口更新信息
public void run() { try { discoveryClient.refreshInstanceInfo(); //如果实例信息发生改变，则需要调用register更新InstanceInfo Long dirtyTimestamp = instanceInfo.isDirtyWithTime(); if (dirtyTimestamp != null) { discoveryClient.register(); instanceInfo.unsetIsDirty(dirtyTimestamp); } } catch (Throwable t) { logger.warn("There was a problem with the instance info replicator", t); } finally { Future next = scheduler.schedule(this, replicationIntervalSeconds, TimeUnit.SECONDS); scheduledPeriodicRef.set(next); } }
- 在定时renew时，如果renew接口返回404（代表这个实例在EurekaServer上面找不到），可能是之前注册失败或者注册过期导致的。这时需要调用register重新注册
boolean renew() { EurekaHttpResponse<InstanceInfo> httpResponse; try { httpResponse = eurekaTransport.registrationClient.sendHeartBeat(instanceInfo.getAppName(), instanceInfo.getId(), instanceInfo, null); logger.debug("{} - Heartbeat status: {}", PREFIX + appPathIdentifier, httpResponse.getStatusCode()); //如果renew接口返回404（代表这个实例在EurekaServer上面找不到），可能是之前注册失败或者注册过期导致的 if (httpResponse.getStatusCode() == 404) { REREGISTER_COUNTER.increment(); logger.info("{} - Re-registering apps/{}", PREFIX + appPathIdentifier, instanceInfo.getAppName()); long timestamp = instanceInfo.setIsDirtyWithTime(); boolean success = register(); if (success) { instanceInfo.unsetIsDirty(timestamp); } return success; } return httpResponse.getStatusCode() == 200; } catch (Throwable e) { logger.error("{} - was unable to send heartbeat!", PREFIX + appPathIdentifier, e); return false; } }
向Eureka发送注册请求EurekaServer发生了什么？
主要有两个存储，一个是之前提到过的registry，还有一个最近变化队列，后面我们会知道，这个最近变化队列里面就是客户端获取增量实例信息的内容：
# 整体注册信息缓存 private final ConcurrentHashMap<String, Map<String, Lease<InstanceInfo>>> registry = new ConcurrentHashMap<String, Map<String, Lease<InstanceInfo>>>(); # 最近变化队列 private ConcurrentLinkedQueue<RecentlyChangedItem> recentlyChangedQueue = new ConcurrentLinkedQueue<RecentlyChangedItem>();
EurekaServer收到实例注册主要分两步：
- 调用父类方法注册
- 同步到其他EurekaServer实例
public void register(InstanceInfo info, boolean isReplication) { int leaseDuration = 90; if (info.getLeaseInfo() != null && info.getLeaseInfo().getDurationInSecs() > 0) { leaseDuration = info.getLeaseInfo().getDurationInSecs(); } //调用父类方法注册 super.register(info, leaseDuration, isReplication); //同步到其他EurekaServer实例 this.replicateToPeers(PeerAwareInstanceRegistryImpl.Action.Register, info.getAppName(), info.getId(), info, (InstanceStatus)null, isReplication); }
我们先看同步到其他EurekaServer实例
其实就是，注册到的EurekaServer再依次调用其他集群内的EurekaServer的Register方法将实例信息同步过去
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; } 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 = 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); } }
然后看看调用父类方法注册：
public void register(InstanceInfo registrant, int leaseDuration, boolean isReplication) { try { //register虽然看上去好像是修改，但是这里用的是读锁，后面会解释 read.lock(); //从registry中查看这个app是否存在 Map<String, Lease<InstanceInfo>> gMap = registry.get(registrant.getAppName()); //不存在就创建 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; } } //查看这个app的这个实例是否已存在 Lease<InstanceInfo> existingLease = gMap.get(registrant.getId()); if (existingLease != null && (existingLease.getHolder() != null)) { //如果已存在，对比时间戳，保留比较新的实例信息...... } else { // 如果不存在，证明是一个新的实例 //更新自我保护监控变量的值的代码..... } Lease<InstanceInfo> lease = new Lease<InstanceInfo>(registrant, leaseDuration); if (existingLease != null) { lease.setServiceUpTimestamp(existingLease.getServiceUpTimestamp()); } //放入registry gMap.put(registrant.getId(), lease); //加入最近修改的记录队列 recentlyChangedQueue.add(new RecentlyChangedItem(lease)); //初始化状态，记录时间等相关代码...... //主动让Response缓存失效 invalidateCache(registrant.getAppName(), registrant.getVIPAddress(), registrant.getSecureVipAddress()); } finally { read.unlock(); } }
总结起来，就是主要三件事：
1.将实例注册信息放入或者更新registry
2.将实例注册信息加入最近修改的记录队列
3.主动让Response缓存失效
我们来类比下服务取消
服务取消CANCEL
protected boolean internalCancel(String appName, String id, boolean isReplication) { try { //cancel虽然看上去好像是修改，但是这里用的是读锁，后面会解释 read.lock(); //从registry中剔除这个实例 Map<String, Lease<InstanceInfo>> gMap = registry.get(appName); Lease<InstanceInfo> leaseToCancel = null; if (gMap != null) { leaseToCancel = gMap.remove(id); } if (leaseToCancel == null) { logger.warn("DS: Registry: cancel failed because Lease is not registered for: {}/{}", appName, id); return false; } else { //改变状态，记录状态修改时间等相关代码...... if (instanceInfo != null) { instanceInfo.setActionType(ActionType.DELETED); //加入最近修改的记录队列 recentlyChangedQueue.add(new RecentlyChangedItem(leaseToCancel)); } //主动让Response缓存失效 invalidateCache(appName, vip, svip); logger.info("Cancelled instance {}/{} (replication={})", appName, id, isReplication); return true; } } finally { read.unlock(); } }
总结起来，也是主要三件事：
1.从registry中剔除这个实例
2.将实例注册信息加入最近修改的记录队列
3.主动让Response缓存失效
这里我们注意到了这个最近修改队列，我们来详细看看
最近修改队列
这个最近修改队列和消费者定时获取服务实例列表有着密切的关系
private TimerTask getDeltaRetentionTask() { return new TimerTask() { @Override public void run() { Iterator<RecentlyChangedItem> it = recentlyChangedQueue.iterator(); while (it.hasNext()) { if (it.next().getLastUpdateTime() < System.currentTimeMillis() - serverConfig.getRetentionTimeInMSInDeltaQueue()) { it.remove(); } else { break; } } } }; }
这个RetentionTimeInMSInDeltaQueue默认是180s（配置是eureka.server.retention-time-in-m-s-in-delta-queue,默认是180s，官网写错了），可以看出这个队列是一个长度为180s的滑动窗口，保存最近180s以内的应用实例信息修改，后面我们会看到，客户端调用获取增量信息，实际上就是从这个queue中读取，所以可能一段时间内读取到的信息都是一样的。
