聊聊spring cloud eureka的instanceEnabledOnit屬性

時間 2019-12-04

標籤聊聊 spring cloud eureka instanceenabledonit 屬性欄目 Spring 简体版

原文原文鏈接

序

本文主要研究下spring cloud eureka的instanceEnabledOnit屬性java

EurekaInstanceConfigBean

spring-cloud-netflix-eureka-client-2.0.0.RC1-sources.jar!/org/springframework/cloud/netflix/eureka/EurekaInstanceConfigBean.javaspring

@ConfigurationProperties("eureka.instance")
public class EurekaInstanceConfigBean implements CloudEurekaInstanceConfig, EnvironmentAware {
    //......
    /**
     * Indicates whether the instance should be enabled for taking traffic as soon as it
     * is registered with eureka. Sometimes the application might need to do some
     * pre-processing before it is ready to take traffic.
     */
    private boolean instanceEnabledOnit;

    //......
}

這個屬性用來決定應用服務是否一註冊上就能夠開始接收請求

配置定義以下
spring-cloud-netflix-eureka-client-2.0.0.RC1.jar!/META-INF/spring-configuration-metadata.jsonjson

{
      "sourceType": "org.springframework.cloud.netflix.eureka.EurekaInstanceConfigBean",
      "defaultValue": false,
      "name": "eureka.instance.instance-enabled-onit",
      "description": "Indicates whether the instance should be enabled for taking traffic as soon as it\n is registered with eureka. Sometimes the application might need to do some\n pre-processing before it is ready to take traffic.",
      "type": "java.lang.Boolean"
    }

PropertiesInstanceConfig

eureka-client-1.8.8-sources.jar!/com/netflix/appinfo/PropertiesInstanceConfig.javaapp

static final String TRAFFIC_ENABLED_ON_INIT_KEY = "traffic.enabled";

    @Override
    public boolean isInstanceEnabledOnit() {
        return configInstance.getBooleanProperty(namespace + TRAFFIC_ENABLED_ON_INIT_KEY,
                super.isInstanceEnabledOnit()).get();
    }

這裏走的是super.isInstanceEnabledOnit()

變動爲STARTING

InstanceInfoFactory

spring-cloud-netflix-eureka-client-2.0.0.RC1-sources.jar!/org/springframework/cloud/netflix/eureka/InstanceInfoFactory.javaide

public class InstanceInfoFactory {

    private static final Log log = LogFactory.getLog(InstanceInfoFactory.class);

    public InstanceInfo create(EurekaInstanceConfig config) {
        LeaseInfo.Builder leaseInfoBuilder = LeaseInfo.Builder.newBuilder()
                .setRenewalIntervalInSecs(config.getLeaseRenewalIntervalInSeconds())
                .setDurationInSecs(config.getLeaseExpirationDurationInSeconds());

        // Builder the instance information to be registered with eureka
        // server
        InstanceInfo.Builder builder = InstanceInfo.Builder.newBuilder();

        String namespace = config.getNamespace();
        if (!namespace.endsWith(".")) {
            namespace = namespace + ".";
        }
        builder.setNamespace(namespace).setAppName(config.getAppname())
                .setInstanceId(config.getInstanceId())
                .setAppGroupName(config.getAppGroupName())
                .setDataCenterInfo(config.getDataCenterInfo())
                .setIPAddr(config.getIpAddress()).setHostName(config.getHostName(false))
                .setPort(config.getNonSecurePort())
                .enablePort(InstanceInfo.PortType.UNSECURE,
                        config.isNonSecurePortEnabled())
                .setSecurePort(config.getSecurePort())
                .enablePort(InstanceInfo.PortType.SECURE, config.getSecurePortEnabled())
                .setVIPAddress(config.getVirtualHostName())
                .setSecureVIPAddress(config.getSecureVirtualHostName())
                .setHomePageUrl(config.getHomePageUrlPath(), config.getHomePageUrl())
                .setStatusPageUrl(config.getStatusPageUrlPath(),
                        config.getStatusPageUrl())
                .setHealthCheckUrls(config.getHealthCheckUrlPath(),
                        config.getHealthCheckUrl(), config.getSecureHealthCheckUrl())
                .setASGName(config.getASGName());

        // Start off with the STARTING state to avoid traffic
        if (!config.isInstanceEnabledOnit()) {
            InstanceInfo.InstanceStatus initialStatus = InstanceInfo.InstanceStatus.STARTING;
            if (log.isInfoEnabled()) {
                log.info("Setting initial instance status as: " + initialStatus);
            }
            builder.setStatus(initialStatus);
        }
        else {
            if (log.isInfoEnabled()) {
                log.info("Setting initial instance status as: "
                        + InstanceInfo.InstanceStatus.UP
                        + ". This may be too early for the instance to advertise itself as available. "
                        + "You would instead want to control this via a healthcheck handler.");
            }
        }

        // Add any user-specific metadata information
        for (Map.Entry<String, String> mapEntry : config.getMetadataMap().entrySet()) {
            String key = mapEntry.getKey();
            String value = mapEntry.getValue();
            // only add the metadata if the value is present
            if (value != null && !value.isEmpty()) {
                builder.add(key, value);
            }
        }

        InstanceInfo instanceInfo = builder.build();
        instanceInfo.setLeaseInfo(leaseInfoBuilder.build());
        return instanceInfo;
    }
}

