This is an automated email from the ASF dual-hosted git repository.
iluo pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-dubbo.git
The following commit(s) were added to refs/heads/master by this push:
new ff50a29 consul enhancement (#3963)
ff50a29 is described below
commit ff50a292a04eb174464600012257db541730ef2a
Author: myPrecious <[email protected]>
AuthorDate: Sun May 5 14:49:07 2019 +0800
consul enhancement (#3963)
* revert to use http ttl check for consul
* clean code
---
.../dubbo/registry/consul/ConsulRegistry.java | 41 ++++++++++++++++------
1 file changed, 30 insertions(+), 11 deletions(-)
diff --git
a/dubbo-registry/dubbo-registry-consul/src/main/java/org/apache/dubbo/registry/consul/ConsulRegistry.java
b/dubbo-registry/dubbo-registry-consul/src/main/java/org/apache/dubbo/registry/consul/ConsulRegistry.java
index 888faa4..7bcd6b0 100644
---
a/dubbo-registry/dubbo-registry-consul/src/main/java/org/apache/dubbo/registry/consul/ConsulRegistry.java
+++
b/dubbo-registry/dubbo-registry-consul/src/main/java/org/apache/dubbo/registry/consul/ConsulRegistry.java
@@ -40,9 +40,12 @@ import java.util.List;
import java.util.ArrayList;
import java.util.Map;
import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.Executors;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
import java.util.stream.Collectors;
import static java.util.concurrent.Executors.newCachedThreadPool;
@@ -57,31 +60,34 @@ public class ConsulRegistry extends FailbackRegistry {
private static final String SERVICE_TAG = "dubbo";
private static final String URL_META_KEY = "url";
private static final String WATCH_TIMEOUT = "consul-watch-timeout";
- private static final String CHECK_INTERVAL = "consul-check-interval";
- private static final String CHECK_TIMEOUT = "consul-check-timeout";
+ private static final String CHECK_PASS_INTERVAL =
"consul-check-pass-interval";
private static final String DEREGISTER_AFTER =
"consul-deregister-critical-service-after";
private static final int DEFAULT_PORT = 8500;
// default watch timeout in millisecond
private static final int DEFAULT_WATCH_TIMEOUT = 60 * 1000;
- // default tcp check interval
- private static final String DEFAULT_CHECK_INTERVAL = "10s";
- // default tcp check timeout
- private static final String DEFAULT_CHECK_TIMEOUT = "1s";
+ // default time-to-live in millisecond
+ private static final long DEFAULT_CHECK_PASS_INTERVAL = 16000L;
// default deregister critical server after
private static final String DEFAULT_DEREGISTER_TIME = "20s";
private ConsulClient client;
-
+ private long checkPassInterval;
private ExecutorService notifierExecutor = newCachedThreadPool(
new NamedThreadFactory("dubbo-consul-notifier", true));
private ConcurrentMap<URL, ConsulNotifier> notifiers = new
ConcurrentHashMap<>();
+ private ScheduledExecutorService ttlConsulCheckExecutor;
+
public ConsulRegistry(URL url) {
super(url);
String host = url.getHost();
int port = url.getPort() != 0 ? url.getPort() : DEFAULT_PORT;
client = new ConsulClient(host, port);
+ checkPassInterval = url.getParameter(CHECK_PASS_INTERVAL,
DEFAULT_CHECK_PASS_INTERVAL);
+ ttlConsulCheckExecutor = Executors.newSingleThreadScheduledExecutor();
+ ttlConsulCheckExecutor.scheduleAtFixedRate(this::checkPass,
checkPassInterval / 8,
+ checkPassInterval / 8, TimeUnit.MILLISECONDS);
}
@Override
@@ -164,7 +170,7 @@ public class ConsulRegistry extends FailbackRegistry {
}
try {
String service = url.getServiceKey();
- Response<List<HealthService>> result =
client.getHealthServices(service,
HealthServicesRequest.newBuilder().setTag(SERVICE_TAG).build());
+ Response<List<HealthService>> result = getHealthServices(service,
-1, buildWatchTimeout(url));
if (result == null || result.getValue() == null ||
result.getValue().isEmpty()) {
return new ArrayList<>();
} else {
@@ -184,6 +190,21 @@ public class ConsulRegistry extends FailbackRegistry {
public void destroy() {
super.destroy();
notifierExecutor.shutdown();
+ ttlConsulCheckExecutor.shutdown();
+ }
+
+ private void checkPass() {
+ for (URL url : getRegistered()) {
+ String checkId = buildId(url);
+ try {
+ client.agentCheckPass("service:" + checkId);
+ if (logger.isDebugEnabled()) {
+ logger.debug("check pass for url: " + url + " with check
id: " + checkId);
+ }
+ } catch (Throwable t) {
+ logger.warn("fail to check pass for url: " + url + ", check id
is: " + checkId);
+ }
+ }
}
private Response<List<HealthService>> getHealthServices(String service,
long index, int watchTimeout) {
@@ -258,9 +279,7 @@ public class ConsulRegistry extends FailbackRegistry {
private NewService.Check buildCheck(URL url) {
NewService.Check check = new NewService.Check();
- check.setTcp(url.getAddress());
- check.setInterval(url.getParameter(CHECK_INTERVAL,
DEFAULT_CHECK_INTERVAL));
- check.setTimeout(url.getParameter(CHECK_TIMEOUT,
DEFAULT_CHECK_TIMEOUT));
+ check.setTtl((checkPassInterval / 1000) + "s");
check.setDeregisterCriticalServiceAfter(url.getParameter(DEREGISTER_AFTER,
DEFAULT_DEREGISTER_TIME));
return check;
}
