SpringCloud源码剖析-Eureka Server服务续约

@Produces({"application/xml", "application/json"})
publicclassInstanceResource {
    ...省略...
/**来自客户端实例的续订租约的put请求* A put request for renewing lease from a client instance.** @param isReplication*          请求头isreplicated 是否是从其他节点复制的*            a header parameter containing information whether this is*            replicated from other nodes.* @param overriddenStatus*          覆盖状态*            overridden status if any.* @param status ：续约的服务状态，一般都是UP*            the {@link InstanceStatus} of the instance.* @param lastDirtyTimestamp 此实例信息更新的最后时间戳*            last timestamp when this instance information was updated.指示操作是成功还是失败的响应。* @return response indicating whether the operation was a success or*         failure.*/@PUTpublicResponserenewLease(
@HeaderParam(PeerEurekaNode.HEADER_REPLICATION) StringisReplication,
@QueryParam("overriddenstatus") StringoverriddenStatus,
@QueryParam("status") Stringstatus,
@QueryParam("lastDirtyTimestamp") StringlastDirtyTimestamp) {
//是否是其他节点复制booleanisFromReplicaNode="true".equals(isReplication);
//执行续约流程booleanisSuccess=registry.renew(app.getName(), id, isFromReplicaNode);
// Not found in the registry, immediately ask for a registerif (!isSuccess) {
//没续约成功，返回NOT_FOUND状态logger.warn("Not Found (Renew): {} - {}", app.getName(), id);
returnResponse.status(Status.NOT_FOUND).build();
        }
// 检查是否需要根据最后更新间戳进行同步，客户端实例可能已更改了某些值// Check if we need to sync based on dirty time stamp, the client// instance might have changed some valueResponseresponse=null;
if (lastDirtyTimestamp!=null&&serverConfig.shouldSyncWhenTimestampDiffers()) {
response=this.validateDirtyTimestamp(Long.valueOf(lastDirtyTimestamp), isFromReplicaNode);
// Store the overridden status since the validation found out the node that replicates winsif (response.getStatus() ==Response.Status.NOT_FOUND.getStatusCode()
&& (overriddenStatus!=null)
&&!(InstanceStatus.UNKNOWN.name().equals(overriddenStatus))
&&isFromReplicaNode) {
//修改服务端维护的注册表中注册的服务的覆盖的状态registry.storeOverriddenStatusIfRequired(app.getAppName(), id, InstanceStatus.valueOf(overriddenStatus));
            }
        } else {
response=Response.ok().build();
        }
logger.debug("Found (Renew): {} - {}; reply status={}", app.getName(), id, response.getStatus());
returnresponse;
    }
}

@Overridepublicbooleanrenew(finalStringappName, finalStringserverId,
booleanisReplication) {
log("renew "+appName+" serverId "+serverId+", isReplication {}"+isReplication);
//获取服务端本地存储的服务注册表，//其实是从一个AbstractInstanceRegistry中名字叫 registry的map中获取//因为服务注册就是存储到这个registry的map中的List<Application>applications=getSortedApplications();
for (Applicationinput : applications) {
//找到当前续约服务名对应的服务if (input.getName().equals(appName)) {
InstanceInfoinstance=null;
for (InstanceInfoinfo : input.getInstances()) {
//找到服务名相同，服务id也相同的服务if (info.getId().equals(serverId)) {
instance=info;
break;
                }
            }
//抛出续约事件publishEvent(newEurekaInstanceRenewedEvent(this, appName, serverId,
instance, isReplication));
break;
        }
    }
//调用super的续约方法进行续约returnsuper.renew(appName, serverId, isReplication);
}

/** (non-Javadoc)** @see com.netflix.eureka.registry.InstanceRegistry#renew(java.lang.String,* java.lang.String, long, boolean)*/publicbooleanrenew(finalStringappName, finalStringid, finalbooleanisReplication) {
//调用super的续约方法if (super.renew(appName, id, isReplication)) {
//续约的实例信息复制到其他的Eureka节点//注意这里是传入的Action是Action.Heartbeat 心跳replicateToPeers(Action.Heartbeat, appName, id, null, null, isReplication);
returntrue;
    }
returnfalse;
}