能夠看到這裏判斷isInstanceEnabledOnit，若是不是，則InstanceStatus初始狀態爲InstanceInfo.InstanceStatus.STARTING；而InstanceInfo建立的時候，默認初始化status爲InstanceStatus.UP，因此若是爲true，這裏就log一下

EurekaRegistration

spring-cloud-netflix-eureka-client-2.0.0.RC1-sources.jar!/org/springframework/cloud/netflix/eureka/serviceregistry/EurekaRegistration.javaui

public EurekaRegistration build() {
            Assert.notNull(instanceConfig, "instanceConfig may not be null");

            if (this.applicationInfoManager == null) {
                InstanceInfo instanceInfo = new InstanceInfoFactory().create(this.instanceConfig);
                this.applicationInfoManager = new ApplicationInfoManager(this.instanceConfig, instanceInfo);
            }
            if (this.eurekaClient == null) {
                Assert.notNull(this.clientConfig, "if eurekaClient is null, EurekaClientConfig may not be null");
                Assert.notNull(this.publisher, "if eurekaClient is null, ApplicationEventPublisher may not be null");

                this.eurekaClient = new CloudEurekaClient(this.applicationInfoManager, this.clientConfig, this.publisher);
            }
            return new EurekaRegistration(instanceConfig, eurekaClient, applicationInfoManager, healthCheckHandler);
        }

EurekaRegistration建立的時候，使用了new InstanceInfoFactory().create(this.instanceConfig)來建立InstanceInfo

DiscoveryClient.initScheduledTasks

eureka-client-1.8.8-sources.jar!/com/netflix/discovery/DiscoveryClient.javathis

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");
        }
    }

該client構造器初始化initScheduledTasks，這裏頭調用了instanceInfoReplicator.start(clientConfig.getInitialInstanceInfoReplicationIntervalSeconds())，默認是延遲40秒執行

InstanceInfoReplicator

eureka-client-1.8.8-sources.jar!/com/netflix/discovery/InstanceInfoReplicator.javaspa

class InstanceInfoReplicator implements Runnable {
    
    public void start(int initialDelayMs) {
        if (started.compareAndSet(false, true)) {
            instanceInfo.setIsDirty();  // for initial register
            Future next = scheduler.schedule(this, initialDelayMs, TimeUnit.SECONDS);
            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.set(next);
        }
    }

    public boolean onDemandUpdate() {
        if (rateLimiter.acquire(burstSize, allowedRatePerMinute)) {
            if (!scheduler.isShutdown()) {
                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);
                        }
    
                        InstanceInfoReplicator.this.run();
                    }
                });
                return true;
            } else {
                logger.warn("Ignoring onDemand update due to stopped scheduler");
                return false;
            }
        } else {
            logger.warn("Ignoring onDemand update due to rate limiter");
            return false;
        }
    }
}

