聊聊Eureka的心跳机制-好运工具

聊聊Eureka的心跳机制

2019-09-03

Eureka是Netflix公司开发的一套开源服务框架，后被引入到Spring的生态圈中，于是有了Spring Cloud Eureka。Spring Cloud Eureka是对原版的二次开发，再次基础上添加了Spring Boot的自动化配置，使用者可以通过简单的注解配置即可完成注册中心构建，实例的注册和服务的消费。但是这三点都不是本文要讲的，本文要说的是在实例注册后，实例通过心跳告诉注册中心自己还“活着”并为自己“续约”。

切入重点，也就是Eureka里面干活的类：

com.netflix.discovery.DiscoveryClient

这是Eureka的核心类，他的主要工作就是和Eureka服务端（即注册中心）进行交互，涉及到的操作有：服务注册，服务续约，服务删除。在DiscoveryClient的构造器中调用了如下方法，也就是续约操作存在地方。

    private void initScheduledTasks() {
        if (clientConfig.shouldFetchRegistry()) {
            // registry cache refresh timer
            int registryFetchIntervalSeconds = clientConfig.getRegistryFetchIntervalSeconds();
            int expBackOffBound = clientConfig.getCacheRefreshExecutorExponentialBackOffBound();
            scheduler.schedule(
                    new TimedSupervisorTask(
                            "cacheRefresh",
                            scheduler,
                            cacheRefreshExecutor,
                            registryFetchIntervalSeconds,
                            TimeUnit.SECONDS,
                            expBackOffBound,
                            new CacheRefreshThread()
                    ),
                    registryFetchIntervalSeconds, TimeUnit.SECONDS);
        }

        // 服务续约代码
        if (clientConfig.shouldRegisterWithEureka()) {
            int renewalIntervalInSecs = instanceInfo.getLeaseInfo().getRenewalIntervalInSecs();
            int expBackOffBound = clientConfig.getHeartbeatExecutorExponentialBackOffBound();
            logger.info("Starting heartbeat executor: " + "renew interval is: " + renewalIntervalInSecs);

            // Heartbeat timer
            scheduler.schedule(
                    new TimedSupervisorTask(
                            "heartbeat",
                            scheduler,
                            heartbeatExecutor,
                            renewalIntervalInSecs,
                            TimeUnit.SECONDS,
                            expBackOffBound,
                            new HeartbeatThread()
                    ),
                    renewalIntervalInSecs, TimeUnit.SECONDS);

            // InstanceInfo replicator
            instanceInfoReplicator = new InstanceInfoReplicator(
                    this,
                    instanceInfo,
                    clientConfig.getInstanceInfoReplicationIntervalSeconds(),
                    2); // burstSize

            statusChangeListener = new ApplicationInfoManager.StatusChangeListener() {
                @Override
                public String getId() {
                    return "statusChangeListener";
                }

                @Override
                public void notify(StatusChangeEvent statusChangeEvent) {
                    if (InstanceStatus.DOWN == statusChangeEvent.getStatus() ||
                            InstanceStatus.DOWN == statusChangeEvent.getPreviousStatus()) {
                        // log at warn level if DOWN was involved
                        logger.warn("Saw local status change event {}", statusChangeEvent);
                    } else {
                        logger.info("Saw local status change event {}", statusChangeEvent);
                    }
                    instanceInfoReplicator.onDemandUpdate();
                }
            };

            if (clientConfig.shouldOnDemandUpdateStatusChange()) {
                applicationInfoManager.registerStatusChangeListener(statusChangeListener);
            }

            instanceInfoReplicator.start(clientConfig.getInitialInstanceInfoReplicationIntervalSeconds());
        } else {
            logger.info("Not registering with Eureka server per configuration");
        }
    }

在上面的代码中可以看到两个scheduler，第一个是从服务端获取服务列表，我们将不细表述。我们说下第二个，也就是需要的Heartbeat timer。

定义renewalIntervalInSecs参数为心跳检测的时间，单位为秒。提交的任务是TimedSupervisorTask，执行时间延迟renewalIntervalInSecs。在任务的参数中有一个HeartbeatThread线程，这个用于服务续约。但是schedule只能延迟执行一次，他又是如何实现心跳的呢？关键就在TimedSupervisorTask里面了，我看下这个类的run方法。

    public void run() {
        Future future = null;
        try {
            future = executor.submit(task);
            threadPoolLevelGauge.set((long) executor.getActiveCount());
            future.get(timeoutMillis, TimeUnit.MILLISECONDS);  // block until done or timeout
            delay.set(timeoutMillis);
            threadPoolLevelGauge.set((long) executor.getActiveCount());
        } catch (TimeoutException e) {
            logger.error("task supervisor timed out", e);
            timeoutCounter.increment();

            long currentDelay = delay.get();
            long newDelay = Math.min(maxDelay, currentDelay * 2);
            delay.compareAndSet(currentDelay, newDelay);

        } catch (RejectedExecutionException e) {
            logger.error("task supervisor rejected the task", e);
            rejectedCounter.increment();
        } catch (Throwable e) {
            logger.error("task supervisor threw an exception", e);
            throwableCounter.increment();
        } finally {
            if (future != null) {
                future.cancel(true);
            }
            scheduler.schedule(this, delay.get(), TimeUnit.MILLISECONDS);
        }
    }

其中task指的就是前面提到的服务续约线程，返回的是Future对象，future.set(timeoutMillis, TimeUnit.MILLISECONDS)会将代码阻塞直到收到返回或者到了设置的超时时间。这里实现心跳的功能在finally里面，在将之前返回的future删除后，将会再次提交一个schedule，还是相同的延迟和相同的任务，有点类似于递归。就这样，客户端会每隔固定秒数发送“续约”心跳。