Eureka源码采用1.7.2版本
本人小白，此文为本人阅读源码笔记，如果您读到本文，您需要自己甄别是否正确，文中的说明只代表本人理解，不一定是正确的！！！
这里说的服务下线是手动调用shutDown()方法正常下线，不是说宕机，网络故障造成的异常下线。
com.netflix.discovery.DiscoveryClient#shutdown
@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()) {
//将服务实例状态设置为DOWM
applicationInfoManager.setInstanceStatus(InstanceStatus.DOWN);
//发送服务下线请求
unregister();
}
//释放通信资源
if (eurekaTransport != null) {
eurekaTransport.shutdown();
}
//释放监控资源
heartbeatStalenessMonitor.shutdown();
registryStalenessMonitor.shutdown();
logger.info("Completed shut down of DiscoveryClient");
}
}
shutdown()方法主要干了这几件事情：
- 将当前服务实例的状态设置为DOWN，并请求服务注册中心
- 停止当前客户端的调度任务资源
- 释放通信资源
- 释放监控资源
com.netflix.discovery.shared.transport.jersey.AbstractJerseyEurekaHttpClient#cancel
@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();
}
}
}
服务的下线http请求地址http://DESKTOP-1S6DCTA:8080/v2/apps/EUREKA/001，请求方法为delete
服务端接收改请求处理的核心方法：
com.netflix.eureka.registry.PeerAwareInstanceRegistryImpl#cancel
@Override
public boolean cancel(final String appName, final String id,
final boolean isReplication) {
if (super.cancel(appName, id, isReplication)) {
//集群节点同步下线消息
replicateToPeers(Action.Cancel, appName, id, null, null, isReplication);
//期望每分钟心跳数量-2
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;
}
这里调用了父类的节点移除方法，通知了其他集群节点服务下线，并更新了期望每分钟最小心跳数
看看父类下线的核心方法
com.netflix.eureka.registry.AbstractInstanceRegistry#internalCancel
//服务下线
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);
}
//将服务实例加入最近下线队列中
synchronized (recentCanceledQueue) {
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();
}
//删除读写缓存中该服务的实例，只读缓存最坏情况下需要30S才能进行同步啊
invalidateCache(appName, vip, svip);
logger.info("Cancelled instance {}/{} (replication={})", appName, id, isReplication);
return true;
}
} finally {
read.unlock();
}
}
这里我们需要梳理下，在服务下线中主要干了哪些事情
- 从当前注册表中依据服务名称，服务ID，移除对应的租约
- 将该服务实例加入到最近下线队列中 这个还没研究，不知道干嘛用的
- 设置该服务实力的驱逐时间，其实就是evictionTimestamp = System.currentTimeMillis();
- 设置该服务实例的Action为DELETE，并加入到最近修改队列中
- 设置更新时间
- 清除读写缓存中该服务实例对应的数据
<font color='red'>从上面几步看出，主要是多级缓存和注册表的清除，并加入最近修改队列中去，便于客户端进行增量拉取时候，快速感知到该服务实例下线</font>
这里涉及到的读写缓存的过期，具体的去看看多级缓存的架构
