Moves ConsulCatalogWatch to TaskScheduler rather than @Scheduled
fixes gh-146
This commit is contained in:
@@ -170,6 +170,17 @@ public String serviceUrl() {
|
||||
}
|
||||
----
|
||||
|
||||
=== Consul Catalog Watch
|
||||
|
||||
The Consul Catalog Watch takes advantage of the ability of consul to https://www.consul.io/docs/agent/watches.html#services[watch services]. The Catalog Watch makes a blocking Consul HTTP API call to determine if any services have changed. If there is new service data a Heartbeat Event is published.
|
||||
|
||||
To change the frequency of when the Config Watch is called change `spring.cloud.consul.config.discovery.catalog-services-watch-delay`. The default value is 1000, which is in milliseconds. The delay is the amount of time after the end of the previous invocation and the start of the next.
|
||||
|
||||
To disable the Catalog Watch set `spring.cloud.consul.discovery.catalog-services-watch.enabled=false`.
|
||||
|
||||
|
||||
The watch uses a Spring `TaskScheduler` to schedule the call to consul. By default it is a `ThreadPoolTaskScheduler` with a `poolSize` of 1. To change the `TaskScheduler`, create a bean of type `TaskScheduler` named with the `ConsulDiscoveryClientConfiguration.CATALOG_WATCH_TASK_SCHEDULER_NAME` constant.
|
||||
|
||||
[[spring-cloud-consul-config]]
|
||||
== Distributed Configuration with Consul
|
||||
|
||||
|
||||
@@ -19,33 +19,51 @@ package org.springframework.cloud.consul.discovery;
|
||||
import java.math.BigInteger;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ScheduledFuture;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
|
||||
import org.springframework.cloud.client.discovery.event.HeartbeatEvent;
|
||||
import org.springframework.context.ApplicationEventPublisher;
|
||||
import org.springframework.context.ApplicationEventPublisherAware;
|
||||
import org.springframework.scheduling.annotation.Scheduled;
|
||||
|
||||
import com.ecwid.consul.v1.ConsulClient;
|
||||
import com.ecwid.consul.v1.QueryParams;
|
||||
import com.ecwid.consul.v1.Response;
|
||||
|
||||
import io.micrometer.core.annotation.Timed;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
|
||||
import org.springframework.cloud.client.discovery.event.HeartbeatEvent;
|
||||
import org.springframework.context.ApplicationEventPublisher;
|
||||
import org.springframework.context.ApplicationEventPublisherAware;
|
||||
import org.springframework.context.SmartLifecycle;
|
||||
import org.springframework.scheduling.TaskScheduler;
|
||||
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
|
||||
|
||||
/**
|
||||
* @author Spencer Gibb
|
||||
*/
|
||||
@Slf4j
|
||||
public class ConsulCatalogWatch implements ApplicationEventPublisherAware {
|
||||
public class ConsulCatalogWatch implements ApplicationEventPublisherAware, SmartLifecycle {
|
||||
|
||||
private final ConsulDiscoveryProperties properties;
|
||||
private final ConsulClient consul;
|
||||
private final TaskScheduler taskScheduler;
|
||||
private final AtomicReference<BigInteger> catalogServicesIndex = new AtomicReference<>();
|
||||
private final AtomicBoolean running = new AtomicBoolean(false);
|
||||
private ApplicationEventPublisher publisher;
|
||||
private ScheduledFuture<?> watchFuture;
|
||||
|
||||
public ConsulCatalogWatch(ConsulDiscoveryProperties properties, ConsulClient consul) {
|
||||
this(properties, consul, getTaskScheduler());
|
||||
}
|
||||
|
||||
private static ThreadPoolTaskScheduler getTaskScheduler() {
|
||||
ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
|
||||
taskScheduler.initialize();
|
||||
return taskScheduler;
|
||||
}
|
||||
|
||||
public ConsulCatalogWatch(ConsulDiscoveryProperties properties, ConsulClient consul, TaskScheduler taskScheduler) {
|
||||
this.properties = properties;
|
||||
this.consul = consul;
|
||||
this.taskScheduler = taskScheduler;
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -53,7 +71,43 @@ public class ConsulCatalogWatch implements ApplicationEventPublisherAware {
|
||||
this.publisher = publisher;
|
||||
}
|
||||
|
||||
@Scheduled(fixedDelayString = "${spring.cloud.consul.discovery.catalogServicesWatchDelay:30000}")
|
||||
@Override
|
||||
public boolean isAutoStartup() {
|
||||
return true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void stop(Runnable callback) {
|
||||
this.stop();
|
||||
callback.run();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void start() {
|
||||
if (this.running.compareAndSet(false, true)) {
|
||||
this.watchFuture = this.taskScheduler.scheduleWithFixedDelay(this::catalogServicesWatch,
|
||||
this.properties.getCatalogServicesWatchDelay());
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void stop() {
|
||||
if (this.running.compareAndSet(true, false) && this.watchFuture != null) {
|
||||
this.watchFuture.cancel(true);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isRunning() {
|
||||
return false;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int getPhase() {
|
||||
return 0;
|
||||
}
|
||||
|
||||
@Timed(value ="consul.watch-catalog-services")
|
||||
public void catalogServicesWatch() {
|
||||
try {
|
||||
long index = -1;
|
||||
|
||||
@@ -16,11 +16,13 @@
|
||||
|
||||
package org.springframework.cloud.consul.discovery;
|
||||
|
||||
import com.ecwid.consul.v1.ConsulClient;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.beans.factory.annotation.Qualifier;
|
||||
import org.springframework.boot.autoconfigure.AutoConfigureBefore;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||
import org.springframework.boot.autoconfigure.web.ServerProperties;
|
||||
import org.springframework.boot.context.properties.EnableConfigurationProperties;
|
||||
import org.springframework.cloud.client.CommonsClientAutoConfiguration;
|
||||
import org.springframework.cloud.client.discovery.simple.SimpleDiscoveryClientAutoConfiguration;
|
||||
@@ -28,8 +30,8 @@ import org.springframework.cloud.commons.util.InetUtils;
|
||||
import org.springframework.cloud.consul.ConditionalOnConsulEnabled;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
|
||||
import com.ecwid.consul.v1.ConsulClient;
|
||||
import org.springframework.scheduling.TaskScheduler;
|
||||
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
|
||||
|
||||
/**
|
||||
* @author Spencer Gibb
|
||||
@@ -42,6 +44,8 @@ import com.ecwid.consul.v1.ConsulClient;
|
||||
CommonsClientAutoConfiguration.class })
|
||||
public class ConsulDiscoveryClientConfiguration {
|
||||
|
||||
public static final String CATALOG_WATCH_TASK_SCHEDULER_NAME = "catalogWatchTaskScheduler";
|
||||
|
||||
@Autowired
|
||||
private ConsulClient consulClient;
|
||||
|
||||
@@ -75,7 +79,14 @@ public class ConsulDiscoveryClientConfiguration {
|
||||
@ConditionalOnMissingBean
|
||||
@ConditionalOnProperty(name = "spring.cloud.consul.discovery.catalog-services-watch.enabled", matchIfMissing = true)
|
||||
public ConsulCatalogWatch consulCatalogWatch(
|
||||
ConsulDiscoveryProperties discoveryProperties) {
|
||||
return new ConsulCatalogWatch(discoveryProperties, consulClient);
|
||||
ConsulDiscoveryProperties discoveryProperties,
|
||||
@Qualifier(CATALOG_WATCH_TASK_SCHEDULER_NAME) TaskScheduler taskScheduler) {
|
||||
return new ConsulCatalogWatch(discoveryProperties, 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();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -91,9 +91,11 @@ public class ConsulDiscoveryProperties {
|
||||
|
||||
/** Source of how we will determine the address to use */
|
||||
private boolean preferAgentAddress = false;
|
||||
|
||||
private int catalogServicesWatchDelay = 10;
|
||||
|
||||
/** The delay between calls to watch consul catalog in millis, default is 1000. */
|
||||
private int catalogServicesWatchDelay = 1000;
|
||||
|
||||
/** The number of seconds to block while watching consul catalog, default is 2. */
|
||||
private int catalogServicesWatchTimeout = 2;
|
||||
|
||||
/** Service name */
|
||||
|
||||
@@ -21,6 +21,7 @@ import java.util.Arrays;
|
||||
import org.junit.After;
|
||||
import org.junit.Test;
|
||||
import org.junit.runner.RunWith;
|
||||
|
||||
import org.springframework.boot.autoconfigure.context.PropertyPlaceholderAutoConfiguration;
|
||||
import org.springframework.boot.test.util.TestPropertyValues;
|
||||
import org.springframework.cloud.client.DefaultServiceInstance;
|
||||
@@ -40,8 +41,8 @@ import org.springframework.context.annotation.Configuration;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.mockito.BDDMockito.given;
|
||||
import static org.mockito.Mockito.atLeast;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.times;
|
||||
import static org.mockito.Mockito.verify;
|
||||
|
||||
/**
|
||||
@@ -73,7 +74,7 @@ public class DiscoveryClientConfigServiceAutoConfigurationTests {
|
||||
.getBeanNamesForType(ConsulConfigServerAutoConfiguration.class).length);
|
||||
ConsulDiscoveryClient client = this.context.getParent().getBean(
|
||||
ConsulDiscoveryClient.class);
|
||||
verify(client, times(2)).getInstances("configserver");
|
||||
verify(client, atLeast(2)).getInstances("configserver");
|
||||
ConfigClientProperties locator = this.context
|
||||
.getBean(ConfigClientProperties.class);
|
||||
assertEquals("http://foo:7001/", locator.getUri()[0]);
|
||||
|
||||
Reference in New Issue
Block a user