【一起学源码-微服务】Nexflix Eureka 源码十一：EurekaServer自我保护机制竟然有这么多Bug？

public void evict(long additionalLeaseMs) {
    logger.debug("Running the evict task");

    // 是否允许主动删除宕机节点数据，这里判断是否进入自我保护机制，如果是自我保护了则不允许摘除服务
    if (!isLeaseExpirationEnabled()) {
        logger.debug("DS: lease expiration is currently disabled.");
        return;
    }

    // 省略服务摘除等等操作...
}

public boolean isLeaseExpirationEnabled() {
    if (!isSelfPreservationModeEnabled()) {
        // The self preservation mode is disabled, hence allowing the instances to expire.
        return true;
    }

    // 这行代码触发自我保护机制，期望的一分钟要有多少次心跳发送过来，所有服务实例一分钟得发送多少次心跳
    // getNumOfRenewsInLastMin 上一分钟所有服务实例一共发送过来多少心跳，10次
    // 如果上一分钟 的心跳次数太少了（20次）< 我期望的100次，此时会返回false
    return numberOfRenewsPerMinThreshold > 0 && getNumOfRenewsInLastMin() > numberOfRenewsPerMinThreshold;
}

private final MeasuredRate renewsLastMin;

public long getNumOfRenewsInLastMin() {
    return renewsLastMin.getCount();
}

public class MeasuredRate {
    private static final Logger logger = LoggerFactory.getLogger(MeasuredRate.class);
    private final AtomicLong lastBucket = new AtomicLong(0);
    private final AtomicLong currentBucket = new AtomicLong(0);

    private final long sampleInterval;
    private final Timer timer;

    private volatile boolean isActive;

    /**
     * @param sampleInterval in milliseconds
     */
    public MeasuredRate(long sampleInterval) {
        this.sampleInterval = sampleInterval;
        this.timer = new Timer("Eureka-MeasureRateTimer", true);
        this.isActive = false;
    }

    public synchronized void start() {
        if (!isActive) {
            timer.schedule(new TimerTask() {

                @Override
                public void run() {
                    try {
                        // Zero out the current bucket.
                        // renewsLastMin 为1分钟
                        // 每分钟调度一次，将当前的88次总心跳设置到lastBucket中去，然后将当前的currentBucket 设置为0 秒啊！
                        lastBucket.set(currentBucket.getAndSet(0));
                    } catch (Throwable e) {
                        logger.error("Cannot reset the Measured Rate", e);
                    }
                }
            }, sampleInterval, sampleInterval);

            isActive = true;
        }
    }

    public synchronized void stop() {
        if (isActive) {
            timer.cancel();
            isActive = false;
        }
    }

    /**
     * Returns the count in the last sample interval.
     */
    public long getCount() {
        return lastBucket.get();
    }

    /**
     * Increments the count in the current sample interval.
     */
    public void increment() {
        // 心跳次数+1 例如说1分钟所有服务实例共发起了88次心跳
        currentBucket.incrementAndGet();
    }
}

private void updateRenewalThreshold() {
    try {
        // count为注册表中服务实例的个数
        // 将自己作为eureka client，从其他eureka server拉取注册表
        // 合并到自己本地去 将从别的eureka server拉取到的服务实例的数量作为count
        Applications apps = eurekaClient.getApplications();
        int count = 0;
        for (Application app : apps.getRegisteredApplications()) {
            for (InstanceInfo instance : app.getInstances()) {
                if (this.isRegisterable(instance)) {
                    ++count;
                }
            }
        }
        synchronized (lock) {
            // Update threshold only if the threshold is greater than the
            // current expected threshold of if the self preservation is disabled.
            // 这里也是存在bug的，master分支已经修复
            // 一分钟服务实例心跳个数(其他eureka server拉取的服务实例个数 * 2) > 自己本身一分钟所有服务实例实际心跳次数 * 0.85(阈值)
            // 这里主要是跟其他的eureka server去做一下同步
            if ((count * 2) > (serverConfig.getRenewalPercentThreshold() * numberOfRenewsPerMinThreshold)
                    || (!this.isSelfPreservationModeEnabled())) {
                this.expectedNumberOfRenewsPerMin = count * 2;
                this.numberOfRenewsPerMinThreshold = (int) ((count * 2) * serverConfig.getRenewalPercentThreshold());
            }
        }
        logger.info("Current renewal threshold is : {}", numberOfRenewsPerMinThreshold);
    } catch (Throwable e) {
        logger.error("Cannot update renewal threshold", e);
    }
}

public int isBelowRenewThresold() {
    if ((getNumOfRenewsInLastMin() <= numberOfRenewsPerMinThreshold)
            &&
            ((this.startupTime > 0) && (System.currentTimeMillis() > this.startupTime + (serverConfig.getWaitTimeInMsWhenSyncEmpty())))) {
        return 1;
    } else {
        return 0;
    }
}

this.expectedNumberOfRenewsPerMin = count * 2;
// numberOfRenewsPerMinThreshold = count * 2 * 0.85 = 34 期望一分钟 20个服务实例，得有34个心跳
this.numberOfRenewsPerMinThreshold =
        (int) (this.expectedNumberOfRenewsPerMin * serverConfig.getRenewalPercentThreshold());