聊聊spring cloud consul的TtlScheduler

序

本文主要研究一下spring cloud consul的TtlScheduler

TtlScheduler

spring-cloud-consul-discovery-2.1.2.RELEASE-sources.jar!/org/springframework/cloud/consul/discovery/TtlScheduler.java

public class TtlScheduler {

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

    private final Map<String, ScheduledFuture> serviceHeartbeats = new ConcurrentHashMap<>();

    private final TaskScheduler scheduler = new ConcurrentTaskScheduler(
            Executors.newSingleThreadScheduledExecutor());

    private HeartbeatProperties configuration;

    private ConsulClient client;

    public TtlScheduler(HeartbeatProperties configuration, ConsulClient client) {
        this.configuration = configuration;
        this.client = client;
    }

    @Deprecated
    public void add(final NewService service) {
        add(service.getId());
    }

    /**
     * Add a service to the checks loop.
     * @param instanceId instance id
     */
    public void add(String instanceId) {
        ScheduledFuture task = this.scheduler.scheduleAtFixedRate(
                new ConsulHeartbeatTask(instanceId), this.configuration
                        .computeHearbeatInterval().toStandardDuration().getMillis());
        ScheduledFuture previousTask = this.serviceHeartbeats.put(instanceId, task);
        if (previousTask != null) {
            previousTask.cancel(true);
        }
    }

    public void remove(String instanceId) {
        ScheduledFuture task = this.serviceHeartbeats.get(instanceId);
        if (task != null) {
            task.cancel(true);
        }
        this.serviceHeartbeats.remove(instanceId);
    }

    private class ConsulHeartbeatTask implements Runnable {

        private String checkId;

        ConsulHeartbeatTask(String serviceId) {
            this.checkId = serviceId;
            if (!this.checkId.startsWith("service:")) {
                this.checkId = "service:" + this.checkId;
            }
        }

        @Override
        public void run() {
            TtlScheduler.this.client.agentCheckPass(this.checkId);
            if (log.isDebugEnabled()) {
                log.debug("Sending consul heartbeat for: " + this.checkId);
            }
        }

    }

}

• TtlScheduler的构造器接收HeartbeatProperties、ConsulClient；add方法会往scheduler注册一个定时执行的ConsulHeartbeatTask，执行间隔由HeartbeatProperties的ttlValue及intervalRatio计算而来，同时如果previousTask不为null则对其进行cancel；remove方法则cancel调用ScheduledFuture，然后将其从serviceHeartbeats中移除；ConsulHeartbeatTask的run方法则是执行client.agentCheckPass

HeartbeatProperties

spring-cloud-consul-discovery-2.1.2.RELEASE-sources.jar!/org/springframework/cloud/consul/discovery/HeartbeatProperties.java

@ConfigurationProperties(prefix = "spring.cloud.consul.discovery.heartbeat")
@Validated
public class HeartbeatProperties {

    private static final Log log = org.apache.commons.logging.LogFactory
            .getLog(HeartbeatProperties.class);

    // TODO: change enabled to default to true when I stop seeing messages like
    // [WARN] agent: Check 'service:testConsulApp:xtest:8080' missed TTL, is now critical
    boolean enabled = false;

    @Min(1)
    private int ttlValue = 30;

    @NotNull
    private String ttlUnit = "s";

    @DecimalMin("0.1")
    @DecimalMax("0.9")
    private double intervalRatio = 2.0 / 3.0;

    // TODO: did heartbeatInterval need to be a field?

    protected Period computeHearbeatInterval() {
        // heartbeat rate at ratio * ttl, but no later than ttl -1s and, (under lesser
        // priority), no sooner than 1s from now
        double interval = this.ttlValue * this.intervalRatio;
        double max = Math.max(interval, 1);
        int ttlMinus1 = this.ttlValue - 1;
        double min = Math.min(ttlMinus1, max);
        Period heartbeatInterval = new Period(Math.round(1000 * min));
        log.debug("Computed heartbeatInterval: " + heartbeatInterval);
        return heartbeatInterval;
    }

    public String getTtl() {
        return this.ttlValue + this.ttlUnit;
    }

    public boolean isEnabled() {
        return this.enabled;
    }

    public void setEnabled(boolean enabled) {
        this.enabled = enabled;
    }

    public @Min(1) int getTtlValue() {
        return this.ttlValue;
    }

    public void setTtlValue(@Min(1) int ttlValue) {
        this.ttlValue = ttlValue;
    }

    public @NotNull String getTtlUnit() {
        return this.ttlUnit;
    }

