【原创】Eureka源码剖析之五：服务下线

/**
 * 注销服务，调用client的cancel服务，往里面看也就是调用了服务端的http delete 请求进行服务下线
 */
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);
        }
    }
}

@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();
        }
    }
}

@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();

         // 服务实例以及被注册，那么设置实例状态为DOWN，并且进行注销操作
         if (applicationInfoManager != null && clientConfig.shouldRegisterWithEureka()) {
             applicationInfoManager.setInstanceStatus(InstanceStatus.DOWN);
             unregister();
         }
         // 关闭eurekaTransport
         if (eurekaTransport != null) {
             eurekaTransport.shutdown();
         }
         // 关闭监控
         heartbeatStalenessMonitor.shutdown();
         registryStalenessMonitor.shutdown();

         logger.info("Completed shut down of DiscoveryClient");
     }
 }

 // DiscoveryManager调用了DiscveryClient的shutdown方法，这里就是服务下线的入口
 public void shutdownComponent() {
     if (discoveryClient != null) {
         try {
             discoveryClient.shutdown();
             discoveryClient = null;
         } catch (Throwable th) {
             logger.error("Error in shutting down client", th);
         }
     }
 }

// 服务端提供的对外服务下线接口
@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();
    }
}

// 服务端下线入口
@Override
public boolean cancel(final String appName, final String id,
                      final boolean isReplication) {
    // 调用父类cancel方法，将实例信息添加到最近下线队列、最近变更队列，并且使本地缓存失效操作
    if (super.cancel(appName, id, isReplication)) {
        // 服务端集群之间进行Cancel同步操作
        replicateToPeers(Action.Cancel, appName, id, null, null, isReplication);
        synchronized (lock) {
            if (this.expectedNumberOfRenewsPerMin > 0) {
                // Since the client wants to cancel it, reduce the threshold (1 for 30 seconds, 2 for a minute)
                this.expectedNumberOfRenewsPerMin = this.expectedNumberOfRenewsPerMin - 2;
                this.numberOfRenewsPerMinThreshold =
                        (int) (this.expectedNumberOfRenewsPerMin * serverConfig.getRenewalPercentThreshold());
            }
        }
        return true;
    }
    return false;
}

// 最终调用这个方法进行实例信息移除
protected boolean internalCancel(String appName, String id, boolean isReplication) {
    try {
        // 读锁
        read.lock();
        // 取消（下线）计数
        CANCEL.increment(isReplication);
        // 获取续约实例
        Map<String, Lease<InstanceInfo>> gMap = registry.get(appName);
        Lease<InstanceInfo> leaseToCancel = null;
        if (gMap != null) {
            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) {
            // 续约实例不存在，返回false
            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) {
                // 设置实例行为为delete
                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);
            return true;
        }
    } finally {
        // 释放读锁
        read.unlock();
    }
}