這裏最開始設定爲dirty，而後第一次執行的時候，就會觸發register，遠程調用註冊。
DiscoveryClient有個statusChangeListener，它調用的是這裏的onDemandUpdate方法，而onDemandUpdate方法在不超過頻率限制的時候，執行的是這個run方法。

變動爲UP

EurekaAutoServiceRegistration

spring-cloud-netflix-eureka-client-2.0.0.RC1-sources.jar!/org/springframework/cloud/netflix/eureka/serviceregistry/EurekaAutoServiceRegistration.java.net

public class EurekaAutoServiceRegistration implements AutoServiceRegistration, SmartLifecycle, Ordered {
    //......
    public void start() {
        // only set the port if the nonSecurePort or securePort is 0 and this.port != 0
        if (this.port.get() != 0) {
            if (this.registration.getNonSecurePort() == 0) {
                this.registration.setNonSecurePort(this.port.get());
            }

            if (this.registration.getSecurePort() == 0 && this.registration.isSecure()) {
                this.registration.setSecurePort(this.port.get());
            }
        }

        // only initialize if nonSecurePort is greater than 0 and it isn't already running
        // because of containerPortInitializer below
        if (!this.running.get() && this.registration.getNonSecurePort() > 0) {

            this.serviceRegistry.register(this.registration);

            this.context.publishEvent(
                    new InstanceRegisteredEvent<>(this, this.registration.getInstanceConfig()));
            this.running.set(true);
        }
    }
}

這裏自動註冊，serviceRegistry.register(this.registration)

public void register(EurekaRegistration reg) {
        maybeInitializeClient(reg);

        if (log.isInfoEnabled()) {
            log.info("Registering application " + reg.getInstanceConfig().getAppname()
                    + " with eureka with status "
                    + reg.getInstanceConfig().getInitialStatus());
        }

        reg.getApplicationInfoManager()
                .setInstanceStatus(reg.getInstanceConfig().getInitialStatus());

        reg.getHealthCheckHandler().ifAvailable(healthCheckHandler ->
                reg.getEurekaClient().registerHealthCheck(healthCheckHandler));
    }

這裏會變動狀態爲reg.getInstanceConfig().getInitialStatus()，該值默認爲UP

spring-cloud-netflix-eureka-client-2.0.0.RC1-sources.jar!/org/springframework/cloud/netflix/eureka/EurekaInstanceConfigBean.javadebug

/**
     * Initial status to register with rmeote Eureka server.
     */
    private InstanceStatus initialStatus = InstanceStatus.UP;

小結

應用啓動時，能夠根據eureka.instance.instance-enabled-onit配置設定(默認爲false)，來配置初始註冊到eureka server的時候，其status是UP，仍是STARTING。默認初始化的時候是STARTING，以後自動註冊的時候，變動爲UP，表示能夠開始接收請求。

DiscoveryClient初始化了一堆定時任務，其中包括了InstanceInfoReplicator的run方法，它默認是延時40秒執行，它在start的時候先設置爲instanceInfo.setIsDirty()，以後第一次執行就能夠觸發discoveryClient.register()。
而EurekaAutoServiceRegistration在啓動的時候，會調用serviceRegistry.register(this.registration)，以後變動狀態，發佈StatusChangeEvent，DiscoveryClient有個statusChangeListener，它調用的是InstanceInfoReplicator的onDemandUpdate方法，而onDemandUpdate方法在不超過頻率限制的時候，執行的是InstanceInfoReplicator的run方法。

能夠發現無論是InstanceInfoReplicator的延時任務，仍是urekaAutoServiceRegistration變動StatusChangeEvent，最後都是由InstanceInfoReplicator的run方法去觸發discoveryClient.register()，去遠程eureka server註冊。

相關標籤/搜索

每日一句

每一个你不满意的现在，都有一个你没有努力的曾经。