/*** Marks the given instance of the given app name as renewed, and also marks whether it originated from* replication.** @see com.netflix.eureka.lease.LeaseManager#renew(java.lang.String, java.lang.String, boolean)*///appName应用名，id实例ID，isReplication是否是复制publicbooleanrenew(StringappName, Stringid, booleanisReplication) {
//增加给定统计信息的计数器，EurekaMonitors.counter加一，用作Eureka监控RENEW.increment(isReplication);
//从服务端维护的服务实例注册表中获取到当前续约服务名对应的服务信息Map<String, Lease<InstanceInfo>>gMap=registry.get(appName);
Lease<InstanceInfo>leaseToRenew=null;
if (gMap!=null) {
//获取服务的租约对象leaseToRenew=gMap.get(id);
    }
//如果没找到服务，返回falseif (leaseToRenew==null) {
//续约服务没找到的计数器增加RENEW_NOT_FOUND.increment(isReplication);
logger.warn("DS: Registry: lease doesn't exist, registering resource: {} - {}", appName, id);
returnfalse;
    } else {
//根据租约对象获取到服务的实例信息对象InstanceInfoInstanceInfoinstanceInfo=leaseToRenew.getHolder();
//如果服务实例信息不为空if (instanceInfo!=null) {
// touchASGCache(instanceInfo.getASGName());//获取服务的状态InstanceStatusoverriddenInstanceStatus=this.getOverriddenInstanceStatus(
instanceInfo, leaseToRenew, isReplication);
//如果服务的状态是UNKNOWN，说明服务可能被删除了，需要重新注册，返回falseif (overriddenInstanceStatus==InstanceStatus.UNKNOWN) {
logger.info("Instance status UNKNOWN possibly due to deleted override for instance {}"+"; re-register required", instanceInfo.getId());
//续约服务没找到的计数器增加RENEW_NOT_FOUND.increment(isReplication);
returnfalse;
            }
//实例状态与实例的OverriddenStatus覆盖实例状态不同。因此将状态设置为OverriddenStatus状态if (!instanceInfo.getStatus().equals(overriddenInstanceStatus)) {
logger.info(
"The instance status {} is different from overridden instance status {} for instance {}. "+"Hence setting the status to overridden status", instanceInfo.getStatus().name(),
instanceInfo.getOverriddenStatus().name(),
instanceInfo.getId());
//修改状态为overriddenInstanceStatus，instanceInfo.setStatusWithoutDirty(overriddenInstanceStatus);
            }
        }
//增加最后更新计数renewsLastMin.increment();
//执行续约，com.netflix.eureka.lease.Lease#renewleaseToRenew.renew();
//续约成功returntrue;
    }
}

privatevoidreplicateToPeers(Actionaction, StringappName, Stringid,
InstanceInfoinfo/* optional */,
InstanceStatusnewStatus/* optional */, booleanisReplication) {
Stopwatchtracer=action.getTimer().start();
try {
if (isReplication) {
numberOfReplicationsLastMin.increment();
        }
// If it is a replication already, do not replicate again as this will create a poison replicationif (peerEurekaNodes==Collections.EMPTY_LIST||isReplication) {
return;
        }
//循环除开自己的所有Eureka节点都要同步for (finalPeerEurekaNodenode : 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();
    }
}
/*** Replicates all instance changes to peer eureka nodes except for* replication traffic to this node.**/privatevoidreplicateInstanceActionsToPeers(Actionaction, StringappName,
Stringid, InstanceInfoinfo, InstanceStatusnewStatus,
PeerEurekaNodenode) {
try {
InstanceInfoinfoFromRegistry=null;
CurrentRequestVersion.set(Version.V2);
switch (action) {
caseCancel:
node.cancel(appName, id);
break;
caseHeartbeat:
//续约心跳，获取服务的状态InstanceStatusoverriddenStatus=overriddenInstanceStatusMap.get(id);
//根据名字和id获取实例InstanceInfoinfoFromRegistry=getInstanceByAppAndId(appName, id, false);
//调用PeerEurekaNode.heartbeat方法node.heartbeat(appName, id, infoFromRegistry, overriddenStatus, false);
break;
caseRegister:
node.register(info);
break;
caseStatusUpdate:
infoFromRegistry=getInstanceByAppAndId(appName, id, false);
node.statusUpdate(appName, id, newStatus, infoFromRegistry);
break;
caseDeleteStatusOverride:
infoFromRegistry=getInstanceByAppAndId(appName, id, false);
node.deleteStatusOverride(appName, id, infoFromRegistry);
break;
        }
    } catch (Throwablet) {
logger.error("Cannot replicate information to {} for action {}", node.getServiceUrl(), action.name(), t);
    }
}

