Eureka 的 InstanceInfoReplicator 类（服务注册辅助工具）

//略去不相关代码...//实例化InstanceInfoReplicator对象instanceInfoReplicator = new InstanceInfoReplicator(                    this,                    instanceInfo,                    clientConfig.getInstanceInfoReplicationIntervalSeconds(),                    2); // burstSize
            //监听器，用来监听作为Eureka client的自身的状态变化            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);                    }                    //状态变化时notify方法会被执行，此时上报最新状态到Eureka server                    instanceInfoReplicator.onDemandUpdate();                }            };
            if (clientConfig.shouldOnDemandUpdateStatusChange()) {                //注册监听器                applicationInfoManager.registerStatusChangeListener(statusChangeListener);            }            //服务注册            instanceInfoReplicator.start(clientConfig.getInitialInstanceInfoReplicationIntervalSeconds());

/** * A task for updating and replicating the local instanceinfo to the remote server. Properties of this task are: * - configured with a single update thread to guarantee sequential update to the remote server * - update tasks can be scheduled on-demand via onDemandUpdate() * - task processing is rate limited by burstSize * - a new update task is always scheduled automatically after an earlier update task. However if an on-demand task *   is started, the scheduled automatic update task is discarded (and a new one will be scheduled after the new *   on-demand update). * *   @author dliu */

InstanceInfoReplicator(DiscoveryClient discoveryClient, InstanceInfo instanceInfo, int replicationIntervalSeconds, int burstSize) {        this.discoveryClient = discoveryClient;        this.instanceInfo = instanceInfo;    //线程池，core size为1，使用DelayedWorkQueue队列        this.scheduler = Executors.newScheduledThreadPool(1,                new ThreadFactoryBuilder()                        .setNameFormat("DiscoveryClient-InstanceInfoReplicator-%d")                        .setDaemon(true)                        .build());
        this.scheduledPeriodicRef = new AtomicReference<Future>();
        this.started = new AtomicBoolean(false);        //RateLimiter是个限制频率的工具类，用来限制单位时间内的任务次数        this.rateLimiter = new RateLimiter(TimeUnit.MINUTES);        this.replicationIntervalSeconds = replicationIntervalSeconds;        this.burstSize = burstSize;        //通过周期间隔，和burstSize参数，计算每分钟允许的任务数        this.allowedRatePerMinute = 60 * this.burstSize / this.replicationIntervalSeconds;        logger.info("InstanceInfoReplicator onDemand update allowed rate per min is {}", allowedRatePerMinute);    }

public void start(int initialDelayMs) {    //CAS操作，不但保证了只执行一次，多线程场景也能保证    if (started.compareAndSet(false, true)) {            instanceInfo.setIsDirty();  // for initial register            //提交一个任务，延时执行，注意第一个参数是this，因此延时结束时，InstanceInfoReplicator的run方法会被执行            Future next = scheduler.schedule(this, initialDelayMs, TimeUnit.SECONDS);            //这个任务的Feature对象放在成员变量scheduledPeriodicRef中            scheduledPeriodicRef.set(next);    }}

public void run() {        try {            //更新信息，用于稍后的上报            discoveryClient.refreshInstanceInfo();
            Long dirtyTimestamp = instanceInfo.isDirtyWithTime();            if (dirtyTimestamp != null) {                //上报                discoveryClient.register();                instanceInfo.unsetIsDirty(dirtyTimestamp);            }        } catch (Throwable t) {            logger.warn("There was a problem with the instance info replicator", t);        } finally {            //每次执行完毕都会创建一个延时执行的任务，就这样实现了周期性执行的逻辑            Future next = scheduler.schedule(this, replicationIntervalSeconds, TimeUnit.SECONDS);            //每次创建的周期性任务，都要放入scheduledPeriodicRef，            //如果外部调用了onDemandUpdate，就能通过onDemandUpdate取得当前要执行的任务            scheduledPeriodicRef.set(next);        }    }

public boolean onDemandUpdate() {        //没有达到频率限制才会执行        if (rateLimiter.acquire(burstSize, allowedRatePerMinute)) {            //提交一个任务            scheduler.submit(new Runnable() {                @Override                public void run() {                    logger.debug("Executing on-demand update of local InstanceInfo");                    //取出之前已经提交的任务                    Future latestPeriodic = scheduledPeriodicRef.get();                    //如果此任务未完成，就立即取消                    if (latestPeriodic != null && !latestPeriodic.isDone()) {                        logger.debug("Canceling the latest scheduled update, it will be rescheduled at the end of on demand update");                        latestPeriodic.cancel(false);                    }          //通过调用run方法，令任务在延时后执行，相当于周期性任务中的一次                    InstanceInfoReplicator.this.run();                }            });            return true;        } else {            //如果超过了设置的频率限制，本次onDemandUpdate方法就提交任务了            logger.warn("Ignoring onDemand update due to rate limiter");            return false;        }    }