    public void setTtlUnit(@NotNull String ttlUnit) {
        this.ttlUnit = ttlUnit;
    }

    public @DecimalMin("0.1") @DecimalMax("0.9") double getIntervalRatio() {
        return this.intervalRatio;
    }

    public void setIntervalRatio(
            @DecimalMin("0.1") @DecimalMax("0.9") double intervalRatio) {
        this.intervalRatio = intervalRatio;
    }

    @Override
    public String toString() {
        return new ToStringCreator(this).append("enabled", this.enabled)
                .append("ttlValue", this.ttlValue).append("ttlUnit", this.ttlUnit)
                .append("intervalRatio", this.intervalRatio).toString();
    }

}

• HeartbeatProperties有ttlValue、intervalRatio等属性，其computeHearbeatInterval方法会取ttlMinus1及Math.max(interval, 1)的最小值作为heartbeatInterval

ConsulDiscoveryClientConfiguration

spring-cloud-consul-discovery-2.1.2.RELEASE-sources.jar!/org/springframework/cloud/consul/discovery/ConsulDiscoveryClientConfiguration.java

@Configuration
@ConditionalOnConsulEnabled
@ConditionalOnProperty(value = "spring.cloud.consul.discovery.enabled", matchIfMissing = true)
@ConditionalOnDiscoveryEnabled
@EnableConfigurationProperties
@AutoConfigureBefore({ SimpleDiscoveryClientAutoConfiguration.class,
        CommonsClientAutoConfiguration.class })
public class ConsulDiscoveryClientConfiguration {

    /**
     * Name of the catalog watch task scheduler bean.
     */
    public static final String CATALOG_WATCH_TASK_SCHEDULER_NAME = "catalogWatchTaskScheduler";

    @Autowired
    private ConsulClient consulClient;

    @Bean
    @ConditionalOnMissingBean
    @ConditionalOnProperty("spring.cloud.consul.discovery.heartbeat.enabled")
    // TODO: move to service-registry for Edgware
    public TtlScheduler ttlScheduler(HeartbeatProperties heartbeatProperties) {
        return new TtlScheduler(heartbeatProperties, this.consulClient);
    }

    @Bean
    @ConditionalOnMissingBean
    // TODO: move to service-registry for Edgware
    public HeartbeatProperties heartbeatProperties() {
        return new HeartbeatProperties();
    }

    @Bean
    @ConditionalOnMissingBean
    // TODO: Split appropriate values to service-registry for Edgware
    public ConsulDiscoveryProperties consulDiscoveryProperties(InetUtils inetUtils) {
        return new ConsulDiscoveryProperties(inetUtils);
    }

    @Bean
    @ConditionalOnMissingBean
    public ConsulDiscoveryClient consulDiscoveryClient(
            ConsulDiscoveryProperties discoveryProperties) {
        return new ConsulDiscoveryClient(this.consulClient, discoveryProperties);
    }

    @Bean
    @ConditionalOnMissingBean
    @ConditionalOnProperty(name = "spring.cloud.consul.discovery.catalog-services-watch.enabled", matchIfMissing = true)
    public ConsulCatalogWatch consulCatalogWatch(
            ConsulDiscoveryProperties discoveryProperties,
            @Qualifier(CATALOG_WATCH_TASK_SCHEDULER_NAME) TaskScheduler taskScheduler) {
        return new ConsulCatalogWatch(discoveryProperties, this.consulClient,
                taskScheduler);
    }

    @Bean(name = CATALOG_WATCH_TASK_SCHEDULER_NAME)
    @ConditionalOnProperty(name = "spring.cloud.consul.discovery.catalog-services-watch.enabled", matchIfMissing = true)
    public TaskScheduler catalogWatchTaskScheduler() {
        return new ThreadPoolTaskScheduler();
    }
}

• ConsulDiscoveryClientConfiguration在@ConditionalOnMissingBean以及@ConditionalOnProperty("spring.cloud.consul.discovery.heartbeat.enabled")的条件下会自动注册ttlScheduler

小结

TtlScheduler的构造器接收HeartbeatProperties、ConsulClient；add方法会往scheduler注册一个定时执行的ConsulHeartbeatTask，执行间隔由HeartbeatProperties的ttlValue及intervalRatio计算而来，同时如果previousTask不为null则对其进行cancel；remove方法则cancel调用ScheduledFuture，然后将其从serviceHeartbeats中移除；ConsulHeartbeatTask的run方法则是执行client.agentCheckPass

doc

• TtlScheduler