/**发送心跳信息到Eureka节点，如果服务没注册，先注册，然后再发送* Send the heartbeat information of an instance to the node represented by* this class. If the instance does not exist the node, the instance* registration information is sent again to the peer node.** @param appName 服务名*            the application name of the instance.* @param id  服务实例id*            the unique identifier of the instance.* @param info 实例信息对象*            the instance info {@link InstanceInfo} of the instance.* @param overriddenStatus 服务状态*            the overridden status information if any of the instance.* @throws Throwable*/publicvoidheartbeat(finalStringappName, finalStringid,
finalInstanceInfoinfo, finalInstanceStatusoverriddenStatus,
booleanprimeConnection) throwsThrowable {
if (primeConnection) {
//根据调用传入的参数，这里是false，不会执行下面的代码，除非是走Aws亚马逊才是true// We do not care about the result for priming request.replicationClient.sendHeartBeat(appName, id, info, overriddenStatus);
return;
    }
//创建复制任务ReplicationTaskreplicationTask=newInstanceReplicationTask(targetHost, Action.Heartbeat, info, overriddenStatus, false) {
@OverridepublicEurekaHttpResponse<InstanceInfo>execute() throwsThrowable {
//执行心跳发送，参数：服务名，服务id，实例信息InstanceInfo ，服务状态returnreplicationClient.sendHeartBeat(appName, id, info, overriddenStatus);
        }
@OverridepublicvoidhandleFailure(intstatusCode, ObjectresponseEntity) throwsThrowable {
//如果发送失败super.handleFailure(statusCode, responseEntity);
if (statusCode==404) {
logger.warn("{}: missing entry.", getTaskName());
if (info!=null) {
logger.warn("{}: cannot find instance id {} and hence replicating the instance with status {}",
getTaskName(), info.getId(), info.getStatus());
//这里调用注册方法register(info);
                }
            } elseif (config.shouldSyncWhenTimestampDiffers()) {
InstanceInfopeerInstanceInfo= (InstanceInfo) responseEntity;
if (peerInstanceInfo!=null) {
syncInstancesIfTimestampDiffers(appName, id, info, peerInstanceInfo);
                }
            }
        }
    };
longexpiryTime=System.currentTimeMillis() +getLeaseRenewalOf(info);
//分批调度程序，执行任务batchingDispatcher.process(taskId("heartbeat", info), replicationTask, expiryTime);
}

@OverridepublicEurekaHttpResponse<InstanceInfo>sendHeartBeat(StringappName, Stringid, InstanceInfoinfo, InstanceStatusoverriddenStatus) {
//发送地址StringurlPath="apps/"+appName+'/'+id;
ClientResponseresponse=null;
try {
//封装http参数WebResourcewebResource=jerseyClient.getClient().resource(serviceUrl)
            .path(urlPath)
//状态            .queryParam("status", info.getStatus().toString())
//最后更新时间            .queryParam("lastDirtyTimestamp", info.getLastDirtyTimestamp().toString());
if (overriddenStatus!=null) {
webResource=webResource.queryParam("overriddenstatus", overriddenStatus.name());
        }
BuilderrequestBuilder=webResource.getRequestBuilder();
addExtraHeaders(requestBuilder);
//发送put请求response=requestBuilder.accept(MediaType.APPLICATION_JSON_TYPE).put(ClientResponse.class);
InstanceInfoinfoFromPeer=null;
if (response.getStatus() ==Status.CONFLICT.getStatusCode() &&response.hasEntity()) {
infoFromPeer=response.getEntity(InstanceInfo.class);
        }
returnanEurekaHttpResponse(response.getStatus(), infoFromPeer).type(MediaType.APPLICATION_JSON_TYPE).build();
    } finally {
if (logger.isDebugEnabled()) {
logger.debug("[heartbeat] Jersey HTTP PUT {}; statusCode={}", urlPath, response==null?"N/A" : response.getStatus());
        }
if (response!=null) {
response.close();
        }
    }
}