Spring Cloud——Eureka服务实例下线

应用服务实例下线

1、Eureka Client发起下线

应用实例关闭时，Eureka Client向Eureka Server发起下线应用实例。需要满足如下条件才可发起：

• 配置eureka.registration.enabled=true，应用实例开启注册开关。默认为false
• 配置eureka.shouldUnregisterOnShutdown=true，应用实例开启关闭时下线开关。默认为true

Eureka Client端有一个shutdown方法，服务实例下线的时候，会主动调用这个方法。

@Singleton public class DiscoveryClient implements EurekaClient { /** * 关闭Eureka客户端。还将注销请求发送到Eureka服务器 */ @PreDestroy @Override public synchronized void shutdown() { if (isShutdown.compareAndSet(false, true)) { logger.info("Shutting down DiscoveryClient ..."); if (statusChangeListener != null && applicationInfoManager != null) { applicationInfoManager.unregisterStatusChangeListener(statusChangeListener.getId()); } //将定时调度任务都停止 cancelScheduledTasks(); // If APPINFO was registered //如果注册到了Eureka Server，通知Server下线 if (applicationInfoManager != null && clientConfig.shouldRegisterWithEureka() && clientConfig.shouldUnregisterOnShutdown()) { applicationInfoManager.setInstanceStatus(InstanceStatus.DOWN); //服务下线 unregister(); } if (eurekaTransport != null) { eurekaTransport.shutdown(); } heartbeatStalenessMonitor.shutdown(); registryStalenessMonitor.shutdown(); Monitors.unregisterObject(this); logger.info("Completed shut down of DiscoveryClient"); } } void unregister() { // It can be null if shouldRegisterWithEureka == false if(eurekaTransport != null && eurekaTransport.registrationClient != null) { try { logger.info("Unregistering ..."); // 调用AbstractJerseyEurekaHttpClient的cancel方法 EurekaHttpResponse<Void> httpResponse = eurekaTransport.registrationClient.cancel(instanceInfo.getAppName(), instanceInfo.getId()); logger.info(PREFIX + "{} - deregister status: {}", appPathIdentifier, httpResponse.getStatusCode()); } catch (Exception e) { logger.error(PREFIX + "{} - de-registration failed{}", appPathIdentifier, e.getMessage(), e); } } } }

将定时调度任务都停止——cancelScheduledTasks()

cancelScheduledTasks方法中会将client端启动的定时任务都停止，包括：心跳续约，定时拉取增量注册表等等。

@Singleton public class DiscoveryClient implements EurekaClient { private void cancelScheduledTasks() { if (instanceInfoReplicator != null) { instanceInfoReplicator.stop(); } if (heartbeatExecutor != null) { heartbeatExecutor.shutdownNow(); } if (cacheRefreshExecutor != null) { cacheRefreshExecutor.shutdownNow(); } if (scheduler != null) { scheduler.shutdownNow(); } if (cacheRefreshTask != null) { cacheRefreshTask.cancel(); } if (heartbeatTask != null) { heartbeatTask.cancel(); } } }

AbstractJerseyEurekaHttpClient的cancel()方法使用DELETE请求调用Eureka Server的apps/${APP_NAME}/${INSTANCE_INFO_ID}接口，实现应用实例信息的下线

public abstract class AbstractJerseyEurekaHttpClient implements EurekaHttpClient { @Override public EurekaHttpResponse<Void> cancel(String appName, String id) { String urlPath = "apps/" + appName + '/' + id; ClientResponse response = null; try { Builder resourceBuilder = jerseyClient.resource(serviceUrl).path(urlPath).getRequestBuilder(); addExtraHeaders(resourceBuilder); response = resourceBuilder.delete(ClientResponse.class); return anEurekaHttpResponse(response.getStatus()).headers(headersOf(response)).build(); } finally { if (logger.isDebugEnabled()) { logger.debug("Jersey HTTP DELETE {}/{}; statusCode={}", serviceUrl, urlPath, response == null ? "N/A" : response.getStatus()); } if (response != null) { response.close(); } } } }

2、Eureka Server接收下线

Eureka Server接收下线请求核心流程如下图：

1）、接收下线请求

@Produces({"application/xml", "application/json"}) public class InstanceResource { private final PeerAwareInstanceRegistry registry; @DELETE public Response cancelLease( @HeaderParam(PeerEurekaNode.HEADER_REPLICATION) String isReplication) { try { // 服务下线 boolean isSuccess = registry.cancel(app.getName(), id, "true".equals(isReplication)); if (isSuccess) { logger.debug("Found (Cancel): {} - {}", app.getName(), id); return Response.ok().build(); } else { logger.info("Not Found (Cancel): {} - {}", app.getName(), id); return Response.status(Status.NOT_FOUND).build(); } } catch (Throwable e) { logger.error("Error (cancel): {} - {}", app.getName(), id, e); return Response.serverError().build(); } } } @Singleton public class PeerAwareInstanceRegistryImpl extends AbstractInstanceRegistry implements PeerAwareInstanceRegistry { @Override public boolean cancel(final String appName, final String id, final boolean isReplication) { // 调用父类方法下线服务实例 if (super.cancel(appName, id, isReplication)) { // Eureka Server复制 replicateToPeers(Action.Cancel, appName, id, null, null, isReplication); return true; } return false; } }

PeerAwareInstanceRegistryImpl中调用了父类AbstractInstanceRegistry的cancel(...)方法下线应用实例信息。

2）、下线应用实例信息

调用了AbstractInstanceRegistry的cancel(...)方法，下线应用实例信息，代码如下：

public abstract class AbstractInstanceRegistry implements InstanceRegistry { @Override public boolean cancel(String appName, String id, boolean isReplication) { return internalCancel(appName, id, isReplication); } protected boolean internalCancel(String appName, String id, boolean isReplication) { try { // 获得读锁 read.lock(); // 增加取消注册次数到监控 CANCEL.increment(isReplication); // 获取appName 对应的所有实例集合 Map<String, Lease<InstanceInfo>> gMap = registry.get(appName); Lease<InstanceInfo> leaseToCancel = null; // 移除租约映射 if (gMap != null) { // 根据实例id 移除对应的实例租约信息 leaseToCancel = gMap.remove(id); } // 添加到最近取消注册的队列，用于统计 recentCanceledQueue.add(new Pair<Long, String>(System.currentTimeMillis(), appName + "(" + id + ")")); // 移除应用实例覆盖状态映射 InstanceStatus instanceStatus = overriddenInstanceStatusMap.remove(id); if (instanceStatus != null) { logger.debug("Removed instance id {} from the overridden map which has value {}", id, instanceStatus.name()); } // 租约不存在 if (leaseToCancel == null) { // 添加取消注册不存在到监控 CANCEL_NOT_FOUND.increment(isReplication); logger.warn("DS: Registry: cancel failed because Lease is not registered for: {}/{}", appName, id); return false; } else { // 设置租约的取消注册时间戳 leaseToCancel.cancel(); // 添加到最近租约变更记录队列 InstanceInfo instanceInfo = leaseToCancel.getHolder(); String vip = null; String svip = null; if (instanceInfo != null) { //放入最近改变队列中 instanceInfo.setActionType(ActionType.DELETED); recentlyChangedQueue.add(new RecentlyChangedItem(leaseToCancel)); instanceInfo.setLastUpdatedTimestamp(); vip = instanceInfo.getVIPAddress(); svip = instanceInfo.getSecureVipAddress(); } // 设置响应缓存过期 invalidateCache(appName, vip, svip); logger.info("Cancelled instance {}/{} (replication={})", appName, id, isReplication); } } finally { // 释放锁 read.unlock(); } // 更新 需要发送心跳的客户端数量 synchronized (lock) { if (this.expectedNumberOfClientsSendingRenews > 0) { // Since the client wants to cancel it, reduce the number of clients to send renews. this.expectedNumberOfClientsSendingRenews = this.expectedNumberOfClientsSendingRenews - 1; updateRenewsPerMinThreshold(); } } return true; } }

更新需要发送续约信息客户端数量，毕竟是服务下线，就少了一个需要发送续约信息的客户端，更新自我保护机制触发阈值，这块涉及到Eureka Server 自我保护机制实现原理。

public class Lease<T> { enum Action { Register, Cancel, Renew }; private long evictionTimestamp; public void cancel() { if (evictionTimestamp <= 0) { // 设置取消注册时间戳 evictionTimestamp = System.currentTimeMillis(); } } }

方法逻辑不复杂，有四步：

• 1、从原始注册表数据结构中，删除下线的实例
• 2、对实例执行下线的逻辑，其实没什么逻辑，就是保存了一下下线的时间戳evictionTimestamp
• 3、放入最近变更队列中，以便client拉取增量注册表的时候可以感知到有服务实例下线了
• 4、清空读写缓存，避免client拉取缓存时拉取到已经下线的服务实例

服务下线不会主动通知其他客户端，而是等其他客户端来拉取增量注册表的时候，才会感知到。

 

参考： https://blog.csdn.net/qq_40378034/article/details/119079180