美文网首页
Eureka系列(五) 服务续约流程具体实现

Eureka系列(五) 服务续约流程具体实现

作者: 偷吃虾的猫 | 来源:发表于2020-04-05 16:03 被阅读0次

    服务续约执行简要流程图

      下面这张图大致描述了服务续约从Client端到Server端的大致流程,详情如下: Eureka 服务续约.jpg

    服务续约Client源码分析

      我们先来看看服务续约定时任务的初始化。那我们的服务续约定时任务什么时候会被初始化呢,那肯定是我们启用我们Eureka Client的时候,当我们启动Client时,Eureka会先处理相关的配置,然后初始化我们Client的相关信息,我们的定时任务也就是此时进行的初始化,具体来说我们的服务续约定时任务就是在DiscoveryClient这个类中initScheduledTasks方法中被初始化的。代码如下:

    private final ScheduledExecutorService scheduler;
    private void initScheduledTasks() {
         …省略其他代码
        // 初始化定时服务续约任务
        scheduler.schedule(
            new TimedSupervisorTask(
                    "heartbeat",
                    scheduler,
                    heartbeatExecutor,
                    renewalIntervalInSecs,
                    TimeUnit.SECONDS,
                    expBackOffBound,
                    new HeartbeatThread()
            ),
            renewalIntervalInSecs, TimeUnit.SECONDS);
         …省略其他代码   
    }
    

      由此可见,我们的定时任务其实是Client进行初始化完成的,并且还是使用ScheduledExecutorService线程池来完成我们的定时任务。接下来我们看看HeartbeatThread类相关的源码:

    private class HeartbeatThread implements Runnable {
        public void run() {
            if (renew()) {
                lastSuccessfulHeartbeatTimestamp = System.currentTimeMillis();
            }
        }
    }
    

      HeartbeatThread类中run方法调用了renew方法,我们接着看下去:

    /**
    * Renew with the eureka service by making the appropriate REST call
    */
    boolean renew() {
        EurekaHttpResponse<InstanceInfo> httpResponse;
        try {
            // 调用Eureka Server提供的服务续约Http接口
            httpResponse = eurekaTransport.registrationClient.sendHeartBeat(instanceInfo.getAppName(), instanceInfo.getId(), instanceInfo, null);
            logger.debug(PREFIX + "{} - Heartbeat status: {}", appPathIdentifier, httpResponse.getStatusCode());
            if (httpResponse.getStatusCode() == 404) {
                REREGISTER_COUNTER.increment();
                logger.info(PREFIX + "{} - Re-registering apps/{}", appPathIdentifier, instanceInfo.getAppName());
                long timestamp = instanceInfo.setIsDirtyWithTime();
                boolean success = register();
                if (success) {
                    instanceInfo.unsetIsDirty(timestamp);
                }
                return success;
            }
            return httpResponse.getStatusCode() == 200;
        } catch (Throwable e) {
            logger.error(PREFIX + "{} - was unable to send heartbeat!", appPathIdentifier, e);
            return false;
        }
    }
    

       通过eurekaTransport.registrationClient.sendHeartBeat(instanceInfo.getAppName(), instanceInfo.getId(), instanceInfo, null) 这行代码,我们则可以知道Client端的续约其实就是用调用Server提供的接口,然后Server端实现了服务续约的操作。


    服务续约Server源码实现

      由上面可见,我们Client的续约操作是直接调用的Server端提供的接口,以此来实现的具体操作,那我们接下来就看看Server是如何进行服务续约操作的,源码如下:

    @PUT
    public Response renewLease(
                @HeaderParam(PeerEurekaNode.HEADER_REPLICATION) String isReplication,
                @QueryParam("overriddenstatus") String overriddenStatus,
                @QueryParam("status") String status,
                @QueryParam("lastDirtyTimestamp") String lastDirtyTimestamp) {
            boolean isFromReplicaNode = "true".equals(isReplication);
            // 进行服务续约操作
            boolean isSuccess = registry.renew(app.getName(), id, isFromReplicaNode);
    
            // Not found in the registry, immediately ask for a register
            if (!isSuccess) {
                logger.warn("Not Found (Renew): {} - {}", app.getName(), id);
                return Response.status(Status.NOT_FOUND).build();
            }
            // Check if we need to sync based on dirty time stamp, the client
            // instance might have changed some value
            Response response = 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 wins
                if (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());
            return response;
        }
    

      由registry.renew(app.getName(), id, isFromReplicaNode);这里我们可以看到调用了renew()方法来实现续约操作,那我们就看下这个方法的实现:

    Override
        public boolean renew(final String appName, final String serverId,
                boolean isReplication) {
            log("renew " + appName + " serverId " + serverId + ", isReplication {}"
                    + isReplication);
            List<Application> applications = getSortedApplications();
            for (Application input : applications) {
                if (input.getName().equals(appName)) {
                    InstanceInfo instance = null;
                    for (InstanceInfo info : input.getInstances()) {
                        if (info.getId().equals(serverId)) {
                            instance = info;
                            break;
                        }
                    }
                    publishEvent(new EurekaInstanceRenewedEvent(this, appName, serverId,
                            instance, isReplication));
                    break;
                }
            }
            return super.renew(appName, serverId, isReplication);
        }
    

      我们可以看到方法最后调用了父类的renew()方法,我们继续看下去:

    public boolean renew(final String appName, final String id, final boolean isReplication) {
            if (super.renew(appName, id, isReplication)) {
                // 向其他server发送服务续约消息
                replicateToPeers(Action.Heartbeat, appName, id, null, null, isReplication);
                return true;
            }
            return false;
        }
    
     public boolean renew(String appName, String id, boolean isReplication) {
            RENEW.increment(isReplication);
            Map<String, Lease<InstanceInfo>> gMap = registry.get(appName);
            Lease<InstanceInfo> leaseToRenew = null;
            if (gMap != null) {
                leaseToRenew = gMap.get(id);
            }
            if (leaseToRenew == null) {
                RENEW_NOT_FOUND.increment(isReplication);
                logger.warn("DS: Registry: lease doesn't exist, registering resource: {} - {}", appName, id);
                return false;
            } else {
                InstanceInfo instanceInfo = leaseToRenew.getHolder();
                if (instanceInfo != null) {
                    // touchASGCache(instanceInfo.getASGName());
                    InstanceStatus overriddenInstanceStatus = this.getOverriddenInstanceStatus(
                            instanceInfo, leaseToRenew, isReplication);
                    if (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);
                        return false;
                    }
                    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());
                        instanceInfo.setStatusWithoutDirty(overriddenInstanceStatus);
    
                    }
                }
                renewsLastMin.increment();
                leaseToRenew.renew();
                return true;
            }
        }
    
    public void renew() {
            //  更新上次更新时间
            lastUpdateTimestamp = System.currentTimeMillis() + duration;
    }
    

      由上面的代码可见,Eureka Server服务续约的流程可大致分为以下几步:
        1.更新实例状态
        2.更新实例上次更新时间
        3.向其他Server发送服务续约消息


      下面为自己总结的Eureka相关的知识点,有兴趣地小伙伴可以看一看,当然再点下赞就更棒了,创作不易!
      Eureka系列(一)Eureka功能介绍
      Eureka系列(二) 服务注册Server端具体实现
      Eureka系列(三)获取服务Client端具体实现
      Eureka系列(四) 获取服务Server端具体实现
      Eureka系列(五) 服务续约流程具体实现
      Eureka系列(六) TimedSupervisorTask类解析
      Eureka系列(七) 服务下线Server端具体实现
      Eureka系列(八)服务剔除具体实现
      Eureka系列(九)Eureka自我保护机制

    相关文章

      网友评论

          本文标题:Eureka系列(五) 服务续约流程具体实现

          本文链接:https://www.haomeiwen.com/subject/bpmuphtx.html