Eureka 下线

Eureka 下线

// DiscoveryClient.java
   @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
            if (applicationInfoManager != null
                    && clientConfig.shouldRegisterWithEureka()
                    && clientConfig.shouldUnregisterOnShutdown()) {
                //讲服务实例的状态设置为DOWN
                applicationInfoManager.setInstanceStatus(InstanceStatus.DOWN);
                //取消注册
                unregister();
            }
 
            //关闭网络通信组件
            if (eurekaTransport != null) {
                eurekaTransport.shutdown();
            }
 
            //关闭监听器
            heartbeatStalenessMonitor.shutdown();
            registryStalenessMonitor.shutdown();
 
            Monitors.unregisterObject(this);
 
            logger.info("Completed shut down of DiscoveryClient");
        }
    }

• 调用 ApplicationInfoManager#setInstanceStatus(...) 方法，设置应用实例为关闭( DOWN )。
• 调用 #unregister() 方法，实现代码如下：

// DiscoveryClient.java
void unregister() {
   // It can be null if shouldRegisterWithEureka == false
   if(eurekaTransport != null && eurekaTransport.registrationClient != null) {
       try {
           logger.info("Unregistering ...");
           EurekaHttpResponse<Void> httpResponse = eurekaTransport.registrationClient.cancel(instanceInfo.getAppName(), instanceInfo.getId());
           logger.info(PREFIX + appPathIdentifier + " - deregister  status: " + httpResponse.getStatusCode());
       } catch (Exception e) {
           logger.error(PREFIX + appPathIdentifier + " - de-registration failed" + e.getMessage(), e);
       }
   }
}
 
// AbstractJerseyEurekaHttpClient.java
@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();
        }
    }
}

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

com.netflix.eureka.resources.InstanceResource，处理单个应用实例信息的请求操作的 Resource ( Controller )。

下线应用实例信息的请求，映射 InstanceResource#cancelLease() 方法，实现代码如下：

@DELETE
public Response cancelLease(
       @HeaderParam(PeerEurekaNode.HEADER_REPLICATION) String isReplication) {
   // 下线
   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();
   }
}

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

@Override
    public boolean cancel(String appName, String id, boolean isReplication) {
        return internalCancel(appName, id, isReplication);
    }
 
    /**
     * {@link #cancel(String, String, boolean)} method is overridden by {@link PeerAwareInstanceRegistry}, so each
     * cancel request is replicated to the peers. This is however not desired for expires which would be counted
     * in the remote peers as valid cancellations, so self preservation mode would not kick-in.
     */
    protected boolean internalCancel(String appName, String id, boolean isReplication) {
        read.lock();
        try {
            CANCEL.increment(isReplication);
            Map<String, Lease<InstanceInfo>> gMap = registry.get(appName);
            Lease<InstanceInfo> leaseToCancel = null;
            if (gMap != null) {
                leaseToCancel = gMap.remove(id);
            }
 
            //将服务实例加入最近下线的queue中
            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 {
                //调用cancel，保存一个时间戳
                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();
 
                    //服务的注册，下线，过期，都会代表这个服务实例变化了，都会将自己放入最近改变的队列中
                    //这个最近改变的队列，只会保留最近3分钟的实例
                    //所以说eureka client拉取增量注册表的时候，其实就是拉取最近3分钟有变化的服务实例。
                }
                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;
    }