You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dubbo.apache.org by al...@apache.org on 2021/11/03 03:13:17 UTC
[dubbo] branch 3.0 updated: [3.0] Add connectivity validation,
router snapshot (#9112)
This is an automated email from the ASF dual-hosted git repository.
albumenj pushed a commit to branch 3.0
in repository https://gitbox.apache.org/repos/asf/dubbo.git
The following commit(s) were added to refs/heads/3.0 by this push:
new efa6900 [3.0] Add connectivity validation, router snapshot (#9112)
efa6900 is described below
commit efa6900bb1b59c70f131af868396a8903b098a18
Author: Albumen Kevin <jh...@gmail.com>
AuthorDate: Wed Nov 3 11:13:04 2021 +0800
[3.0] Add connectivity validation, router snapshot (#9112)
* refactor BitList
* refactor BitList
* remove dependency
* add connectivity directory support
* Fix ut
* Fix ut
* fix npe
* fix npe
* add disable action
* add comment
* add comment
* fix ut
* add router snapshot
* Add comment
* add test
* fix ut
* fix npe
* fix ut
* add ut
* set consumerUrl
* refactor BitList add tailList
* add comment
* support tailList in router chain
* fix default
* fix ut
* fix log
* fix dup
* Fix Directory available check
* change order
* fix ut
* fix logic
* fix concurrent
* fix random problem
* recover common router
* ensure semaphore
* recover meta
* fix invoker
---
dubbo-cluster/pom.xml | 4 -
.../org/apache/dubbo/rpc/cluster/Directory.java | 29 ++
.../java/org/apache/dubbo/rpc/cluster/Router.java | 23 +-
.../org/apache/dubbo/rpc/cluster/RouterChain.java | 143 +++++-
.../rpc/cluster/directory/AbstractDirectory.java | 261 ++++++++++-
.../rpc/cluster/directory/StaticDirectory.java | 29 +-
.../dubbo/rpc/cluster/router/RouterResult.java | 55 +++
.../rpc/cluster/router/RouterSnapshotNode.java | 117 +++++
.../cluster/router/condition/ConditionRouter.java | 20 +-
.../router/condition/config/ListenableRouter.java | 10 +-
.../cluster/router/mesh/route/MeshRuleRouter.java | 12 +-
.../cluster/router/mock/MockInvokersSelector.java | 15 +-
.../rpc/cluster/router/script/ScriptRouter.java | 13 +-
.../cluster/router/state/AbstractStateRouter.java | 14 +-
.../dubbo/rpc/cluster/router/state/BitList.java | 485 +++++++++++++++------
.../rpc/cluster/router/state/StateRouter.java | 25 +-
.../cluster/router/state/StateRouterResult.java | 53 +++
.../cluster/router/tag/TagDynamicStateRouter.java | 12 +-
.../dubbo/rpc/cluster/router/tag/TagRouter.java | 37 +-
.../cluster/router/tag/TagStaticStateRouter.java | 10 +-
.../cluster/support/AbstractClusterInvoker.java | 137 ++++--
.../support/wrapper/MockClusterInvoker.java | 2 +
.../rpc/cluster/directory/StaticDirectoryTest.java | 5 +-
.../loadbalance/ConsistentHashLoadBalanceTest.java | 6 +-
.../router/condition/ConditionRouterTest.java | 44 +-
.../cluster/router/file/FileRouterEngineTest.java | 10 +
.../router/mesh/route/MeshRuleRouterTest.java | 19 +-
.../router/mock/MockInvokersSelectorTest.java | 5 +-
.../cluster/router/script/ScriptRouterTest.java | 8 +-
.../rpc/cluster/router/state/BitListTest.java | 479 +++++++++++++++++++-
.../support/AbstractClusterInvokerTest.java | 11 +-
.../support/ConnectivityValidationTest.java | 292 +++++++++++++
.../support/FailoverClusterInvokerTest.java | 10 +-
.../dubbo/common/constants/CommonConstants.java | 15 +-
.../manager/DefaultExecutorRepository.java | 12 +-
.../threadpool/manager/ExecutorRepository.java | 7 +
dubbo-dependencies-bom/pom.xml | 6 -
dubbo-distribution/dubbo-all/pom.xml | 4 -
.../client/ServiceDiscoveryRegistryDirectory.java | 28 +-
.../registry/integration/DynamicDirectory.java | 25 +-
.../registry/integration/RegistryDirectory.java | 50 +--
41 files changed, 2129 insertions(+), 413 deletions(-)
diff --git a/dubbo-cluster/pom.xml b/dubbo-cluster/pom.xml
index fbb0d17..c20fb74 100644
--- a/dubbo-cluster/pom.xml
+++ b/dubbo-cluster/pom.xml
@@ -32,10 +32,6 @@
</properties>
<dependencies>
<dependency>
- <groupId>org.roaringbitmap</groupId>
- <artifactId>RoaringBitmap</artifactId>
- </dependency>
- <dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-rpc-api</artifactId>
<version>${project.parent.version}</version>
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/Directory.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/Directory.java
index c797218..7ac843b 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/Directory.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/Directory.java
@@ -43,11 +43,16 @@ public interface Directory<T> extends Node {
/**
* list invokers.
+ * filtered by invocation
*
* @return invokers
*/
List<Invoker<T>> list(Invocation invocation) throws RpcException;
+ /**
+ * list invokers
+ * include all invokers from registry
+ */
List<Invoker<T>> getAllInvokers();
URL getConsumerUrl();
@@ -66,6 +71,30 @@ public interface Directory<T> extends Node {
RouterChain<T> getRouterChain();
+ /**
+ * invalidate an invoker, add it into reconnect task, remove from list next time
+ * will be recovered by address refresh notification or reconnect success notification
+ *
+ * @param invoker invoker to invalidate
+ */
+ void addInvalidateInvoker(Invoker<T> invoker);
+
+ /**
+ * disable an invoker, remove from list next time
+ * will be removed when invoker is removed by address refresh notification
+ * using in service offline notification
+ *
+ * @param invoker invoker to invalidate
+ */
+ void addDisabledInvoker(Invoker<T> invoker);
+
+ /**
+ * recover a disabled invoker
+ *
+ * @param invoker invoker to invalidate
+ */
+ void recoverDisabledInvoker(Invoker<T> invoker);
+
default boolean isNotificationReceived() {
return false;
}
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/Router.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/Router.java
index 2877012..23c4fef 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/Router.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/Router.java
@@ -20,6 +20,7 @@ import org.apache.dubbo.common.URL;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.RpcException;
+import org.apache.dubbo.rpc.cluster.router.RouterResult;
import java.util.List;
@@ -28,7 +29,7 @@ import java.util.List;
* <p>
* <a href="http://en.wikipedia.org/wiki/Routing">Routing</a>
*
- * @see org.apache.dubbo.rpc.cluster.Cluster#join(Directory)
+ * @see org.apache.dubbo.rpc.cluster.Cluster#join(Directory, boolean)
* @see org.apache.dubbo.rpc.cluster.Directory#list(Invocation)
*/
public interface Router extends Comparable<Router> {
@@ -51,8 +52,26 @@ public interface Router extends Comparable<Router> {
* @return routed invokers
* @throws RpcException
*/
- <T> List<Invoker<T>> route(List<Invoker<T>> invokers, URL url, Invocation invocation) throws RpcException;
+ @Deprecated
+ default <T> List<Invoker<T>> route(List<Invoker<T>> invokers, URL url, Invocation invocation) throws RpcException {
+ return null;
+ }
+ /**
+ * ** This method can return the state of whether routerChain needed to continue route. **
+ * Filter invokers with current routing rule and only return the invokers that comply with the rule.
+ *
+ * @param invokers invoker list
+ * @param url refer url
+ * @param invocation invocation
+ * @param needToPrintMessage whether to print router state. Such as `use router branch a`.
+ * @return state with route result
+ * @throws RpcException
+ */
+ default <T> RouterResult<Invoker<T>> route(List<Invoker<T>> invokers, URL url, Invocation invocation,
+ boolean needToPrintMessage) throws RpcException {
+ return new RouterResult<>(route(invokers, url, invocation));
+ }
/**
* Notify the router the invoker list. Invoker list may change from time to time. This method gives the router a
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/RouterChain.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/RouterChain.java
index dc09a33..84ec662 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/RouterChain.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/RouterChain.java
@@ -17,17 +17,22 @@
package org.apache.dubbo.rpc.cluster;
import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.Version;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.threadpool.manager.ExecutorRepository;
import org.apache.dubbo.common.utils.CollectionUtils;
+import org.apache.dubbo.common.utils.NetUtils;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
+import org.apache.dubbo.rpc.cluster.router.RouterResult;
+import org.apache.dubbo.rpc.cluster.router.RouterSnapshotNode;
import org.apache.dubbo.rpc.cluster.router.state.AddrCache;
import org.apache.dubbo.rpc.cluster.router.state.BitList;
import org.apache.dubbo.rpc.cluster.router.state.RouterCache;
import org.apache.dubbo.rpc.cluster.router.state.StateRouter;
import org.apache.dubbo.rpc.cluster.router.state.StateRouterFactory;
+import org.apache.dubbo.rpc.cluster.router.state.StateRouterResult;
import java.util.ArrayList;
import java.util.Collections;
@@ -53,7 +58,7 @@ public class RouterChain<T> {
/**
* full list of addresses from registry, classified by method name.
*/
- private volatile List<Invoker<T>> invokers = Collections.emptyList();
+ private volatile BitList<Invoker<T>> invokers = BitList.emptyList();
/**
* containing all routers, reconstruct every time 'route://' urls change.
@@ -163,40 +168,138 @@ public class RouterChain<T> {
* @param invocation
* @return
*/
- public List<Invoker<T>> route(URL url, Invocation invocation) {
+ public List<Invoker<T>> route(URL url, BitList<Invoker<T>> availableInvokers, Invocation invocation) {
AddrCache<T> cache = this.cache.get();
- List<Invoker<T>> finalInvokers = null;
+ BitList<Invoker<T>> resultInvokers = availableInvokers.clone();
+ // 1. route state router
if (cache != null) {
- BitList<Invoker<T>> finalBitListInvokers = new BitList<>(invokers, false);
for (StateRouter stateRouter : stateRouters) {
if (stateRouter.isEnable()) {
RouterCache<T> routerCache = cache.getCache().get(stateRouter.getName());
- finalBitListInvokers = stateRouter.route(finalBitListInvokers, routerCache, url, invocation);
+ StateRouterResult<Invoker<T>> routeResult = stateRouter.route(resultInvokers, routerCache, url, invocation, false);
+ resultInvokers = routeResult.getResult();
+ if (resultInvokers.isEmpty()) {
+ printRouterSnapshot(url, availableInvokers, invocation);
+ return BitList.emptyList();
+ }
+
+ // stop continue routing
+ if (!routeResult.isNeedContinueRoute()) {
+ return routeResult.getResult();
+ }
}
}
- finalInvokers = new ArrayList<>(finalBitListInvokers.size());
+ }
- finalInvokers.addAll(finalBitListInvokers);
+ List<Invoker<T>> commonRouterResult = new ArrayList<>(resultInvokers);
+ // 2. route common router
+ for (Router router : routers) {
+ // Copy resultInvokers to a arrayList. BitList not support
+ RouterResult<Invoker<T>> routeResult = router.route(commonRouterResult, url, invocation, false);
+ commonRouterResult = routeResult.getResult();
+ if (CollectionUtils.isEmpty(commonRouterResult)) {
+ printRouterSnapshot(url, availableInvokers, invocation);
+ return BitList.emptyList();
+ }
+
+ // stop continue routing
+ if (!routeResult.isNeedContinueRoute()) {
+ return commonRouterResult;
+ }
}
+ return commonRouterResult;
+ }
+
+ /**
+ * store each router's input and output, log out if empty
+ */
+ private void printRouterSnapshot(URL url, BitList<Invoker<T>> availableInvokers, Invocation invocation) {
+ logRouterSnapshot(url, invocation, buildRouterSnapshot(url, availableInvokers, invocation));
+ }
+
+ /**
+ * Build each router's result
+ */
+ private RouterSnapshotNode<T> buildRouterSnapshot(URL url, BitList<Invoker<T>> availableInvokers, Invocation invocation) {
+ AddrCache<T> cache = this.cache.get();
+ BitList<Invoker<T>> resultInvokers = availableInvokers.clone();
+ RouterSnapshotNode<T> snapshotNode = new RouterSnapshotNode<T>("Parent", resultInvokers.size());
+ snapshotNode.setOutputInvokers(resultInvokers.clone());
+
+ // 1. route state router
+ if (cache != null) {
+ for (StateRouter stateRouter : stateRouters) {
+ if (stateRouter.isEnable()) {
+ BitList<Invoker<T>> inputInvokers = resultInvokers.clone();
+
+ RouterSnapshotNode<T> currentNode = new RouterSnapshotNode<T>(stateRouter.getName(), inputInvokers.size());
+ snapshotNode.appendNode(currentNode);
- if (finalInvokers == null) {
- finalInvokers = new ArrayList<>(invokers);
+ RouterCache<T> routerCache = cache.getCache().get(stateRouter.getName());
+ StateRouterResult<Invoker<T>> routeResult = stateRouter.route(inputInvokers, routerCache, url, invocation, true);
+ resultInvokers = routeResult.getResult();
+ String routerMessage = routeResult.getMessage();
+
+ currentNode.setOutputInvokers(resultInvokers);
+ currentNode.setRouterMessage(routerMessage);
+
+ // result is empty, log out
+ if (resultInvokers.isEmpty()) {
+ return snapshotNode;
+ }
+
+ if (!routeResult.isNeedContinueRoute()) {
+ return snapshotNode;
+ }
+ }
+ }
}
+ List<Invoker<T>> commonRouterResult = resultInvokers;
+ // 2. route common router
for (Router router : routers) {
- finalInvokers = router.route(finalInvokers, url, invocation);
+ // Copy resultInvokers to a arrayList. BitList not support
+ List<Invoker<T>> inputInvokers = new ArrayList<>(commonRouterResult);
+
+ RouterSnapshotNode<T> currentNode = new RouterSnapshotNode<T>(router.getClass().getSimpleName(), inputInvokers.size());
+ snapshotNode.appendNode(currentNode);
+
+ RouterResult<Invoker<T>> routeStateResult = router.route(inputInvokers, url, invocation, true);
+ List<Invoker<T>> routeResult = routeStateResult.getResult();
+ String routerMessage = routeStateResult.getMessage();
+
+ currentNode.setOutputInvokers(routeResult);
+ currentNode.setRouterMessage(routerMessage);
+
+ // result is empty, log out
+ if (CollectionUtils.isEmpty(routeResult)) {
+ return snapshotNode;
+ } else {
+ commonRouterResult = routeResult;
+ }
+
+ if (!routeStateResult.isNeedContinueRoute()) {
+ return snapshotNode;
+ }
}
- return finalInvokers;
+ return snapshotNode;
+ }
+
+ private void logRouterSnapshot(URL url, Invocation invocation, RouterSnapshotNode<T> snapshotNode) {
+ logger.warn("No provider available after route for the service " + url.getServiceKey()
+ + " from registry " + url.getAddress()
+ + " on the consumer " + NetUtils.getLocalHost()
+ + " using the dubbo version " + Version.getVersion() + ". Router snapshot is below: \n" + snapshotNode.toString());
}
/**
* Notify router chain of the initial addresses from registry at the first time.
* Notify whenever addresses in registry change.
*/
- public void setInvokers(List<Invoker<T>> invokers) {
- this.invokers = (invokers == null ? Collections.emptyList() : invokers);
+ public void setInvokers(BitList<Invoker<T>> invokers) {
+ this.invokers = (invokers == null ? BitList.emptyList() : invokers);
stateRouters.forEach(router -> router.notify(this.invokers));
routers.forEach(router -> router.notify(this.invokers));
loop(true);
@@ -204,6 +307,7 @@ public class RouterChain<T> {
/**
* Build the asynchronous address cache for stateRouter.
+ *
* @param notify Whether the addresses in registry have changed.
*/
private void buildCache(boolean notify) {
@@ -232,10 +336,11 @@ public class RouterChain<T> {
/**
* Cache the address list for each StateRouter.
- * @param router router
- * @param origin The original address cache
+ *
+ * @param router router
+ * @param origin The original address cache
* @param invokers The full address list
- * @param notify Whether the addresses in registry has changed.
+ * @param notify Whether the addresses in registry has changed.
* @return
*/
private RouterCache poolRouter(StateRouter router, AddrCache<T> origin, List<Invoker<T>> invokers, boolean notify) {
@@ -263,7 +368,7 @@ public class RouterChain<T> {
* @param notify Whether the addresses in registry has changed.
*/
public void loop(boolean notify) {
- if (firstBuildCache.compareAndSet(true,false)) {
+ if (firstBuildCache.compareAndSet(true, false)) {
buildCache(notify);
}
@@ -278,7 +383,7 @@ public class RouterChain<T> {
}
}
} catch (RejectedExecutionException e) {
- if (loopPool.isShutdown()){
+ if (loopPool.isShutdown()) {
logger.warn("loopPool executor service is shutdown, ignoring notify loop");
return;
}
@@ -304,7 +409,7 @@ public class RouterChain<T> {
}
public void destroy() {
- invokers = Collections.emptyList();
+ invokers = BitList.emptyList();
for (Router router : routers) {
try {
router.stop();
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/directory/AbstractDirectory.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/directory/AbstractDirectory.java
index 6f87362..a6b41ea 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/directory/AbstractDirectory.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/directory/AbstractDirectory.java
@@ -17,21 +17,42 @@
package org.apache.dubbo.rpc.cluster.directory;
import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.Version;
+import org.apache.dubbo.common.config.Configuration;
+import org.apache.dubbo.common.config.ConfigurationUtils;
+import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.common.threadpool.manager.ExecutorRepository;
+import org.apache.dubbo.common.utils.ConcurrentHashSet;
+import org.apache.dubbo.common.utils.NetUtils;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
+import org.apache.dubbo.rpc.RpcContext;
import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.cluster.Directory;
import org.apache.dubbo.rpc.cluster.Router;
import org.apache.dubbo.rpc.cluster.RouterChain;
+import org.apache.dubbo.rpc.cluster.router.state.BitList;
import org.apache.dubbo.rpc.cluster.support.ClusterUtils;
import org.apache.dubbo.rpc.model.ApplicationModel;
+import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
+import java.util.LinkedList;
import java.util.List;
import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
import static org.apache.dubbo.common.constants.CommonConstants.DUBBO;
import static org.apache.dubbo.common.constants.CommonConstants.INTERFACE_KEY;
@@ -43,7 +64,6 @@ import static org.apache.dubbo.rpc.cluster.Constants.REFER_KEY;
/**
* Abstract implementation of Directory: Invoker list returned from this Directory's list method have been filtered by Routers
- *
*/
public abstract class AbstractDirectory<T> implements Directory<T> {
@@ -60,6 +80,44 @@ public abstract class AbstractDirectory<T> implements Directory<T> {
protected final Map<String, String> queryMap;
+ protected final Lock invokerLock = new ReentrantLock();
+
+ /**
+ * All invokers from registry
+ */
+ protected volatile BitList<Invoker<T>> invokers;
+
+ /**
+ * Valid Invoker. All invokers from registry exclude unavailable and disabled invokers.
+ */
+ protected volatile BitList<Invoker<T>> validInvokers;
+
+ /**
+ * Waiting to reconnect invokers.
+ */
+ protected volatile List<Invoker<T>> invokersToReconnect = new CopyOnWriteArrayList<>();
+
+ /**
+ * Disabled Invokers. Will not be recovered in reconnect task, but be recovered if registry remove it.
+ */
+ protected final Set<Invoker<T>> disabledInvokers = new ConcurrentHashSet<>();
+
+ private final Semaphore checkConnectivityPermit = new Semaphore(1);
+
+ private final ScheduledExecutorService connectivityExecutor;
+
+ private volatile ScheduledFuture<?> connectivityCheckFuture;
+
+ /**
+ * The max count of invokers for each reconnect task select to try to reconnect.
+ */
+ private final int reconnectTaskTryCount;
+
+ /**
+ * The period of reconnect task if needed. (in ms)
+ */
+ private final int reconnectTaskPeriod;
+
public AbstractDirectory(URL url) {
this(url, null, false);
}
@@ -76,7 +134,7 @@ public abstract class AbstractDirectory<T> implements Directory<T> {
this.url = url.removeAttribute(REFER_KEY).removeAttribute(MONITOR_KEY);
Map<String, String> queryMap;
- Object referParams = url.getAttribute(REFER_KEY);
+ Object referParams = url.getAttribute(REFER_KEY);
if (referParams instanceof Map) {
queryMap = (Map<String, String>) referParams;
this.consumerUrl = (URL) url.getAttribute(CONSUMER_URL_KEY);
@@ -94,10 +152,10 @@ public abstract class AbstractDirectory<T> implements Directory<T> {
String consumedProtocol = queryMap.get(PROTOCOL_KEY) == null ? DUBBO : queryMap.get(PROTOCOL_KEY);
URL consumerUrlFrom = this.url
- .setHost(host)
- .setPort(0)
- .setProtocol(consumedProtocol)
- .setPath(path == null ? queryMap.get(INTERFACE_KEY) : path);
+ .setHost(host)
+ .setPort(0)
+ .setProtocol(consumedProtocol)
+ .setPath(path == null ? queryMap.get(INTERFACE_KEY) : path);
if (isUrlFromRegistry) {
// reserve parameters if url is already a consumer url
consumerUrlFrom = consumerUrlFrom.clearParameters().setServiceModel(url.getServiceModel()).setScopeModel(url.getScopeModel());
@@ -105,6 +163,10 @@ public abstract class AbstractDirectory<T> implements Directory<T> {
this.consumerUrl = consumerUrlFrom.addParameters(queryMap).removeAttribute(MONITOR_KEY);
}
+ this.connectivityExecutor = applicationModel.getExtensionLoader(ExecutorRepository.class).getDefaultExtension().getConnectivityScheduledExecutor();
+ Configuration configuration = ConfigurationUtils.getGlobalConfiguration(url.getOrDefaultModuleModel());
+ this.reconnectTaskTryCount = configuration.getInt(CommonConstants.RECONNECT_TASK_TRY_COUNT, CommonConstants.DEFAULT_RECONNECT_TASK_TRY_COUNT);
+ this.reconnectTaskPeriod = configuration.getInt(CommonConstants.RECONNECT_TASK_PERIOD, CommonConstants.DEFAULT_RECONNECT_TASK_PERIOD);
setRouterChain(routerChain);
}
@@ -114,7 +176,24 @@ public abstract class AbstractDirectory<T> implements Directory<T> {
throw new RpcException("Directory already destroyed .url: " + getUrl());
}
- return doList(invocation);
+ BitList<Invoker<T>> availableInvokers;
+ if (validInvokers != null) {
+ availableInvokers = validInvokers;
+ } else {
+ availableInvokers = invokers;
+ }
+
+ List<Invoker<T>> routedResult = doList(availableInvokers, invocation);
+ if (routedResult.isEmpty()) {
+ logger.warn("No provider available after connectivity filter for the service " + getConsumerUrl().getServiceKey()
+ + " all validInvokers' size: " + (validInvokers == null ? 0 : validInvokers.size())
+ + "/ all routed invokers' size: " + routedResult.size()
+ + "/ all invokers' size: " + (invokers == null ? 0 : invokers.size())
+ + " from registry " + getUrl().getAddress()
+ + " on the consumer " + NetUtils.getLocalHost()
+ + " using the dubbo version " + Version.getVersion() + ".");
+ }
+ return Collections.unmodifiableList(routedResult);
}
@Override
@@ -151,6 +230,14 @@ public abstract class AbstractDirectory<T> implements Directory<T> {
@Override
public void destroy() {
destroyed = true;
+ if (invokers != null) {
+ invokers.clear();
+ }
+ if (validInvokers != null) {
+ validInvokers.clear();
+ }
+ invokersToReconnect.clear();
+ disabledInvokers.clear();
}
@Override
@@ -158,6 +245,164 @@ public abstract class AbstractDirectory<T> implements Directory<T> {
// do nothing by default
}
- protected abstract List<Invoker<T>> doList(Invocation invocation) throws RpcException;
+ @Override
+ public void addInvalidateInvoker(Invoker<T> invoker) {
+ invokerLock.lock();
+ try {
+ // 1. remove this invoker from validInvokers list, this invoker will not be listed in the next time
+ if (validInvokers.remove(invoker)) {
+ // 2. add this invoker to reconnect list
+ invokersToReconnect.add(invoker);
+ // 3. try start check connectivity task
+ checkConnectivity();
+ }
+ } finally {
+ invokerLock.unlock();
+ }
+ }
+
+ public void checkConnectivity() {
+ // try to submit task, to ensure there is only one task at most for each directory
+ if (checkConnectivityPermit.tryAcquire()) {
+ this.connectivityCheckFuture = connectivityExecutor.schedule(() -> {
+ try {
+ if (isDestroyed()) {
+ return;
+ }
+ RpcContext.getServiceContext().setConsumerUrl(getConsumerUrl());
+ List<Invoker<T>> needDeleteList = new ArrayList<>();
+ List<Invoker<T>> invokersToTry = new ArrayList<>();
+
+ // 1. pick invokers from invokersToReconnect
+ // limit max reconnectTaskTryCount, prevent this task hang up all the connectivityExecutor for long time
+ if (invokersToReconnect.size() < reconnectTaskTryCount) {
+ invokersToTry.addAll(invokersToReconnect);
+ } else {
+ for (int i = 0; i < reconnectTaskTryCount; i++) {
+ Invoker<T> tInvoker = invokersToReconnect.get(ThreadLocalRandom.current().nextInt(invokersToReconnect.size()));
+ if (!invokersToTry.contains(tInvoker)) {
+ // ignore if is selected, invokersToTry's size is always smaller than reconnectTaskTryCount + 1
+ invokersToTry.add(tInvoker);
+ }
+ }
+ }
+
+ // 2. try to check the invoker's status
+ for (Invoker<T> invoker : invokersToTry) {
+ if (invokers.contains(invoker)) {
+ if (invoker.isAvailable()) {
+ needDeleteList.add(invoker);
+ }
+ } else {
+ needDeleteList.add(invoker);
+ }
+ }
+
+ // 3. recover valid invoker
+ invokerLock.lock();
+ try {
+ for (Invoker<T> tInvoker : needDeleteList) {
+ if (invokers.contains(tInvoker)) {
+ validInvokers.add(tInvoker);
+ logger.info("Recover service address: " + tInvoker.getUrl() + " from invalid list.");
+ }
+ invokersToReconnect.remove(tInvoker);
+ }
+ } finally {
+ invokerLock.unlock();
+ }
+ } finally {
+ checkConnectivityPermit.release();
+ }
+
+ // 4. submit new task if it has more to recover
+ if (!invokersToReconnect.isEmpty()) {
+ checkConnectivity();
+ }
+ }, reconnectTaskPeriod, TimeUnit.MILLISECONDS);
+ }
+ }
+
+ /**
+ * Refresh invokers from total invokers
+ * 1. all the invokers in need to reconnect list should be removed in the valid invokers list
+ * 2. all the invokers in disabled invokers list should be removed in the valid invokers list
+ * 3. all the invokers disappeared from total invokers should be removed in the need to reconnect list
+ * 4. all the invokers disappeared from total invokers should be removed in the disabled invokers list
+ */
+ public synchronized void refreshInvoker() {
+ invokerLock.lock();
+ try {
+ if (invokers != null) {
+ BitList<Invoker<T>> copiedInvokers = invokers.clone();
+ refreshInvokers(copiedInvokers, invokersToReconnect);
+ refreshInvokers(copiedInvokers, disabledInvokers);
+ validInvokers = copiedInvokers;
+ }
+ } finally {
+ invokerLock.unlock();
+ }
+ }
+
+ private void refreshInvokers(BitList<Invoker<T>> targetInvokers, Collection<Invoker<T>> invokersToRemove) {
+ List<Invoker<T>> needToRemove = new LinkedList<>();
+ for (Invoker<T> tInvoker : invokersToRemove) {
+ if (targetInvokers.contains(tInvoker)) {
+ targetInvokers.remove(tInvoker);
+ } else {
+ needToRemove.add(tInvoker);
+ }
+ }
+ invokersToRemove.removeAll(needToRemove);
+ }
+
+ @Override
+ public void addDisabledInvoker(Invoker<T> invoker) {
+ invokerLock.lock();
+ try {
+ if (invokers.contains(invoker)) {
+ disabledInvokers.add(invoker);
+ validInvokers.remove(invoker);
+ logger.info("Disable service address: " + invoker.getUrl() + ".");
+ }
+ } finally {
+ invokerLock.unlock();
+ }
+ }
+
+ @Override
+ public void recoverDisabledInvoker(Invoker<T> invoker) {
+ invokerLock.lock();
+ try {
+ if (disabledInvokers.remove(invoker)) {
+ try {
+ validInvokers.add(invoker);
+ logger.info("Recover service address: " + invoker.getUrl() + " from disabled list.");
+ } catch (Throwable ignore) {
+
+ }
+ }
+ } finally {
+ invokerLock.unlock();
+ }
+ }
+
+ /**
+ * for ut only
+ */
+ @Deprecated
+ public Semaphore getCheckConnectivityPermit() {
+ return checkConnectivityPermit;
+ }
+
+ /**
+ * for ut only
+ */
+ @Deprecated
+ public ScheduledFuture<?> getConnectivityCheckFuture() {
+ return connectivityCheckFuture;
+ }
+
+ protected abstract List<Invoker<T>> doList(BitList<Invoker<T>> invokers, Invocation invocation) throws RpcException;
}
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/directory/StaticDirectory.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/directory/StaticDirectory.java
index 2b1a3d7..e11f4e0 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/directory/StaticDirectory.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/directory/StaticDirectory.java
@@ -24,8 +24,8 @@ import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.cluster.RouterChain;
+import org.apache.dubbo.rpc.cluster.router.state.BitList;
-import java.util.Collections;
import java.util.List;
/**
@@ -34,8 +34,6 @@ import java.util.List;
public class StaticDirectory<T> extends AbstractDirectory<T> {
private static final Logger logger = LoggerFactory.getLogger(StaticDirectory.class);
- private final List<Invoker<T>> invokers;
-
public StaticDirectory(List<Invoker<T>> invokers) {
this(null, invokers, null);
}
@@ -53,7 +51,8 @@ public class StaticDirectory<T> extends AbstractDirectory<T> {
if (CollectionUtils.isEmpty(invokers)) {
throw new IllegalArgumentException("invokers == null");
}
- this.invokers = invokers;
+ this.invokers = new BitList<>(invokers);
+ this.validInvokers = this.invokers.clone();
}
@Override
@@ -71,7 +70,7 @@ public class StaticDirectory<T> extends AbstractDirectory<T> {
if (isDestroyed()) {
return false;
}
- for (Invoker<T> invoker : invokers) {
+ for (Invoker<T> invoker : validInvokers) {
if (invoker.isAvailable()) {
return true;
}
@@ -84,11 +83,10 @@ public class StaticDirectory<T> extends AbstractDirectory<T> {
if (isDestroyed()) {
return;
}
- super.destroy();
for (Invoker<T> invoker : invokers) {
invoker.destroy();
}
- invokers.clear();
+ super.destroy();
}
public void buildRouterChain() {
@@ -98,17 +96,26 @@ public class StaticDirectory<T> extends AbstractDirectory<T> {
this.setRouterChain(routerChain);
}
+ public void notify(List<Invoker<T>> invokers) {
+ this.invokers = new BitList<>(invokers);
+ refreshInvoker();
+ if (routerChain != null) {
+ routerChain.setInvokers(this.invokers);
+ }
+ }
+
@Override
- protected List<Invoker<T>> doList(Invocation invocation) throws RpcException {
- List<Invoker<T>> finalInvokers = invokers;
+ protected List<Invoker<T>> doList(BitList<Invoker<T>> invokers, Invocation invocation) throws RpcException {
if (routerChain != null) {
try {
- finalInvokers = routerChain.route(getConsumerUrl(), invocation);
+ List<Invoker<T>> finalInvokers = routerChain.route(getConsumerUrl(), invokers, invocation);
+ return finalInvokers == null ? BitList.emptyList() : finalInvokers;
} catch (Throwable t) {
logger.error("Failed to execute router: " + getUrl() + ", cause: " + t.getMessage(), t);
+ return BitList.emptyList();
}
}
- return finalInvokers == null ? Collections.emptyList() : finalInvokers;
+ return invokers;
}
}
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/RouterResult.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/RouterResult.java
new file mode 100644
index 0000000..5d08d5b
--- /dev/null
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/RouterResult.java
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.dubbo.rpc.cluster.router;
+
+import java.util.List;
+
+public class RouterResult<T> {
+ private final boolean needContinueRoute;
+ private final List<T> result;
+ private final String message;
+
+ public RouterResult(List<T> result) {
+ this.needContinueRoute = true;
+ this.result = result;
+ this.message = null;
+ }
+
+ public RouterResult(List<T> result, String message) {
+ this.needContinueRoute = true;
+ this.result = result;
+ this.message = message;
+ }
+
+ public RouterResult(boolean needContinueRoute, List<T> result, String message) {
+ this.needContinueRoute = needContinueRoute;
+ this.result = result;
+ this.message = message;
+ }
+
+ public boolean isNeedContinueRoute() {
+ return needContinueRoute;
+ }
+
+ public List<T> getResult() {
+ return result;
+ }
+
+ public String getMessage() {
+ return message;
+ }
+}
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/RouterSnapshotNode.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/RouterSnapshotNode.java
new file mode 100644
index 0000000..28ac112
--- /dev/null
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/RouterSnapshotNode.java
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.dubbo.rpc.cluster.router;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.rpc.Invoker;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class RouterSnapshotNode<T> {
+ private final String name;
+ private final int beforeSize;
+ private int afterSize;
+ private String routerMessage;
+ private List<Invoker<T>> outputInvokers;
+ private RouterSnapshotNode<T> nextNode;
+
+ public RouterSnapshotNode(String name, int beforeSize) {
+ this.name = name;
+ this.beforeSize = beforeSize;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public int getBeforeSize() {
+ return beforeSize;
+ }
+
+ public int getAfterSize() {
+ return afterSize;
+ }
+
+ public String getRouterMessage() {
+ return routerMessage;
+ }
+
+ public void setRouterMessage(String routerMessage) {
+ this.routerMessage = routerMessage;
+ }
+
+ public List<Invoker<T>> getOutputInvokers() {
+ return outputInvokers;
+ }
+
+ public void setOutputInvokers(List<Invoker<T>> outputInvokers) {
+ this.outputInvokers = outputInvokers;
+ this.afterSize = outputInvokers == null ? 0 : outputInvokers.size();
+ }
+
+ public RouterSnapshotNode<T> getNextNode() {
+ return nextNode;
+ }
+
+ public void appendNode(RouterSnapshotNode<T> nextNode) {
+ if (this.nextNode != null) {
+ RouterSnapshotNode<T> node = this.nextNode;
+ while (node.nextNode != null) {
+ node = node.nextNode;
+ }
+ node.nextNode = nextNode;
+ } else {
+ this.nextNode = nextNode;
+ }
+ }
+
+ @Override
+ public String toString() {
+ return toString(1);
+ }
+
+ public String toString(int level) {
+ StringBuilder stringBuilder = new StringBuilder();
+ stringBuilder.append("[ ")
+ .append(name)
+ .append(" ")
+ .append("invokers: ")
+ .append(beforeSize).append(" -> ").append(afterSize)
+ .append(" ")
+ .append(routerMessage == null ? "" : routerMessage)
+ .append("] ")
+ .append(outputInvokers == null ? "" :
+ outputInvokers.subList(0, Math.min(5, outputInvokers.size()))
+ .stream()
+ .map(Invoker::getUrl)
+ .map(URL::getAddress)
+ .collect(Collectors.joining(",")));
+
+ if (outputInvokers != null && outputInvokers.size() > 5) {
+ stringBuilder.append("...");
+ }
+ if (nextNode != null) {
+ stringBuilder.append("\n");
+ for (int i = 0; i < level; i++) {
+ stringBuilder.append(" ");
+ }
+ stringBuilder.append(nextNode.toString(level + 1));
+ }
+ return stringBuilder.toString();
+ }
+}
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/condition/ConditionRouter.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/condition/ConditionRouter.java
index 17f3afa..7144bc5 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/condition/ConditionRouter.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/condition/ConditionRouter.java
@@ -28,6 +28,7 @@ import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.cluster.Constants;
import org.apache.dubbo.rpc.cluster.router.AbstractRouter;
+import org.apache.dubbo.rpc.cluster.router.RouterResult;
import java.text.ParseException;
import java.util.ArrayList;
@@ -176,23 +177,24 @@ public class ConditionRouter extends AbstractRouter {
}
@Override
- public <T> List<Invoker<T>> route(List<Invoker<T>> invokers, URL url, Invocation invocation)
- throws RpcException {
+ public <T> RouterResult<Invoker<T>> route(List<Invoker<T>> invokers, URL url,
+ Invocation invocation, boolean needToPrintMessage) throws RpcException {
+
if (!enabled) {
- return invokers;
+ return new RouterResult<>(invokers);
}
if (CollectionUtils.isEmpty(invokers)) {
- return invokers;
+ return new RouterResult<>(invokers);
}
try {
if (!matchWhen(url, invocation)) {
- return invokers;
+ return new RouterResult<>(invokers);
}
List<Invoker<T>> result = new ArrayList<Invoker<T>>();
if (thenCondition == null) {
logger.warn("The current consumer in the service blacklist. consumer: " + NetUtils.getLocalHost() + ", service: " + url.getServiceKey());
- return result;
+ return new RouterResult<>(result);
}
for (Invoker<T> invoker : invokers) {
if (matchThen(invoker.getUrl(), url)) {
@@ -200,15 +202,15 @@ public class ConditionRouter extends AbstractRouter {
}
}
if (!result.isEmpty()) {
- return result;
+ return new RouterResult<>(result);
} else if (this.isForce()) {
logger.warn("The route result is empty and force execute. consumer: " + NetUtils.getLocalHost() + ", service: " + url.getServiceKey() + ", router: " + url.getParameterAndDecoded(RULE_KEY));
- return result;
+ return new RouterResult<>(result);
}
} catch (Throwable t) {
logger.error("Failed to execute condition router rule: " + getUrl() + ", invokers: " + invokers + ", cause: " + t.getMessage(), t);
}
- return invokers;
+ return new RouterResult<>(invokers);
}
@Override
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/condition/config/ListenableRouter.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/condition/config/ListenableRouter.java
index a33f4ea..ee08c94 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/condition/config/ListenableRouter.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/condition/config/ListenableRouter.java
@@ -30,6 +30,7 @@ import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.cluster.Router;
import org.apache.dubbo.rpc.cluster.router.AbstractRouter;
+import org.apache.dubbo.rpc.cluster.router.RouterResult;
import org.apache.dubbo.rpc.cluster.router.condition.ConditionRouter;
import org.apache.dubbo.rpc.cluster.router.condition.config.model.ConditionRouterRule;
import org.apache.dubbo.rpc.cluster.router.condition.config.model.ConditionRuleParser;
@@ -79,17 +80,18 @@ public abstract class ListenableRouter extends AbstractRouter implements Configu
}
@Override
- public <T> List<Invoker<T>> route(List<Invoker<T>> invokers, URL url, Invocation invocation) throws RpcException {
+ public <T> RouterResult<Invoker<T>> route(List<Invoker<T>> invokers, URL url,
+ Invocation invocation, boolean needToPrintMessage) throws RpcException {
if (CollectionUtils.isEmpty(invokers) || conditionRouters.size() == 0) {
- return invokers;
+ return new RouterResult<>(invokers);
}
// We will check enabled status inside each router.
for (Router router : conditionRouters) {
- invokers = router.route(invokers, url, invocation);
+ invokers = router.route(invokers, url, invocation, needToPrintMessage).getResult();
}
- return invokers;
+ return new RouterResult<>(invokers);
}
@Override
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mesh/route/MeshRuleRouter.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mesh/route/MeshRuleRouter.java
index 5b95978..4103d2c 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mesh/route/MeshRuleRouter.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mesh/route/MeshRuleRouter.java
@@ -24,6 +24,7 @@ import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.cluster.Router;
+import org.apache.dubbo.rpc.cluster.router.RouterResult;
import org.apache.dubbo.rpc.cluster.router.mesh.rule.VsDestinationGroup;
import org.apache.dubbo.rpc.cluster.router.mesh.rule.destination.DestinationRule;
import org.apache.dubbo.rpc.cluster.router.mesh.rule.destination.DestinationRuleSpec;
@@ -75,12 +76,13 @@ public class MeshRuleRouter implements Router, VsDestinationGroupRuleListener {
}
@Override
- public <T> List<Invoker<T>> route(List<Invoker<T>> invokers, URL url, Invocation invocation) throws RpcException {
+ public <T> RouterResult<Invoker<T>> route(List<Invoker<T>> invokers, URL url,
+ Invocation invocation, boolean needToPrintMessage) throws RpcException {
List<DubboRouteDestination> routeDestination = getDubboRouteDestination(invocation);
if (routeDestination == null) {
- return invokers;
+ return new RouterResult<>(invokers);
} else {
DubboRouteDestination dubboRouteDestination = routeDestination.get(ThreadLocalRandom.current().nextInt(routeDestination.size()));
@@ -99,7 +101,7 @@ public class MeshRuleRouter implements Router, VsDestinationGroupRuleListener {
result = subsetMapCopy.get(subset);
if (CollectionUtils.isNotEmpty(result)) {
- return (List) result;
+ return new RouterResult<>((List) result);
}
dubboRouteDestination = dubboDestination.getFallback();
@@ -111,11 +113,11 @@ public class MeshRuleRouter implements Router, VsDestinationGroupRuleListener {
subset = dubboDestination.getSubset();
} while (true);
- return null;
+ return new RouterResult<>(null);
}
}
- return invokers;
+ return new RouterResult<>(invokers);
}
@Override
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mock/MockInvokersSelector.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mock/MockInvokersSelector.java
index c2ca880..b3bd9f1 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mock/MockInvokersSelector.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mock/MockInvokersSelector.java
@@ -22,6 +22,7 @@ import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.cluster.router.AbstractRouter;
+import org.apache.dubbo.rpc.cluster.router.RouterResult;
import java.util.ArrayList;
import java.util.List;
@@ -43,23 +44,23 @@ public class MockInvokersSelector extends AbstractRouter {
}
@Override
- public <T> List<Invoker<T>> route(final List<Invoker<T>> invokers,
- URL url, final Invocation invocation) throws RpcException {
+ public <T> RouterResult<Invoker<T>> route(List<Invoker<T>> invokers, URL url,
+ Invocation invocation, boolean needToPrintMessage) throws RpcException {
if (CollectionUtils.isEmpty(invokers)) {
- return invokers;
+ return new RouterResult<>(invokers);
}
if (invocation.getObjectAttachments() == null) {
- return getNormalInvokers(invokers);
+ return new RouterResult<>(getNormalInvokers(invokers));
} else {
String value = (String) invocation.getObjectAttachments().get(INVOCATION_NEED_MOCK);
if (value == null) {
- return getNormalInvokers(invokers);
+ return new RouterResult<>(getNormalInvokers(invokers));
} else if (Boolean.TRUE.toString().equalsIgnoreCase(value)) {
- return getMockedInvokers(invokers);
+ return new RouterResult<>(getMockedInvokers(invokers));
}
}
- return invokers;
+ return new RouterResult<>(invokers);
}
private <T> List<Invoker<T>> getMockedInvokers(final List<Invoker<T>> invokers) {
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/script/ScriptRouter.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/script/ScriptRouter.java
index 1e7d075..aeecb22 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/script/ScriptRouter.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/script/ScriptRouter.java
@@ -25,6 +25,7 @@ import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.RpcContext;
import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.cluster.router.AbstractRouter;
+import org.apache.dubbo.rpc.cluster.router.RouterResult;
import javax.script.Bindings;
import javax.script.Compilable;
@@ -121,21 +122,23 @@ public class ScriptRouter extends AbstractRouter {
});
}
+
@Override
- public <T> List<Invoker<T>> route(List<Invoker<T>> invokers, URL url, Invocation invocation) throws RpcException {
+ public <T> RouterResult<Invoker<T>> route(List<Invoker<T>> invokers, URL url,
+ Invocation invocation, boolean needToPrintMessage) throws RpcException {
if (engine == null || function == null) {
- return invokers;
+ return new RouterResult<>(invokers);
}
Bindings bindings = createBindings(invokers, invocation);
- return getRoutedInvokers(AccessController.doPrivileged((PrivilegedAction<Object>) () -> {
+ return new RouterResult<>(getRoutedInvokers(AccessController.doPrivileged((PrivilegedAction<Object>) () -> {
try {
return function.eval(bindings);
} catch (ScriptException e) {
logger.error("route error, rule has been ignored. rule: " + rule + ", method:" +
- invocation.getMethodName() + ", url: " + RpcContext.getContext().getUrl(), e);
+ invocation.getMethodName() + ", url: " + RpcContext.getContext().getUrl(), e);
return invokers;
}
- }, accessControlContext));
+ }, accessControlContext)));
}
/**
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/AbstractStateRouter.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/AbstractStateRouter.java
index a52410b..686e598 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/AbstractStateRouter.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/AbstractStateRouter.java
@@ -84,23 +84,27 @@ public abstract class AbstractStateRouter implements StateRouter {
}
@Override
- public <T> BitList<Invoker<T>> route(BitList<Invoker<T>> invokers, RouterCache<T> cache, URL url,
- Invocation invocation) throws RpcException {
+ public <T> StateRouterResult<Invoker<T>> route(BitList<Invoker<T>> invokers, RouterCache<T> cache, URL url,
+ Invocation invocation, boolean needToPrintMessage) throws RpcException {
List<String> tags = getTags(url, invocation);
if (tags == null) {
- return invokers;
+ return new StateRouterResult<>(invokers);
}
for (String tag : tags) {
BitList<Invoker<T>> tagInvokers = cache.getAddrPool().get(tag);
if (tagMatchFail(tagInvokers)) {
continue;
}
- return tagInvokers.intersect(invokers, invokers.getUnmodifiableList());
+ if (needToPrintMessage) {
+ return new StateRouterResult<>(invokers.and(tagInvokers), "use tag " + tag + " to route");
+ } else {
+ return new StateRouterResult<>(invokers.and(tagInvokers));
+ }
}
- return invokers;
+ return new StateRouterResult<>(invokers);
}
protected List<String> getTags(URL url, Invocation invocation) {
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/BitList.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/BitList.java
index 71456cb..09f250d 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/BitList.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/BitList.java
@@ -16,217 +16,442 @@
*/
package org.apache.dubbo.rpc.cluster.router.state;
-import java.util.Arrays;
-import java.util.Collection;
+import org.apache.dubbo.common.utils.CollectionUtils;
+
+import java.util.AbstractList;
+import java.util.ArrayList;
+import java.util.BitSet;
+import java.util.Collections;
import java.util.Iterator;
+import java.util.LinkedList;
import java.util.List;
import java.util.ListIterator;
-
-import org.roaringbitmap.IntIterator;
-import org.roaringbitmap.RoaringBitmap;
+import java.util.NoSuchElementException;
/**
* BitList based on BitMap implementation.
+ * BitList is consists of `originList`, `rootSet` and `tailList`.
+ *
+ * originList: Initial elements of the list. This list will not be changed
+ * in modification actions (expect clear all).
+ * rootSet: A bitMap to store the indexes of originList are still exist.
+ * Most of the modification actions are operated on this bitMap.
+ * tailList: An additional list for BitList. Worked when adding totally new
+ * elements to list. These elements will be appended to the last
+ * of the BitList.
+ *
+ * An example of BitList:
+ * originList: A B C D E (5 elements)
+ * rootSet: x v x v v
+ * 0 1 0 1 1 (5 elements)
+ * tailList: F G H (3 elements)
+ * resultList: B D E F G H (6 elements)
+ *
* @param <E>
* @since 3.0
*/
-public class BitList<E> implements List<E> {
- private final RoaringBitmap rootMap;
- private final List<E> unmodifiableList;
+public class BitList<E> extends AbstractList<E> {
+ private final BitSet rootSet;
+ private final List<E> originList;
+ private final static BitList emptyList = new BitList(Collections.emptyList());
+ private volatile List<E> tailList = null;
+
+ public BitList(List<E> originList) {
+ this(originList, false);
+ }
- public BitList(List<E> unmodifiableList, boolean empty) {
- this.unmodifiableList = unmodifiableList;
- this.rootMap = new RoaringBitmap();
+ public BitList(List<E> originList, boolean empty) {
+ this.originList = new ArrayList<>(originList);
+ this.rootSet = new BitSet();
if (!empty) {
- this.rootMap.add(0L, unmodifiableList.size());
+ this.rootSet.set(0, originList.size());
}
}
- public BitList(List<E> unmodifiableList, RoaringBitmap rootMap) {
- this.unmodifiableList = unmodifiableList;
- this.rootMap = rootMap;
+ public BitList(List<E> originList, BitSet rootSet, List<E> tailList) {
+ this.originList = new ArrayList<>(originList);
+ this.rootSet = rootSet;
+ this.tailList = tailList;
}
- public BitList(List<E> unmodifiableList) {
- this(unmodifiableList, false);
- }
-
- public List<E> getUnmodifiableList() {
- return unmodifiableList;
+ // Provided by BitList only
+ public List<E> getOriginList() {
+ return originList;
}
public void addIndex(int index) {
- this.rootMap.add(index);
+ this.rootSet.set(index);
}
- public BitList<E> intersect(List<E> b, List<E> totalList) {
- RoaringBitmap resultMap = rootMap.clone();
- resultMap.and(((BitList)b).rootMap);
- return new BitList<>(totalList, resultMap);
+ /**
+ * And operation between two bitList. Return a new cloned list.
+ * TailList in source bitList will be totally saved even if it is not appeared in the target bitList.
+ *
+ * @param target target bitList
+ * @return a new bitList only contains those elements contain in both two list and source bitList's tailList
+ */
+ public BitList<E> and(BitList<E> target) {
+ BitSet resultSet = (BitSet) rootSet.clone();
+ resultSet.and(target.rootSet);
+ return new BitList<>(originList, resultSet, tailList);
}
- @Override
- public int size() {
- return rootMap.getCardinality();
+ public boolean hasMoreElementInTailList() {
+ return CollectionUtils.isNotEmpty(tailList) && tailList.size() > 0;
}
- @Override
- public boolean isEmpty() {
- return rootMap.isEmpty();
+ public List<E> getTailList() {
+ return tailList;
}
- @Override
- public boolean contains(Object o) {
- int idx = unmodifiableList.indexOf(o);
- return idx >= 0 && rootMap.contains(idx);
+ @SuppressWarnings("unchecked")
+ public static <T> BitList<T> emptyList() {
+ return emptyList;
}
+ // Provided by JDK List interface
@Override
- public Iterator<E> iterator() {
- return new Iterator<E>() {
- private int prev = -1;
-
- @Override
- public boolean hasNext() {
- return -1 != rootMap.nextValue(prev + 1);
- }
-
- @Override
- public E next() {
- prev = (int)rootMap.nextValue(prev + 1);
- return unmodifiableList.get(prev);
- }
-
- @Override
- public void remove() {
- rootMap.remove(prev);
- }
- };
+ public int size() {
+ return rootSet.cardinality() + (CollectionUtils.isNotEmpty(tailList) ? tailList.size() : 0);
}
@Override
- public Object[] toArray() {
- int size = size();
- Object[] obj = new Object[size];
- for (int i = 0; i < size; i++) {
- obj[i] = unmodifiableList.get(rootMap.select(i));
- }
- return obj;
+ public boolean contains(Object o) {
+ int idx = originList.indexOf(o);
+ return (idx >= 0 && rootSet.get(idx)) || (CollectionUtils.isNotEmpty(tailList) && tailList.contains(o));
}
@Override
- public <T> T[] toArray(T[] a) {
- int size = size();
- Object[] arr = toArray();
- if (a.length < size)
- // Make a new array of a's runtime type, but my contents:
- { return (T[])Arrays.copyOf(arr, size, a.getClass()); }
- System.arraycopy(arr, 0, a, 0, size);
- if (a.length > size) { a[size] = null; }
- return null;
+ public Iterator<E> iterator() {
+ return new BitListIterator<>(this, 0);
}
+ /**
+ * If the element to added is appeared in originList even if it is not in rootSet,
+ * directly set its index in rootSet to true. (This may change the order of elements.)
+ *
+ * If the element is not contained in originList, allocate tailList and add to tailList.
+ *
+ * Notice: It is not recommended adding duplicated element.
+ */
@Override
public boolean add(E e) {
- throw new UnsupportedOperationException();
+ int index = originList.indexOf(e);
+ if (index > -1) {
+ rootSet.set(index);
+ return true;
+ } else {
+ if (tailList == null) {
+ tailList = new LinkedList<>();
+ }
+ return tailList.add(e);
+ }
}
+ /**
+ * If the element to added is appeared in originList,
+ * directly set its index in rootSet to false. (This may change the order of elements.)
+ *
+ * If the element is not contained in originList, try to remove from tailList.
+ */
@Override
public boolean remove(Object o) {
- int idx = unmodifiableList.indexOf(o);
- if (idx > -1) {
- rootMap.remove(idx);
+ int idx = originList.indexOf(o);
+ if (idx > -1 && rootSet.get(idx)) {
+ rootSet.set(idx, false);
+ return true;
}
- return true;
- }
-
- @Override
- public boolean containsAll(Collection<?> c) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public boolean addAll(Collection<? extends E> c) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public boolean addAll(int index, Collection<? extends E> c) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public boolean removeAll(Collection<?> c) {
- for (Object o : c) {
- remove(o);
+ if (CollectionUtils.isNotEmpty(tailList)) {
+ return tailList.remove(o);
}
- return true;
- }
-
- @Override
- public boolean retainAll(Collection<?> c) {
return false;
}
+ /**
+ * Caution: This operation will clear originList for removing references purpose.
+ * This may change the default behaviour when adding new element later.
+ */
@Override
public void clear() {
- rootMap.clear();
+ rootSet.clear();
+ // to remove references
+ originList.clear();
+ if (CollectionUtils.isNotEmpty(tailList)) {
+ tailList.clear();
+ }
}
@Override
public E get(int index) {
- int real = rootMap.select(index);
- return unmodifiableList.get(real);
- }
-
- @Override
- public E set(int index, E element) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public void add(int index, E element) {
- throw new UnsupportedOperationException();
+ int bitIndex = -1;
+ if (index < 0) {
+ throw new IndexOutOfBoundsException();
+ }
+ if (index >= rootSet.cardinality()) {
+ if (CollectionUtils.isNotEmpty(tailList)) {
+ return tailList.get(index - rootSet.cardinality());
+ } else {
+ throw new IndexOutOfBoundsException();
+ }
+ } else {
+ for (int i = 0; i <= index; i++) {
+ bitIndex = rootSet.nextSetBit(bitIndex + 1);
+ }
+ return originList.get(bitIndex);
+ }
}
@Override
public E remove(int index) {
- rootMap.remove(index);
- return null;
+ int bitIndex = -1;
+ if (index >= rootSet.cardinality()) {
+ if (CollectionUtils.isNotEmpty(tailList)) {
+ return tailList.remove(index - rootSet.cardinality());
+ } else {
+ throw new IndexOutOfBoundsException();
+ }
+ } else {
+ for (int i = 0; i <= index; i++) {
+ bitIndex = rootSet.nextSetBit(bitIndex + 1);
+ }
+ rootSet.set(bitIndex, false);
+ return originList.get(bitIndex);
+ }
}
@Override
public int indexOf(Object o) {
- IntIterator intIterator = rootMap.getIntIterator();
- int st = 0;
- while (intIterator.hasNext()) {
- int idxInMap = intIterator.next();
- if (unmodifiableList.get(idxInMap).equals(o)) {
- return st;
+ int bitIndex = -1;
+ for (int i = 0; i < rootSet.cardinality(); i++) {
+ bitIndex = rootSet.nextSetBit(bitIndex + 1);
+ if (originList.get(bitIndex).equals(o)) {
+ return i;
+ }
+ }
+ if (CollectionUtils.isNotEmpty(tailList)) {
+ int indexInTailList = tailList.indexOf(o);
+ if (indexInTailList != -1) {
+ return indexInTailList + rootSet.cardinality();
+ } else {
+ return -1;
}
- st++;
}
return -1;
}
@Override
public int lastIndexOf(Object o) {
- throw new UnsupportedOperationException();
+ int bitIndex = -1;
+ int index = -1;
+ if (CollectionUtils.isNotEmpty(tailList)) {
+ int indexInTailList = tailList.lastIndexOf(o);
+ if (indexInTailList > -1) {
+ return indexInTailList + rootSet.cardinality();
+ }
+ }
+ for (int i = 0; i < rootSet.cardinality(); i++) {
+ bitIndex = rootSet.nextSetBit(bitIndex + 1);
+ if (originList.get(bitIndex).equals(o)) {
+ index = i;
+ }
+ }
+ return index;
}
@Override
public ListIterator<E> listIterator() {
- throw new UnsupportedOperationException();
+ return new BitListIterator<>(this, 0);
}
@Override
public ListIterator<E> listIterator(int index) {
- throw new UnsupportedOperationException();
+ return new BitListIterator<>(this, index);
}
@Override
- public List<E> subList(int fromIndex, int toIndex) {
- throw new UnsupportedOperationException();
+ public BitList<E> subList(int fromIndex, int toIndex) {
+ BitSet resultSet = (BitSet) rootSet.clone();
+ List<E> copiedTailList = tailList == null ? null : new LinkedList<>(tailList);
+ if (toIndex < size()) {
+ if (toIndex < rootSet.cardinality()) {
+ copiedTailList = null;
+ resultSet.set(toIndex, resultSet.length(), false);
+ } else {
+ copiedTailList = copiedTailList == null ? null : copiedTailList.subList(0, toIndex - rootSet.cardinality());
+ }
+ }
+ if (fromIndex > 0) {
+ if (fromIndex < rootSet.cardinality()) {
+ resultSet.set(0, fromIndex, false);
+ } else {
+ resultSet.clear();
+ copiedTailList = copiedTailList == null ? null : copiedTailList.subList(fromIndex - rootSet.cardinality(), copiedTailList.size());
+ }
+ }
+ return new BitList<>(originList, resultSet, copiedTailList);
}
+ public static class BitListIterator<E> implements ListIterator<E> {
+ private BitList<E> bitList;
+ private int index;
+ private ListIterator<E> tailListIterator;
+ private int curBitIndex = -1;
+ private boolean isInTailList = false;
+ private int lastReturnedIndex = -1;
+
+ public BitListIterator(BitList<E> bitList, int index) {
+ this.bitList = bitList;
+ this.index = index - 1;
+ for (int i = 0; i < index; i++) {
+ if (!isInTailList) {
+ curBitIndex = bitList.rootSet.nextSetBit(curBitIndex + 1);
+ if (curBitIndex == -1) {
+ if (CollectionUtils.isNotEmpty(bitList.tailList)) {
+ isInTailList = true;
+ tailListIterator = bitList.tailList.listIterator();
+ tailListIterator.next();
+ } else {
+ break;
+ }
+ }
+ } else {
+ tailListIterator.next();
+ }
+ }
+ }
+
+ @Override
+ public boolean hasNext() {
+ if (isInTailList) {
+ return tailListIterator.hasNext();
+ } else {
+ int nextBit = bitList.rootSet.nextSetBit(curBitIndex + 1);
+ if (nextBit == -1) {
+ return bitList.hasMoreElementInTailList();
+ } else {
+ return true;
+ }
+ }
+ }
+
+ @Override
+ public E next() {
+ if (isInTailList) {
+ if (tailListIterator.hasNext()) {
+ index += 1;
+ lastReturnedIndex = index;
+ }
+ return tailListIterator.next();
+ } else {
+ int nextBitIndex = bitList.rootSet.nextSetBit(curBitIndex + 1);
+ if (nextBitIndex == -1) {
+ if (bitList.hasMoreElementInTailList()) {
+ tailListIterator = bitList.tailList.listIterator();
+ isInTailList = true;
+ index += 1;
+ lastReturnedIndex = index;
+ return tailListIterator.next();
+ } else {
+ throw new NoSuchElementException();
+ }
+ } else {
+ index += 1;
+ lastReturnedIndex = index;
+ curBitIndex = nextBitIndex;
+ return bitList.originList.get(nextBitIndex);
+ }
+ }
+ }
+
+ @Override
+ public boolean hasPrevious() {
+ if (isInTailList) {
+ boolean hasPreviousInTailList = tailListIterator.hasPrevious();
+ if (hasPreviousInTailList) {
+ return true;
+ } else {
+ return bitList.rootSet.previousSetBit(bitList.rootSet.size()) != -1;
+ }
+ } else {
+ return curBitIndex != -1;
+ }
+ }
+
+ @Override
+ public E previous() {
+ if (isInTailList) {
+ boolean hasPreviousInTailList = tailListIterator.hasPrevious();
+ if (hasPreviousInTailList) {
+ lastReturnedIndex = index;
+ index -= 1;
+ return tailListIterator.previous();
+ } else {
+ int lastIndexInBit = bitList.rootSet.previousSetBit(bitList.rootSet.size());
+ if (lastIndexInBit == -1) {
+ throw new NoSuchElementException();
+ } else {
+ isInTailList = false;
+ curBitIndex = bitList.rootSet.previousSetBit(lastIndexInBit - 1);
+ lastReturnedIndex = index;
+ index -= 1;
+ return bitList.originList.get(lastIndexInBit);
+ }
+ }
+ } else {
+ if (curBitIndex == -1) {
+ throw new NoSuchElementException();
+ }
+ int nextBitIndex = curBitIndex;
+ curBitIndex = bitList.rootSet.previousSetBit(curBitIndex - 1);
+ lastReturnedIndex = index;
+ index -= 1;
+ return bitList.originList.get(nextBitIndex);
+ }
+ }
+
+ @Override
+ public int nextIndex() {
+ return hasNext() ? index + 1 : index;
+ }
+
+ @Override
+ public int previousIndex() {
+ return index;
+ }
+
+ @Override
+ public void remove() {
+ if (lastReturnedIndex == -1) {
+ throw new IllegalStateException();
+ } else {
+ if (lastReturnedIndex >= bitList.rootSet.cardinality()) {
+ tailListIterator.remove();
+ } else {
+ int bitIndex = -1;
+ for (int i = 0; i <= lastReturnedIndex; i++) {
+ bitIndex = bitList.rootSet.nextSetBit(bitIndex + 1);
+ }
+ bitList.rootSet.set(bitIndex, false);
+ }
+ }
+ if (lastReturnedIndex <= index) {
+ index -= 1;
+ }
+ }
+
+ @Override
+ public void set(E e) {
+ throw new UnsupportedOperationException("Set method is not supported in BitListIterator!");
+ }
+
+ @Override
+ public void add(E e) {
+ throw new UnsupportedOperationException("Add method is not supported in BitListIterator!");
+ }
+ }
+
+ @Override
+ public BitList<E> clone() {
+ return new BitList<>(originList, (BitSet) rootSet.clone(), tailList == null ? null : new LinkedList<>(tailList));
+ }
}
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/StateRouter.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/StateRouter.java
index cf479b0..10a1898 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/StateRouter.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/StateRouter.java
@@ -56,9 +56,28 @@ public interface StateRouter extends Comparable<StateRouter> {
* @throws RpcException
* @Since 3.0
*/
- <T> BitList<Invoker<T>> route(BitList<Invoker<T>> invokers, RouterCache<T> cache, URL url, Invocation invocation)
- throws
- RpcException;
+ @Deprecated
+ default <T> BitList<Invoker<T>> route(BitList<Invoker<T>> invokers, RouterCache<T> cache, URL url, Invocation invocation)
+ throws RpcException {
+ return null;
+ }
+
+ /***
+ * ** This method can return the state of whether routerChain needed to continue route. **
+ * Filter invokers with current routing rule and only return the invokers that comply with the rule.
+ * Caching address lists in BitMap mode improves routing performance.
+ * @param invokers invoker bit list
+ * @param cache router address cache
+ * @param url refer url
+ * @param invocation invocation
+ * @param needToPrintMessage whether to print router state. Such as `use router branch a`.
+ * @return state with route result
+ * @since 3.0
+ */
+ default <T> StateRouterResult<Invoker<T>> route(BitList<Invoker<T>> invokers, RouterCache<T> cache, URL url, Invocation invocation,
+ boolean needToPrintMessage) throws RpcException {
+ return new StateRouterResult<>(route(invokers, cache, url, invocation));
+ }
default <T> void notify(List<Invoker<T>> invokers) {
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/StateRouterResult.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/StateRouterResult.java
new file mode 100644
index 0000000..b59d569
--- /dev/null
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/StateRouterResult.java
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.dubbo.rpc.cluster.router.state;
+
+public class StateRouterResult<T> {
+ private final boolean needContinueRoute;
+ private final BitList<T> result;
+ private final String message;
+
+ public StateRouterResult(BitList<T> result) {
+ this.needContinueRoute = true;
+ this.result = result;
+ this.message = null;
+ }
+
+ public StateRouterResult(BitList<T> result, String message) {
+ this.needContinueRoute = true;
+ this.result = result;
+ this.message = message;
+ }
+
+ public StateRouterResult(boolean needContinueRoute, BitList<T> result, String message) {
+ this.needContinueRoute = needContinueRoute;
+ this.result = result;
+ this.message = message;
+ }
+
+ public boolean isNeedContinueRoute() {
+ return needContinueRoute;
+ }
+
+ public BitList<T> getResult() {
+ return result;
+ }
+
+ public String getMessage() {
+ return message;
+ }
+}
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/tag/TagDynamicStateRouter.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/tag/TagDynamicStateRouter.java
index 32dd485..5d4132e 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/tag/TagDynamicStateRouter.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/tag/TagDynamicStateRouter.java
@@ -33,6 +33,7 @@ import org.apache.dubbo.rpc.cluster.RouterChain;
import org.apache.dubbo.rpc.cluster.router.state.AbstractStateRouter;
import org.apache.dubbo.rpc.cluster.router.state.BitList;
import org.apache.dubbo.rpc.cluster.router.state.RouterCache;
+import org.apache.dubbo.rpc.cluster.router.state.StateRouterResult;
import org.apache.dubbo.rpc.cluster.router.tag.model.TagRouterRule;
import org.apache.dubbo.rpc.cluster.router.tag.model.TagRuleParser;
@@ -83,9 +84,8 @@ public class TagDynamicStateRouter extends AbstractStateRouter implements Config
}
@Override
- public <T> BitList<Invoker<T>> route(BitList<Invoker<T>> invokers, RouterCache<T> cache, URL url,
- Invocation invocation) throws RpcException {
-
+ public <T> StateRouterResult<Invoker<T>> route(BitList<Invoker<T>> invokers, RouterCache<T> cache, URL url,
+ Invocation invocation, boolean needToPrintMessage) throws RpcException {
final TagRouterRule tagRouterRuleCopy = (TagRouterRule) cache.getAddrMetadata();
@@ -95,16 +95,16 @@ public class TagDynamicStateRouter extends AbstractStateRouter implements Config
ConcurrentMap<String, BitList<Invoker<T>>> addrPool = cache.getAddrPool();
if (StringUtils.isEmpty(tag)) {
- return invokers.intersect(addrPool.get(NO_TAG), invokers.getUnmodifiableList());
+ return new StateRouterResult<>(invokers.and(addrPool.get(NO_TAG)));
} else {
BitList<Invoker<T>> result = addrPool.get(tag);
if (CollectionUtils.isNotEmpty(result) || (tagRouterRuleCopy != null && tagRouterRuleCopy.isForce())
|| isForceUseTag(invocation)) {
- return invokers.intersect(result, invokers.getUnmodifiableList());
+ return new StateRouterResult<>(invokers.and(result));
} else {
invocation.setAttachment(TAG_KEY, NO_TAG);
- return invokers;
+ return new StateRouterResult<>(invokers);
}
}
}
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/tag/TagRouter.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/tag/TagRouter.java
index fad6636..08c0ef5 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/tag/TagRouter.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/tag/TagRouter.java
@@ -30,6 +30,7 @@ import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.cluster.router.AbstractRouter;
+import org.apache.dubbo.rpc.cluster.router.RouterResult;
import org.apache.dubbo.rpc.cluster.router.tag.model.TagRouterRule;
import org.apache.dubbo.rpc.cluster.router.tag.model.TagRuleParser;
@@ -62,7 +63,7 @@ public class TagRouter extends AbstractRouter implements ConfigurationListener {
public synchronized void process(ConfigChangedEvent event) {
if (logger.isDebugEnabled()) {
logger.debug("Notification of tag rule, change type is: " + event.getChangeType() + ", raw rule is:\n " +
- event.getContent());
+ event.getContent());
}
try {
@@ -73,25 +74,27 @@ public class TagRouter extends AbstractRouter implements ConfigurationListener {
}
} catch (Exception e) {
logger.error("Failed to parse the raw tag router rule and it will not take effect, please check if the " +
- "rule matches with the template, the raw rule is:\n ", e);
+ "rule matches with the template, the raw rule is:\n ", e);
}
}
@Override
- public <T> List<Invoker<T>> route(List<Invoker<T>> invokers, URL url, Invocation invocation) throws RpcException {
+ public <T> RouterResult<Invoker<T>> route(List<Invoker<T>> invokers, URL url,
+ Invocation invocation, boolean needToPrintMessage) throws RpcException {
+
if (CollectionUtils.isEmpty(invokers)) {
- return invokers;
+ return new RouterResult<>(invokers);
}
// since the rule can be changed by config center, we should copy one to use.
final TagRouterRule tagRouterRuleCopy = tagRouterRule;
if (tagRouterRuleCopy == null || !tagRouterRuleCopy.isValid() || !tagRouterRuleCopy.isEnabled()) {
- return filterUsingStaticTag(invokers, url, invocation);
+ return new RouterResult<>(filterUsingStaticTag(invokers, url, invocation));
}
List<Invoker<T>> result = invokers;
String tag = StringUtils.isEmpty(invocation.getAttachment(TAG_KEY)) ? url.getParameter(TAG_KEY) :
- invocation.getAttachment(TAG_KEY);
+ invocation.getAttachment(TAG_KEY);
// if we are requesting for a Provider with a specific tag
if (StringUtils.isNotEmpty(tag)) {
@@ -101,7 +104,7 @@ public class TagRouter extends AbstractRouter implements ConfigurationListener {
result = filterInvoker(invokers, invoker -> addressMatches(invoker.getUrl(), addresses));
// if result is not null OR it's null but force=true, return result directly
if (CollectionUtils.isNotEmpty(result) || tagRouterRuleCopy.isForce()) {
- return result;
+ return new RouterResult<>(result);
}
} else {
// dynamic tag group doesn't have any item about the requested app OR it's null after filtered by
@@ -111,13 +114,13 @@ public class TagRouter extends AbstractRouter implements ConfigurationListener {
// If there's no tagged providers that can match the current tagged request. force.tag is set by default
// to false, which means it will invoke any providers without a tag unless it's explicitly disallowed.
if (CollectionUtils.isNotEmpty(result) || isForceUseTag(invocation)) {
- return result;
+ return new RouterResult<>(result);
}
// FAILOVER: return all Providers without any tags.
else {
List<Invoker<T>> tmp = filterInvoker(invokers, invoker -> addressNotMatches(invoker.getUrl(),
- tagRouterRuleCopy.getAddresses()));
- return filterInvoker(tmp, invoker -> StringUtils.isEmpty(invoker.getUrl().getParameter(TAG_KEY)));
+ tagRouterRuleCopy.getAddresses()));
+ return new RouterResult<>(filterInvoker(tmp, invoker -> StringUtils.isEmpty(invoker.getUrl().getParameter(TAG_KEY))));
}
} else {
// List<String> addresses = tagRouterRule.filter(providerApp);
@@ -127,15 +130,15 @@ public class TagRouter extends AbstractRouter implements ConfigurationListener {
result = filterInvoker(invokers, invoker -> addressNotMatches(invoker.getUrl(), addresses));
// 1. all addresses are in dynamic tag group, return empty list.
if (CollectionUtils.isEmpty(result)) {
- return result;
+ return new RouterResult<>(result);
}
// 2. if there are some addresses that are not in any dynamic tag group, continue to filter using the
// static tag group.
}
- return filterInvoker(result, invoker -> {
+ return new RouterResult<>(filterInvoker(result, invoker -> {
String localTag = invoker.getUrl().getParameter(TAG_KEY);
return StringUtils.isEmpty(localTag) || !tagRouterRuleCopy.getTagNames().contains(localTag);
- });
+ }));
}
}
@@ -157,7 +160,7 @@ public class TagRouter extends AbstractRouter implements ConfigurationListener {
List<Invoker<T>> result;
// Dynamic param
String tag = StringUtils.isEmpty(invocation.getAttachment(TAG_KEY)) ? url.getParameter(TAG_KEY) :
- invocation.getAttachment(TAG_KEY);
+ invocation.getAttachment(TAG_KEY);
// Tag request
if (!StringUtils.isEmpty(tag)) {
result = filterInvoker(invokers, invoker -> tag.equals(invoker.getUrl().getParameter(TAG_KEY)));
@@ -191,8 +194,8 @@ public class TagRouter extends AbstractRouter implements ConfigurationListener {
}
return invokers.stream()
- .filter(predicate)
- .collect(Collectors.toList());
+ .filter(predicate)
+ .collect(Collectors.toList());
}
private boolean addressMatches(URL url, List<String> addresses) {
@@ -235,7 +238,7 @@ public class TagRouter extends AbstractRouter implements ConfigurationListener {
if (StringUtils.isEmpty(providerApplication)) {
logger.error("TagRouter must getConfig from or subscribe to a specific application, but the application " +
- "in this TagRouter is not specified.");
+ "in this TagRouter is not specified.");
return;
}
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/tag/TagStaticStateRouter.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/tag/TagStaticStateRouter.java
index 75e6d70..5e0998c 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/tag/TagStaticStateRouter.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/tag/TagStaticStateRouter.java
@@ -26,6 +26,7 @@ import org.apache.dubbo.rpc.cluster.RouterChain;
import org.apache.dubbo.rpc.cluster.router.state.AbstractStateRouter;
import org.apache.dubbo.rpc.cluster.router.state.BitList;
import org.apache.dubbo.rpc.cluster.router.state.RouterCache;
+import org.apache.dubbo.rpc.cluster.router.state.StateRouterResult;
import java.util.ArrayList;
import java.util.List;
@@ -52,9 +53,10 @@ public class TagStaticStateRouter extends AbstractStateRouter {
return url;
}
+
@Override
- public <T> BitList<Invoker<T>> route(BitList<Invoker<T>> invokers, RouterCache<T> routerCache, URL url, Invocation invocation)
- throws RpcException {
+ public <T> StateRouterResult<Invoker<T>> route(BitList<Invoker<T>> invokers, RouterCache<T> routerCache, URL url,
+ Invocation invocation, boolean needToPrintMessage) throws RpcException {
String tag = isNoTag(invocation.getAttachment(TAG_KEY)) ? url.getParameter(TAG_KEY) :
invocation.getAttachment(TAG_KEY);
@@ -65,9 +67,9 @@ public class TagStaticStateRouter extends AbstractStateRouter {
ConcurrentMap<String, BitList<Invoker<T>>> pool = routerCache.getAddrPool();
BitList<Invoker<T>> res = pool.get(tag);
if (res == null) {
- return invokers;
+ return new StateRouterResult<>(invokers);
}
- return invokers.intersect(res, invokers.getUnmodifiableList());
+ return new StateRouterResult<>(invokers.and(res));
}
private boolean isNoTag(String tag) {
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/AbstractClusterInvoker.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/AbstractClusterInvoker.java
index 0c0d404..42ad94d 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/AbstractClusterInvoker.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/AbstractClusterInvoker.java
@@ -18,6 +18,8 @@ package org.apache.dubbo.rpc.cluster.support;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.Version;
+import org.apache.dubbo.common.config.Configuration;
+import org.apache.dubbo.common.config.ConfigurationUtils;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.CollectionUtils;
@@ -37,10 +39,14 @@ import org.apache.dubbo.rpc.support.RpcUtils;
import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicBoolean;
import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_LOADBALANCE;
+import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_RESELECT_COUNT;
+import static org.apache.dubbo.common.constants.CommonConstants.ENABLE_CONNECTIVITY_VALIDATION;
import static org.apache.dubbo.common.constants.CommonConstants.LOADBALANCE_KEY;
+import static org.apache.dubbo.common.constants.CommonConstants.RESELECT_COUNT;
import static org.apache.dubbo.rpc.cluster.Constants.CLUSTER_AVAILABLE_CHECK_KEY;
import static org.apache.dubbo.rpc.cluster.Constants.CLUSTER_STICKY_KEY;
import static org.apache.dubbo.rpc.cluster.Constants.DEFAULT_CLUSTER_AVAILABLE_CHECK;
@@ -55,7 +61,11 @@ public abstract class AbstractClusterInvoker<T> implements ClusterInvoker<T> {
protected Directory<T> directory;
- protected boolean availablecheck;
+ protected boolean availableCheck;
+
+ private volatile int reselectCount = DEFAULT_RESELECT_COUNT;
+
+ private volatile boolean enableConnectivityValidation = true;
private AtomicBoolean destroyed = new AtomicBoolean(false);
@@ -75,7 +85,10 @@ public abstract class AbstractClusterInvoker<T> implements ClusterInvoker<T> {
this.directory = directory;
//sticky: invoker.isAvailable() should always be checked before using when availablecheck is true.
- this.availablecheck = url.getParameter(CLUSTER_AVAILABLE_CHECK_KEY, DEFAULT_CLUSTER_AVAILABLE_CHECK);
+ this.availableCheck = url.getParameter(CLUSTER_AVAILABLE_CHECK_KEY, DEFAULT_CLUSTER_AVAILABLE_CHECK);
+ Configuration configuration = ConfigurationUtils.getGlobalConfiguration(url.getOrDefaultModuleModel());
+ this.reselectCount = configuration.getInt(RESELECT_COUNT, DEFAULT_RESELECT_COUNT);
+ this.enableConnectivityValidation = configuration.getBoolean(ENABLE_CONNECTIVITY_VALIDATION, true);
}
@Override
@@ -144,7 +157,7 @@ public abstract class AbstractClusterInvoker<T> implements ClusterInvoker<T> {
String methodName = invocation == null ? StringUtils.EMPTY_STRING : invocation.getMethodName();
boolean sticky = invokers.get(0).getUrl()
- .getMethodParameter(methodName, CLUSTER_STICKY_KEY, DEFAULT_CLUSTER_STICKY);
+ .getMethodParameter(methodName, CLUSTER_STICKY_KEY, DEFAULT_CLUSTER_STICKY);
//ignore overloaded method
if (stickyInvoker != null && !invokers.contains(stickyInvoker)) {
@@ -152,7 +165,7 @@ public abstract class AbstractClusterInvoker<T> implements ClusterInvoker<T> {
}
//ignore concurrency problem
if (sticky && stickyInvoker != null && (selected == null || !selected.contains(stickyInvoker))) {
- if (availablecheck && stickyInvoker.isAvailable()) {
+ if (availableCheck && stickyInvoker.isAvailable()) {
return stickyInvoker;
}
}
@@ -173,15 +186,22 @@ public abstract class AbstractClusterInvoker<T> implements ClusterInvoker<T> {
return null;
}
if (invokers.size() == 1) {
- return invokers.get(0);
+ Invoker<T> tInvoker = invokers.get(0);
+ checkShouldInvalidateInvoker(tInvoker);
+ return tInvoker;
}
Invoker<T> invoker = loadbalance.select(invokers, getUrl(), invocation);
//If the `invoker` is in the `selected` or invoker is unavailable && availablecheck is true, reselect.
- if ((selected != null && selected.contains(invoker))
- || (!invoker.isAvailable() && getUrl() != null && availablecheck)) {
+ boolean isSelected = selected != null && selected.contains(invoker);
+ boolean isUnavailable = availableCheck && !invoker.isAvailable() && getUrl() != null;
+
+ if (isUnavailable) {
+ invalidateInvoker(invoker);
+ }
+ if (isSelected || isUnavailable) {
try {
- Invoker<T> rInvoker = reselect(loadbalance, invocation, invokers, selected, availablecheck);
+ Invoker<T> rInvoker = reselect(loadbalance, invocation, invokers, selected, availableCheck);
if (rInvoker != null) {
invoker = rInvoker;
} else {
@@ -210,48 +230,92 @@ public abstract class AbstractClusterInvoker<T> implements ClusterInvoker<T> {
* @param invocation invocation
* @param invokers invoker candidates
* @param selected exclude selected invokers or not
- * @param availablecheck check invoker available if true
+ * @param availableCheck check invoker available if true
* @return the reselect result to do invoke
* @throws RpcException exception
*/
private Invoker<T> reselect(LoadBalance loadbalance, Invocation invocation,
- List<Invoker<T>> invokers, List<Invoker<T>> selected, boolean availablecheck) throws RpcException {
-
- //Allocating one in advance, this list is certain to be used.
- List<Invoker<T>> reselectInvokers = new ArrayList<>(
- invokers.size() > 1 ? (invokers.size() - 1) : invokers.size());
+ List<Invoker<T>> invokers, List<Invoker<T>> selected, boolean availableCheck) throws RpcException {
+
+ // Allocating one in advance, this list is certain to be used.
+ List<Invoker<T>> reselectInvokers = new ArrayList<>(Math.min(invokers.size(), reselectCount));
+
+ // 1. Try picking some invokers not in `selected`.
+ // 1.1. If all selectable invokers' size is smaller than reselectCount, just add all
+ // 1.2. If all selectable invokers' size is greater than reselectCount, randomly select reselectCount.
+ // The result size of invokers might smaller than reselectCount due to disAvailable or de-duplication (might be zero).
+ // This means there is probable that reselectInvokers is empty however all invoker list may contain available invokers.
+ // Use reselectCount can reduce retry times if invokers' size is huge, which may lead to long time hang up.
+ if (reselectCount >= invokers.size()) {
+ for (Invoker<T> invoker : invokers) {
+ // check if available
+ if (availableCheck && !invoker.isAvailable()) {
+ // add to invalidate invoker
+ invalidateInvoker(invoker);
+ continue;
+ }
- // First, try picking a invoker not in `selected`.
- for (Invoker<T> invoker : invokers) {
- if (availablecheck && !invoker.isAvailable()) {
- continue;
+ if (selected == null || !selected.contains(invoker)) {
+ reselectInvokers.add(invoker);
+ }
}
-
- if (selected == null || !selected.contains(invoker)) {
- reselectInvokers.add(invoker);
+ } else {
+ for (int i = 0; i < reselectCount; i++) {
+ // select one randomly
+ Invoker<T> invoker = invokers.get(ThreadLocalRandom.current().nextInt(invokers.size()));
+ // check if available
+ if (availableCheck && !invoker.isAvailable()) {
+ // add to invalidate invoker
+ invalidateInvoker(invoker);
+ continue;
+ }
+ // de-duplication
+ if (selected == null || !selected.contains(invoker) || !reselectInvokers.contains(invoker)) {
+ reselectInvokers.add(invoker);
+ }
}
}
+ // 2. Use loadBalance to select one (all the reselectInvokers are available)
if (!reselectInvokers.isEmpty()) {
return loadbalance.select(reselectInvokers, getUrl(), invocation);
}
- // Just pick an available invoker using loadbalance policy
+ // 3. reselectInvokers is empty. Unable to find at least one available invoker.
+ // Re-check all the selected invokers. If some in the selected list are available, add to reselectInvokers.
if (selected != null) {
for (Invoker<T> invoker : selected) {
if ((invoker.isAvailable()) // available first
- && !reselectInvokers.contains(invoker)) {
+ && !reselectInvokers.contains(invoker)) {
reselectInvokers.add(invoker);
}
}
}
+
+ // 4. If reselectInvokers is not empty after re-check.
+ // Pick an available invoker using loadBalance policy
if (!reselectInvokers.isEmpty()) {
return loadbalance.select(reselectInvokers, getUrl(), invocation);
}
+ // 5. No invoker match, return null.
return null;
}
+ private void checkShouldInvalidateInvoker(Invoker<T> invoker) {
+ if (availableCheck && !invoker.isAvailable()) {
+ invalidateInvoker(invoker);
+ }
+ }
+
+ private void invalidateInvoker(Invoker<T> invoker) {
+ if (enableConnectivityValidation) {
+ if (getDirectory() != null) {
+ getDirectory().addInvalidateInvoker(invoker);
+ }
+ }
+ }
+
@Override
public Result invoke(final Invocation invocation) throws RpcException {
checkWhetherDestroyed();
@@ -271,8 +335,8 @@ public abstract class AbstractClusterInvoker<T> implements ClusterInvoker<T> {
protected void checkWhetherDestroyed() {
if (destroyed.get()) {
throw new RpcException("Rpc cluster invoker for " + getInterface() + " on consumer " + NetUtils.getLocalHost()
- + " use dubbo version " + Version.getVersion()
- + " is now destroyed! Can not invoke any more.");
+ + " use dubbo version " + Version.getVersion()
+ + " is now destroyed! Can not invoke any more.");
}
}
@@ -284,12 +348,12 @@ public abstract class AbstractClusterInvoker<T> implements ClusterInvoker<T> {
protected void checkInvokers(List<Invoker<T>> invokers, Invocation invocation) {
if (CollectionUtils.isEmpty(invokers)) {
throw new RpcException(RpcException.NO_INVOKER_AVAILABLE_AFTER_FILTER, "Failed to invoke the method "
- + invocation.getMethodName() + " in the service " + getInterface().getName()
- + ". No provider available for the service " + getDirectory().getConsumerUrl().getServiceKey()
- + " from registry " + getDirectory().getUrl().getAddress()
- + " on the consumer " + NetUtils.getLocalHost()
- + " using the dubbo version " + Version.getVersion()
- + ". Please check if the providers have been started and registered.");
+ + invocation.getMethodName() + " in the service " + getInterface().getName()
+ + ". No provider available for the service " + getDirectory().getConsumerUrl().getServiceKey()
+ + " from registry " + getDirectory().getUrl().getAddress()
+ + " on the consumer " + NetUtils.getLocalHost()
+ + " using the dubbo version " + Version.getVersion()
+ + ". Please check if the providers have been started and registered.");
}
}
@@ -307,17 +371,18 @@ public abstract class AbstractClusterInvoker<T> implements ClusterInvoker<T> {
/**
* When using a thread pool to fork a child thread, ThreadLocal cannot be passed.
* In this scenario, please use the invokeWithContextAsync method.
+ *
* @return
*/
protected Result invokeWithContextAsync(Invoker<T> invoker, Invocation invocation, URL consumerUrl) {
- setContext(invoker, consumerUrl);
+ setContext(invoker, consumerUrl);
Result result;
try {
result = invoker.invoke(invocation);
} finally {
clearContext(invoker);
}
- return result;
+ return result;
}
protected abstract Result doInvoke(Invocation invocation, List<Invoker<T>> invokers,
@@ -342,9 +407,9 @@ public abstract class AbstractClusterInvoker<T> implements ClusterInvoker<T> {
ApplicationModel applicationModel = ScopeModelUtil.getApplicationModel(invocation.getModuleModel());
if (CollectionUtils.isNotEmpty(invokers)) {
return applicationModel.getExtensionLoader(LoadBalance.class).getExtension(
- invokers.get(0).getUrl().getMethodParameter(
- RpcUtils.getMethodName(invocation), LOADBALANCE_KEY, DEFAULT_LOADBALANCE
- )
+ invokers.get(0).getUrl().getMethodParameter(
+ RpcUtils.getMethodName(invocation), LOADBALANCE_KEY, DEFAULT_LOADBALANCE
+ )
);
} else {
return applicationModel.getExtensionLoader(LoadBalance.class).getExtension(DEFAULT_LOADBALANCE);
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/wrapper/MockClusterInvoker.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/wrapper/MockClusterInvoker.java
index a5b5200..9e93db2 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/wrapper/MockClusterInvoker.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/wrapper/MockClusterInvoker.java
@@ -25,6 +25,7 @@ import org.apache.dubbo.rpc.AsyncRpcResult;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.Result;
+import org.apache.dubbo.rpc.RpcContext;
import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.RpcInvocation;
import org.apache.dubbo.rpc.cluster.ClusterInvoker;
@@ -177,6 +178,7 @@ public class MockClusterInvoker<T> implements ClusterInvoker<T> {
invocation.setAttachment(INVOCATION_NEED_MOCK, Boolean.TRUE.toString());
//directory will return a list of normal invokers if Constants.INVOCATION_NEED_MOCK is absent or not true in invocation, otherwise, a list of mock invokers will return.
try {
+ RpcContext.getServiceContext().setConsumerUrl(getUrl());
invokers = directory.list(invocation);
} catch (RpcException e) {
if (logger.isInfoEnabled()) {
diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/directory/StaticDirectoryTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/directory/StaticDirectoryTest.java
index 7ce8f4a..c9fe0c7 100644
--- a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/directory/StaticDirectoryTest.java
+++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/directory/StaticDirectoryTest.java
@@ -54,13 +54,14 @@ public class StaticDirectoryTest {
invokers.add(invoker1);
invokers.add(invoker2);
invokers.add(invoker3);
- List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + NetUtils.getLocalHost() + "/com.foo.BarService"), new RpcInvocation());
+ List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + NetUtils.getLocalHost() + "/com.foo.BarService"), new RpcInvocation(), false).getResult();
StaticDirectory<String> staticDirectory = new StaticDirectory<>(filteredInvokers);
boolean isAvailable = staticDirectory.isAvailable();
Assertions.assertTrue(!isAvailable);
List<Invoker<String>> newInvokers = staticDirectory.list(new MockDirInvocation());
Assertions.assertTrue(newInvokers.size() > 0);
staticDirectory.destroy();
- Assertions.assertEquals(0, newInvokers.size());
+ Assertions.assertEquals(0, staticDirectory.invokers.size());
+ Assertions.assertEquals(0, staticDirectory.validInvokers.size());
}
}
diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/loadbalance/ConsistentHashLoadBalanceTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/loadbalance/ConsistentHashLoadBalanceTest.java
index cea2856..ea1d782 100644
--- a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/loadbalance/ConsistentHashLoadBalanceTest.java
+++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/loadbalance/ConsistentHashLoadBalanceTest.java
@@ -19,6 +19,8 @@ package org.apache.dubbo.rpc.cluster.loadbalance;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.cluster.RouterChain;
+import org.apache.dubbo.rpc.cluster.router.state.BitList;
+
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -54,8 +56,8 @@ public class ConsistentHashLoadBalanceTest extends LoadBalanceBaseTest {
int originalHashCode = lb.getCorrespondingHashCode(invokers);
for (int i = 0; i < 100; i++) {
- routerChain.setInvokers(invokers);
- List<Invoker<LoadBalanceBaseTest>> routeInvokers = routerChain.route(url, invocation);
+ routerChain.setInvokers(new BitList<>(invokers));
+ List<Invoker<LoadBalanceBaseTest>> routeInvokers = routerChain.route(url, new BitList<>(invokers), invocation);
Assertions.assertEquals(originalHashCode, lb.getCorrespondingHashCode(routeInvokers));
}
diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/condition/ConditionRouterTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/condition/ConditionRouterTest.java
index ace3cde..5d84f60 100644
--- a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/condition/ConditionRouterTest.java
+++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/condition/ConditionRouterTest.java
@@ -121,12 +121,12 @@ public class ConditionRouterTest {
- List<Invoker<String>> filteredInvokers1 = router1.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation());
- List<Invoker<String>> filteredInvokers2 = router2.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation());
- List<Invoker<String>> filteredInvokers3 = router3.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation());
- List<Invoker<String>> filteredInvokers4 = router4.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation());
- List<Invoker<String>> filteredInvokers5 = router5.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation());
- List<Invoker<String>> filteredInvokers6 = router6.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation());
+ List<Invoker<String>> filteredInvokers1 = router1.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false).getResult();
+ List<Invoker<String>> filteredInvokers2 = router2.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false).getResult();
+ List<Invoker<String>> filteredInvokers3 = router3.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false).getResult();
+ List<Invoker<String>> filteredInvokers4 = router4.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false).getResult();
+ List<Invoker<String>> filteredInvokers5 = router5.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false).getResult();
+ List<Invoker<String>> filteredInvokers6 = router6.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false).getResult();
Assertions.assertEquals(1, filteredInvokers1.size());
Assertions.assertEquals(0, filteredInvokers2.size());
Assertions.assertEquals(0, filteredInvokers3.size());
@@ -174,14 +174,14 @@ public class ConditionRouterTest {
"host = " + LOCAL_HOST + " & methods = getFoo => " + " host = 10.20.3.3").addParameter(
FORCE_KEY, String.valueOf(true)));
List<Invoker<String>> filteredInvokers1 = router4.route(invokers,
- URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), invocation);
+ URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), invocation, false).getResult();
Assertions.assertEquals(1, filteredInvokers1.size());
Router router5 = new ConditionRouterFactory().getRouter(getRouteUrl(
"host = " + LOCAL_HOST + " & methods = unvalidmethod => " + " host = 10.20.3.3")
.addParameter(FORCE_KEY, String.valueOf(true)));
List<Invoker<String>> filteredInvokers2 = router5.route(invokers,
- URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), invocation);
+ URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), invocation, false).getResult();
Assertions.assertEquals(3, filteredInvokers2.size());
// Request a non-exists method
}
@@ -193,7 +193,7 @@ public class ConditionRouterTest {
invokers.add(new MockInvoker<String>());
invokers.add(new MockInvoker<String>());
invokers.add(new MockInvoker<String>());
- List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation());
+ List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false).getResult();
Assertions.assertEquals(0, filteredInvokers.size());
}
@@ -204,7 +204,7 @@ public class ConditionRouterTest {
invokers.add(new MockInvoker<String>());
invokers.add(new MockInvoker<String>());
invokers.add(new MockInvoker<String>());
- List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation());
+ List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false).getResult();
Assertions.assertEquals(0, filteredInvokers.size());
}
@@ -215,7 +215,7 @@ public class ConditionRouterTest {
invokers.add(new MockInvoker<String>(URL.valueOf("dubbo://" + LOCAL_HOST + ":20880/com.foo.BarService")));
invokers.add(new MockInvoker<String>(URL.valueOf("dubbo://" + LOCAL_HOST + ":20880/com.foo.BarService")));
invokers.add(new MockInvoker<String>(URL.valueOf("dubbo://" + LOCAL_HOST + ":20880/com.foo.BarService")));
- List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation());
+ List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false).getResult();
Assertions.assertEquals(invokers, filteredInvokers);
}
@@ -229,7 +229,7 @@ public class ConditionRouterTest {
invokers.add(invoker1);
invokers.add(invoker2);
invokers.add(invoker3);
- List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation());
+ List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false).getResult();
Assertions.assertEquals(2, filteredInvokers.size());
Assertions.assertEquals(invoker2, filteredInvokers.get(0));
Assertions.assertEquals(invoker3, filteredInvokers.get(1));
@@ -245,7 +245,7 @@ public class ConditionRouterTest {
invokers.add(invoker1);
invokers.add(invoker2);
invokers.add(invoker3);
- List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation());
+ List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false).getResult();
Assertions.assertEquals(2, filteredInvokers.size());
Assertions.assertEquals(invoker2, filteredInvokers.get(0));
Assertions.assertEquals(invoker3, filteredInvokers.get(1));
@@ -261,7 +261,7 @@ public class ConditionRouterTest {
invokers.add(invoker1);
invokers.add(invoker2);
invokers.add(invoker3);
- List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation());
+ List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false).getResult();
Assertions.assertEquals(2, filteredInvokers.size());
Assertions.assertEquals(invoker2, filteredInvokers.get(0));
Assertions.assertEquals(invoker3, filteredInvokers.get(1));
@@ -277,7 +277,7 @@ public class ConditionRouterTest {
invokers.add(invoker1);
invokers.add(invoker2);
invokers.add(invoker3);
- List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation());
+ List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false).getResult();
Assertions.assertEquals(2, filteredInvokers.size());
Assertions.assertEquals(invoker2, filteredInvokers.get(0));
Assertions.assertEquals(invoker3, filteredInvokers.get(1));
@@ -293,7 +293,7 @@ public class ConditionRouterTest {
invokers.add(invoker1);
invokers.add(invoker2);
invokers.add(invoker3);
- List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation());
+ List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false).getResult();
Assertions.assertEquals(invokers, filteredInvokers);
}
@@ -307,7 +307,7 @@ public class ConditionRouterTest {
invokers.add(invoker1);
invokers.add(invoker2);
invokers.add(invoker3);
- List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation());
+ List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false).getResult();
Assertions.assertEquals(0, filteredInvokers.size());
}
@@ -324,25 +324,25 @@ public class ConditionRouterTest {
RpcInvocation invocation = new RpcInvocation();
String p = "a";
invocation.setArguments(new Object[]{null});
- List<Invoker<String>> fileredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), invocation);
+ List<Invoker<String>> fileredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), invocation, false).getResult();
Assertions.assertEquals(3, fileredInvokers.size());
invocation.setArguments(new Object[]{p});
- fileredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), invocation);
+ fileredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), invocation, false).getResult();
Assertions.assertEquals(0, fileredInvokers.size());
router = new ConditionRouterFactory().getRouter(getRouteUrl("arguments = b " + " => " + " host = 1.2.3.4").addParameter(FORCE_KEY, String.valueOf(true)));
- fileredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), invocation);
+ fileredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), invocation, false).getResult();
Assertions.assertEquals(3, fileredInvokers.size());
router = new ConditionRouterFactory().getRouter(getRouteUrl("arguments[10].inner = a " + " => " + " host = 1.2.3.4").addParameter(FORCE_KEY, String.valueOf(true)));
- fileredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), invocation);
+ fileredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), invocation, false).getResult();
Assertions.assertEquals(3, fileredInvokers.size());
int integer = 1;
invocation.setArguments(new Object[]{integer});
router = new ConditionRouterFactory().getRouter(getRouteUrl("arguments[0].inner = 1 " + " => " + " host = 1.2.3.4").addParameter(FORCE_KEY, String.valueOf(true)));
- fileredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), invocation);
+ fileredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), invocation, false).getResult();
Assertions.assertEquals(0, fileredInvokers.size());
}
}
diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/file/FileRouterEngineTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/file/FileRouterEngineTest.java
index 6a03001..9eff618 100644
--- a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/file/FileRouterEngineTest.java
+++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/file/FileRouterEngineTest.java
@@ -22,6 +22,7 @@ import org.apache.dubbo.rpc.AppResponse;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.Result;
+import org.apache.dubbo.rpc.RpcContext;
import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.RpcInvocation;
import org.apache.dubbo.rpc.cluster.Directory;
@@ -30,6 +31,7 @@ import org.apache.dubbo.rpc.cluster.RouterFactory;
import org.apache.dubbo.rpc.cluster.directory.StaticDirectory;
import org.apache.dubbo.rpc.cluster.support.AbstractClusterInvoker;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
@@ -40,6 +42,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
+import static org.apache.dubbo.common.constants.CommonConstants.ENABLE_CONNECTIVITY_VALIDATION;
import static org.apache.dubbo.rpc.cluster.Constants.RUNTIME_KEY;
import static org.mockito.BDDMockito.given;
import static org.mockito.Mockito.mock;
@@ -57,6 +60,7 @@ public class FileRouterEngineTest {
@BeforeAll
public static void setUpBeforeClass() throws Exception {
+ System.setProperty(ENABLE_CONNECTIVITY_VALIDATION, "false");
}
@BeforeEach
@@ -65,6 +69,12 @@ public class FileRouterEngineTest {
invokers.add(invoker2);
}
+ @AfterEach
+ public void teardown() throws Exception {
+ System.clearProperty(ENABLE_CONNECTIVITY_VALIDATION);
+ RpcContext.removeContext();
+ }
+
@Test
public void testRouteNotAvailable() {
if (isScriptUnsupported) return;
diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/mesh/route/MeshRuleRouterTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/mesh/route/MeshRuleRouterTest.java
index 1742295..c62c47e 100644
--- a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/mesh/route/MeshRuleRouterTest.java
+++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/mesh/route/MeshRuleRouterTest.java
@@ -33,6 +33,7 @@ import org.apache.dubbo.rpc.cluster.router.mesh.rule.virtualservice.destination.
import org.apache.dubbo.rpc.cluster.router.mesh.rule.virtualservice.destination.DubboRouteDestination;
import org.apache.dubbo.rpc.cluster.router.mesh.rule.virtualservice.match.DubboMethodMatch;
import org.apache.dubbo.rpc.cluster.router.mesh.rule.virtualservice.match.StringMatch;
+
import org.junit.jupiter.api.Test;
import java.util.ArrayList;
@@ -1069,7 +1070,7 @@ public class MeshRuleRouterTest {
virtualServiceRuleList.add(virtualServiceRule);
vsDestinationGroup.setVirtualServiceRuleList(virtualServiceRuleList);
meshRuleRouter.computeSubset();
- assertEquals(inputInvokers, meshRuleRouter.route((List) inputInvokers, inputURL, invocation));
+ assertEquals(inputInvokers, meshRuleRouter.route((List) inputInvokers, inputURL, invocation, false).getResult());
}
@@ -1128,13 +1129,13 @@ public class MeshRuleRouterTest {
virtualServiceRuleList.add(virtualServiceRule);
vsDestinationGroup.setVirtualServiceRuleList(virtualServiceRuleList);
meshRuleRouter.computeSubset();
- assertNotEquals(inputInvokers, meshRuleRouter.route((List) inputInvokers, inputURL, invocation));
- assertEquals(1, meshRuleRouter.route((List) inputInvokers, inputURL, invocation).size());
+ assertNotEquals(inputInvokers, meshRuleRouter.route((List) inputInvokers, inputURL, invocation, false).getResult());
+ assertEquals(1, meshRuleRouter.route((List) inputInvokers, inputURL, invocation, false).getResult().size());
Map<String, String> invokerParameterMap = new HashMap<>();
invokerParameterMap.put("env", "test1");
- assertEquals(invokerParameterMap, ((Invoker) meshRuleRouter.route((List) inputInvokers, inputURL, invocation).get(0)).getUrl().getServiceParameters(url.getProtocolServiceKey()));
+ assertEquals(invokerParameterMap, ((Invoker) meshRuleRouter.route((List) inputInvokers, inputURL, invocation, false).getResult().get(0)).getUrl().getServiceParameters(url.getProtocolServiceKey()));
}
}
@@ -1322,13 +1323,13 @@ public class MeshRuleRouterTest {
virtualServiceRuleList.add(virtualServiceRule);
vsDestinationGroup.setVirtualServiceRuleList(virtualServiceRuleList);
meshRuleRouter.computeSubset();
- assertNotEquals(inputInvokers, meshRuleRouter.route((List) inputInvokers, inputURL, invocation));
- assertEquals(1, meshRuleRouter.route((List) inputInvokers, inputURL, invocation).size());
+ assertNotEquals(inputInvokers, meshRuleRouter.route((List) inputInvokers, inputURL, invocation, false).getResult());
+ assertEquals(1, meshRuleRouter.route((List) inputInvokers, inputURL, invocation, false).getResult().size());
Map<String, String> invokerParameterMap = new HashMap<>();
invokerParameterMap.put("env", "test1");
- assertEquals(invokerParameterMap, ((Invoker) meshRuleRouter.route((List) inputInvokers, inputURL, invocation).get(0)).getUrl().getServiceParameters(url.getProtocolServiceKey()));
+ assertEquals(invokerParameterMap, ((Invoker) meshRuleRouter.route((List) inputInvokers, inputURL, invocation, false).getResult().get(0)).getUrl().getServiceParameters(url.getProtocolServiceKey()));
}
{
@@ -1398,10 +1399,10 @@ public class MeshRuleRouterTest {
vsDestinationGroup.setVirtualServiceRuleList(virtualServiceRuleList);
meshRuleRouter.computeSubset();
- assertNull(meshRuleRouter.route((List) inputInvokers, inputURL, invocation));
+ assertNull(meshRuleRouter.route((List) inputInvokers, inputURL, invocation, false).getResult());
meshRuleRouter.setSubsetMap(null);
- assertNotNull(meshRuleRouter.route((List) inputInvokers, inputURL, invocation));
+ assertNotNull(meshRuleRouter.route((List) inputInvokers, inputURL, invocation, false).getResult());
}
}
}
diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/mock/MockInvokersSelectorTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/mock/MockInvokersSelectorTest.java
index 43dc03a..3cd8e69 100644
--- a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/mock/MockInvokersSelectorTest.java
+++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/mock/MockInvokersSelectorTest.java
@@ -19,6 +19,7 @@ package org.apache.dubbo.rpc.cluster.router.mock;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.RpcInvocation;
+
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
@@ -50,7 +51,7 @@ public class MockInvokersSelectorTest {
URL consumerURL = URL.valueOf("test://127.0.0.1");
// rpcInvocation does not have an attached "invocation.need.mock" parameter, so normal invokers will be filtered out
- List<Invoker<DemoService>> invokers = selector.route(providers, consumerURL, rpcInvocation);
+ List<Invoker<DemoService>> invokers = selector.route(providers, consumerURL, rpcInvocation, false).getResult();
Assertions.assertEquals(invokers.size(),1);
Assertions.assertTrue(invokers.contains(invoker3));
@@ -58,7 +59,7 @@ public class MockInvokersSelectorTest {
Map<String,Object> attachments = new HashMap<>();
attachments.put(INVOCATION_NEED_MOCK,"true");
Mockito.when(rpcInvocation.getObjectAttachments()).thenReturn(attachments);
- invokers = selector.route(providers, consumerURL, rpcInvocation);
+ invokers = selector.route(providers, consumerURL, rpcInvocation, false).getResult();
Assertions.assertEquals(invokers.size(),2);
Assertions.assertTrue(invokers.contains(invoker1));
Assertions.assertTrue(invokers.contains(invoker2));
diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/script/ScriptRouterTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/script/ScriptRouterTest.java
index fa86c00..91819e8 100644
--- a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/script/ScriptRouterTest.java
+++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/script/ScriptRouterTest.java
@@ -56,7 +56,7 @@ public class ScriptRouterTest {
invokers.add(new MockInvoker<String>());
invokers.add(new MockInvoker<String>());
invokers.add(new MockInvoker<String>());
- List<Invoker<String>> filteredInvokers = router.route(invokers, invokers.get(0).getUrl(), new RpcInvocation());
+ List<Invoker<String>> filteredInvokers = router.route(invokers, invokers.get(0).getUrl(), new RpcInvocation(), false).getResult();
Assertions.assertEquals(invokers, filteredInvokers);
}
@@ -79,7 +79,7 @@ public class ScriptRouterTest {
invokers.add(invoker1);
invokers.add(invoker2);
invokers.add(invoker3);
- List<Invoker<String>> filteredInvokers = router.route(invokers, invokers.get(0).getUrl(), new RpcInvocation());
+ List<Invoker<String>> filteredInvokers = router.route(invokers, invokers.get(0).getUrl(), new RpcInvocation(), false).getResult();
Assertions.assertEquals(2, filteredInvokers.size());
Assertions.assertEquals(invoker2, filteredInvokers.get(0));
Assertions.assertEquals(invoker3, filteredInvokers.get(1));
@@ -109,7 +109,7 @@ public class ScriptRouterTest {
"route(invokers, invocation, context) ";
Router router = new ScriptRouterFactory().getRouter(getRouteUrl(script));
- List<Invoker<String>> routeResult = router.route(invokers, invokers.get(0).getUrl(), new RpcInvocation());
+ List<Invoker<String>> routeResult = router.route(invokers, invokers.get(0).getUrl(), new RpcInvocation(), false).getResult();
Assertions.assertEquals(1, routeResult.size());
Assertions.assertEquals(invoker2, routeResult.get(0));
}
@@ -126,7 +126,7 @@ public class ScriptRouterTest {
String script = "/";
Router router = new ScriptRouterFactory().getRouter(getRouteUrl(script));
- List<Invoker<String>> routeResult = router.route(invokers, invokers.get(0).getUrl(), new RpcInvocation());
+ List<Invoker<String>> routeResult = router.route(invokers, invokers.get(0).getUrl(), new RpcInvocation(), false).getResult();
Assertions.assertEquals(3, routeResult.size());
}
}
diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/state/BitListTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/state/BitListTest.java
index 3761f72..76c98ae 100644
--- a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/state/BitListTest.java
+++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/state/BitListTest.java
@@ -20,15 +20,18 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import java.util.Arrays;
-import java.util.Iterator;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.LinkedList;
import java.util.List;
+import java.util.ListIterator;
public class BitListTest {
@Test
public void test() {
List<String> list = Arrays.asList("A", "B", "C");
BitList<String> bitList = new BitList<>(list);
- Assertions.assertEquals(bitList.getUnmodifiableList(), list);
+ Assertions.assertEquals(bitList.getOriginList(), list);
Assertions.assertEquals(3, bitList.size());
Assertions.assertEquals("A", bitList.get(0));
@@ -39,9 +42,7 @@ public class BitListTest {
Assertions.assertTrue(bitList.contains("B"));
Assertions.assertTrue(bitList.contains("C"));
- Iterator<String> iterator = bitList.iterator();
- while (iterator.hasNext()) {
- String str = iterator.next();
+ for (String str : bitList) {
Assertions.assertTrue(list.contains(str));
}
@@ -61,7 +62,10 @@ public class BitListTest {
newObjects = new Object[10];
copiedList = bitList.toArray(newObjects);
- Assertions.assertNull(copiedList);
+ Assertions.assertEquals(copiedList.length, 10);
+ Assertions.assertEquals("A", copiedList[0]);
+ Assertions.assertEquals("B", copiedList[1]);
+ Assertions.assertEquals("C", copiedList[2]);
bitList.remove(0);
Assertions.assertEquals("B", bitList.get(0));
@@ -82,10 +86,471 @@ public class BitListTest {
BitList<String> aBitList = new BitList<>(aList);
BitList<String> bBitList = new BitList<>(bList);
- BitList<String> intersectBitList = aBitList.intersect(bBitList, totalList);
+ BitList<String> intersectBitList = aBitList.and(bBitList);
Assertions.assertEquals(intersectBitList.size(), 2);
Assertions.assertEquals(intersectBitList.get(0), totalList.get(0));
Assertions.assertEquals(intersectBitList.get(1), totalList.get(1));
+
+ aBitList.add("D");
+ intersectBitList = aBitList.and(bBitList);
+ Assertions.assertEquals(intersectBitList.size(), 3);
+ Assertions.assertEquals(intersectBitList.get(0), totalList.get(0));
+ Assertions.assertEquals(intersectBitList.get(1), totalList.get(1));
+ Assertions.assertEquals("D", intersectBitList.get(2));
+ }
+
+ @Test
+ public void testIsEmpty() {
+ List<String> list = Arrays.asList("A", "B", "C");
+ BitList<String> bitList = new BitList<>(list);
+ bitList.add("D");
+
+ bitList.removeAll(list);
+ Assertions.assertEquals(1, bitList.size());
+
+ bitList.remove("D");
+ Assertions.assertTrue(bitList.isEmpty());
+ }
+
+ @Test
+ public void testAdd() {
+ List<String> list = Arrays.asList("A", "B", "C");
+ BitList<String> bitList = new BitList<>(list);
+
+ bitList.remove("A");
+
+ Assertions.assertEquals(2, bitList.size());
+
+ bitList.addAll(Collections.singletonList("A"));
+ Assertions.assertEquals(3, bitList.size());
+
+ bitList.addAll(Collections.singletonList("D"));
+ Assertions.assertEquals(4, bitList.size());
+ Assertions.assertEquals("D", bitList.get(3));
+ Assertions.assertTrue(bitList.hasMoreElementInTailList());
+ Assertions.assertEquals(Collections.singletonList("D"), bitList.getTailList());
+
+ bitList.clear();
+
+ bitList.addAll(Collections.singletonList("A"));
+ Assertions.assertEquals(1, bitList.size());
+ Assertions.assertEquals("A", bitList.get(0));
+ }
+
+ @Test
+ public void testGet() {
+ List<String> list = Arrays.asList("A", "B", "C");
+ BitList<String> bitList = new BitList<>(list);
+
+ Assertions.assertEquals("A", bitList.get(0));
+ Assertions.assertEquals("B", bitList.get(1));
+ Assertions.assertEquals("C", bitList.get(2));
+
+ Assertions.assertThrows(IndexOutOfBoundsException.class, () -> bitList.get(-1));
+ Assertions.assertThrows(IndexOutOfBoundsException.class, () -> bitList.get(3));
+
+ bitList.add("D");
+ Assertions.assertEquals("D", bitList.get(3));
+
+ Assertions.assertThrows(IndexOutOfBoundsException.class, () -> bitList.get(-1));
+ Assertions.assertThrows(IndexOutOfBoundsException.class, () -> bitList.get(4));
+ Assertions.assertThrows(IndexOutOfBoundsException.class, () -> bitList.get(5));
+ }
+
+ @Test
+ public void testRemove() {
+ List<String> list = Arrays.asList("A", "B", "C");
+ BitList<String> bitList = new BitList<>(list);
+
+ Assertions.assertTrue(bitList.remove("A"));
+ Assertions.assertFalse(bitList.remove("A"));
+
+ Assertions.assertTrue(bitList.removeAll(Collections.singletonList("B")));
+ Assertions.assertFalse(bitList.removeAll(Collections.singletonList("B")));
+
+ Assertions.assertFalse(bitList.removeAll(Collections.singletonList("D")));
+
+ bitList.add("D");
+ Assertions.assertTrue(bitList.removeAll(Collections.singletonList("D")));
+ Assertions.assertFalse(bitList.hasMoreElementInTailList());
+
+ bitList.add("A");
+ bitList.add("E");
+ bitList.add("F");
+
+ Assertions.assertEquals(4, bitList.size());
+
+ Assertions.assertFalse(bitList.removeAll(Collections.singletonList("D")));
+ Assertions.assertTrue(bitList.removeAll(Collections.singletonList("A")));
+ Assertions.assertTrue(bitList.removeAll(Collections.singletonList("C")));
+ Assertions.assertTrue(bitList.removeAll(Collections.singletonList("E")));
+ Assertions.assertTrue(bitList.removeAll(Collections.singletonList("F")));
+
+ Assertions.assertTrue(bitList.isEmpty());
+ }
+
+ @Test
+ public void testRemoveIndex() {
+ List<String> list = Arrays.asList("A", "B", "C");
+ BitList<String> bitList = new BitList<>(list);
+
+ Assertions.assertThrows(IndexOutOfBoundsException.class, () -> bitList.remove(3));
+ Assertions.assertNotNull(bitList.remove(1));
+ Assertions.assertNotNull(bitList.remove(0));
+ Assertions.assertThrows(IndexOutOfBoundsException.class, () -> bitList.remove(1));
+
+ bitList.add("D");
+ Assertions.assertNotNull(bitList.remove(1));
+
+ bitList.add("A");
+ bitList.add("E");
+ bitList.add("F");
+ // A C E F
+ Assertions.assertEquals(4, bitList.size());
+
+ Assertions.assertThrows(IndexOutOfBoundsException.class, () -> bitList.remove(4));
+ Assertions.assertEquals("F", bitList.remove(3));
+ Assertions.assertEquals("E", bitList.remove(2));
+ Assertions.assertEquals("C", bitList.remove(1));
+ Assertions.assertEquals("A", bitList.remove(0));
+
+ Assertions.assertThrows(IndexOutOfBoundsException.class, () -> bitList.remove(0));
+
+ Assertions.assertTrue(bitList.isEmpty());
+ }
+
+ @Test
+ public void testRetain() {
+ List<String> list = Arrays.asList("A", "B", "C");
+ BitList<String> bitList = new BitList<>(list);
+
+ List<String> list1 = Arrays.asList("B", "C");
+ Assertions.assertTrue(bitList.retainAll(list1));
+
+ Assertions.assertTrue(bitList.containsAll(list1));
+ Assertions.assertFalse(bitList.containsAll(list));
+ Assertions.assertFalse(bitList.contains("A"));
+
+ bitList = new BitList<>(list);
+ bitList.add("D");
+ bitList.add("E");
+ List<String> list2 = Arrays.asList("B", "C", "D");
+ Assertions.assertTrue(bitList.retainAll(list2));
+
+ Assertions.assertTrue(bitList.containsAll(list2));
+ Assertions.assertFalse(bitList.containsAll(list));
+ Assertions.assertFalse(bitList.contains("A"));
+ Assertions.assertFalse(bitList.contains("E"));
+ }
+
+ @Test
+ public void testIndex() {
+ List<String> list = Arrays.asList("A", "B", "A");
+ BitList<String> bitList = new BitList<>(list);
+
+ Assertions.assertEquals(0, bitList.indexOf("A"));
+ Assertions.assertEquals(2, bitList.lastIndexOf("A"));
+
+ Assertions.assertEquals(-1, bitList.indexOf("D"));
+ Assertions.assertEquals(-1, bitList.lastIndexOf("D"));
+
+ bitList.add("D");
+ bitList.add("E");
+
+ Assertions.assertEquals(0, bitList.indexOf("A"));
+ Assertions.assertEquals(2, bitList.lastIndexOf("A"));
+ Assertions.assertEquals(3, bitList.indexOf("D"));
+ Assertions.assertEquals(3, bitList.lastIndexOf("D"));
+
+ Assertions.assertEquals(-1, bitList.indexOf("F"));
+ }
+
+ @Test
+ public void testSubList() {
+ List<String> list = Arrays.asList("A", "B", "C", "D", "E");
+ BitList<String> bitList = new BitList<>(list);
+ bitList.addAll(Arrays.asList("F", "G", "H", "I"));
+
+ BitList<String> subList1 = bitList.subList(0, 5);
+ Assertions.assertEquals(Arrays.asList("A", "B", "C", "D", "E"), subList1);
+
+ BitList<String> subList2 = bitList.subList(1, 5);
+ Assertions.assertEquals(Arrays.asList("B", "C", "D", "E"), subList2);
+
+ BitList<String> subList3 = bitList.subList(0, 4);
+ Assertions.assertEquals(Arrays.asList("A", "B", "C", "D"), subList3);
+
+ BitList<String> subList4 = bitList.subList(1, 4);
+ Assertions.assertEquals(Arrays.asList("B", "C", "D"), subList4);
+
+ BitList<String> subList5 = bitList.subList(2, 3);
+ Assertions.assertEquals(Collections.singletonList("C"), subList5);
+ Assertions.assertFalse(subList5.hasMoreElementInTailList());
+
+ BitList<String> subList6 = bitList.subList(0, 9);
+ Assertions.assertEquals(Arrays.asList("A", "B", "C", "D", "E", "F", "G", "H", "I"), subList6);
+ Assertions.assertEquals(Arrays.asList("F", "G", "H", "I"), subList6.getTailList());
+
+ BitList<String> subList7 = bitList.subList(1, 8);
+ Assertions.assertEquals(Arrays.asList("B", "C", "D", "E", "F", "G", "H"), subList7);
+ Assertions.assertEquals(Arrays.asList("F", "G", "H"), subList7.getTailList());
+
+ BitList<String> subList8 = bitList.subList(4, 8);
+ Assertions.assertEquals(Arrays.asList("E", "F", "G", "H"), subList8);
+ Assertions.assertEquals(Arrays.asList("F", "G", "H"), subList8.getTailList());
+
+ BitList<String> subList9 = bitList.subList(5, 8);
+ Assertions.assertEquals(Arrays.asList("F", "G", "H"), subList9);
+ Assertions.assertEquals(Arrays.asList("F", "G", "H"), subList9.getTailList());
+
+ BitList<String> subList10 = bitList.subList(6, 8);
+ Assertions.assertEquals(Arrays.asList("G", "H"), subList10);
+ Assertions.assertEquals(Arrays.asList("G", "H"), subList10.getTailList());
+
+ BitList<String> subList11 = bitList.subList(6, 7);
+ Assertions.assertEquals(Collections.singletonList("G"), subList11);
+ Assertions.assertEquals(Collections.singletonList("G"), subList11.getTailList());
}
+ @Test
+ public void testListIterator1() {
+ List<String> list = Arrays.asList("A", "B", "C", "D", "E");
+ BitList<String> bitList = new BitList<>(list);
+
+ ListIterator<String> listIterator = bitList.listIterator(2);
+ ListIterator<String> expectedIterator = list.listIterator(2);
+
+ while (listIterator.hasNext()) {
+ Assertions.assertEquals(expectedIterator.nextIndex(), listIterator.nextIndex());
+ Assertions.assertEquals(expectedIterator.next(), listIterator.next());
+ }
+
+ while (listIterator.hasPrevious()) {
+ Assertions.assertEquals(expectedIterator.previousIndex(), listIterator.previousIndex());
+ Assertions.assertEquals(expectedIterator.previous(), listIterator.previous());
+ }
+ }
+
+ @Test
+ public void testListIterator2() {
+ List<String> list = Arrays.asList("A", "B", "C", "D", "E");
+ BitList<String> bitList = new BitList<>(list);
+ bitList.addAll(Arrays.asList("F", "G", "H", "I"));
+ List<String> expectedResult = Arrays.asList("A", "B", "C", "D", "E", "F", "G", "H", "I");
+
+ ListIterator<String> listIterator = bitList.listIterator(2);
+ ListIterator<String> expectedIterator = expectedResult.listIterator(2);
+
+ while (listIterator.hasNext()) {
+ Assertions.assertEquals(expectedIterator.nextIndex(), listIterator.nextIndex());
+ Assertions.assertEquals(expectedIterator.next(), listIterator.next());
+ }
+
+ while (listIterator.hasPrevious()) {
+ Assertions.assertEquals(expectedIterator.previousIndex(), listIterator.previousIndex());
+ Assertions.assertEquals(expectedIterator.previous(), listIterator.previous());
+ }
+ }
+
+ @Test
+ public void testListIterator3() {
+ List<String> list = Arrays.asList("A", "B", "C", "D", "E");
+ BitList<String> bitList = new BitList<>(list);
+ bitList.addAll(Arrays.asList("F", "G", "H", "I"));
+ List<String> expectedResult = Arrays.asList("A", "B", "C", "D", "E", "F", "G", "H", "I");
+
+ ListIterator<String> listIterator = bitList.listIterator(7);
+ ListIterator<String> expectedIterator = expectedResult.listIterator(7);
+
+ while (listIterator.hasNext()) {
+ Assertions.assertEquals(expectedIterator.nextIndex(), listIterator.nextIndex());
+ Assertions.assertEquals(expectedIterator.next(), listIterator.next());
+ }
+
+ while (listIterator.hasPrevious()) {
+ Assertions.assertEquals(expectedIterator.previousIndex(), listIterator.previousIndex());
+ Assertions.assertEquals(expectedIterator.previous(), listIterator.previous());
+ }
+ }
+
+ @Test
+ public void testListIterator4() {
+ List<String> list = Arrays.asList("A", "B", "C", "D", "E");
+ BitList<String> bitList = new BitList<>(list);
+ bitList.addAll(Arrays.asList("F", "G", "H", "I"));
+ List<String> expectedResult = Arrays.asList("A", "B", "C", "D", "E", "F", "G", "H", "I");
+
+ ListIterator<String> listIterator = bitList.listIterator(8);
+ ListIterator<String> expectedIterator = expectedResult.listIterator(8);
+
+ while (listIterator.hasNext()) {
+ Assertions.assertEquals(expectedIterator.nextIndex(), listIterator.nextIndex());
+ Assertions.assertEquals(expectedIterator.next(), listIterator.next());
+ }
+
+ while (listIterator.hasPrevious()) {
+ Assertions.assertEquals(expectedIterator.previousIndex(), listIterator.previousIndex());
+ Assertions.assertEquals(expectedIterator.previous(), listIterator.previous());
+ }
+ }
+
+ @Test
+ public void testListIterator5() {
+ List<String> list = Arrays.asList("A", "B", "C", "D", "E");
+ BitList<String> bitList = new BitList<>(list);
+ bitList.addAll(Arrays.asList("F", "G", "H", "I"));
+ List<String> expectedResult = new LinkedList<>(Arrays.asList("A", "B", "C", "D", "E", "F", "G", "H", "I"));
+
+ ListIterator<String> listIterator = bitList.listIterator(2);
+ ListIterator<String> expectedIterator = expectedResult.listIterator(2);
+
+ while (listIterator.hasNext()) {
+ Assertions.assertEquals(expectedIterator.nextIndex(), listIterator.nextIndex());
+ Assertions.assertEquals(expectedIterator.next(), listIterator.next());
+ listIterator.remove();
+ expectedIterator.remove();
+ }
+
+ Assertions.assertEquals(expectedResult, bitList);
+
+
+ while (listIterator.hasPrevious()) {
+ Assertions.assertEquals(expectedIterator.previousIndex(), listIterator.previousIndex());
+ Assertions.assertEquals(expectedIterator.previous(), listIterator.previous());
+ listIterator.remove();
+ expectedIterator.remove();
+ }
+
+ Assertions.assertEquals(expectedResult, bitList);
+ }
+
+ @Test
+ public void testListIterator6() {
+ List<String> list = Arrays.asList("A", "B", "C", "D", "E");
+ BitList<String> bitList = new BitList<>(list);
+ bitList.addAll(Arrays.asList("F", "G", "H", "I"));
+ List<String> expectedResult = new LinkedList<>(Arrays.asList("A", "B", "C", "D", "E", "F", "G", "H", "I"));
+
+ ListIterator<String> listIterator = bitList.listIterator(2);
+ ListIterator<String> expectedIterator = expectedResult.listIterator(2);
+
+ while (listIterator.hasPrevious()) {
+ Assertions.assertEquals(expectedIterator.previousIndex(), listIterator.previousIndex());
+ Assertions.assertEquals(expectedIterator.previous(), listIterator.previous());
+ listIterator.remove();
+ expectedIterator.remove();
+ }
+
+ Assertions.assertEquals(expectedResult, bitList);
+
+ while (listIterator.hasNext()) {
+ Assertions.assertEquals(expectedIterator.nextIndex(), listIterator.nextIndex());
+ Assertions.assertEquals(expectedIterator.next(), listIterator.next());
+ listIterator.remove();
+ expectedIterator.remove();
+ }
+
+ Assertions.assertEquals(expectedResult, bitList);
+ }
+
+ @Test
+ public void testListIterator7() {
+ List<String> list = Arrays.asList("A", "B", "C", "D", "E");
+ BitList<String> bitList = new BitList<>(list);
+ bitList.addAll(Arrays.asList("F", "G", "H", "I"));
+ List<String> expectedResult = new LinkedList<>(Arrays.asList("A", "B", "C", "D", "E", "F", "G", "H", "I"));
+
+ ListIterator<String> listIterator = bitList.listIterator(7);
+ ListIterator<String> expectedIterator = expectedResult.listIterator(7);
+
+ while (listIterator.hasNext()) {
+ Assertions.assertEquals(expectedIterator.nextIndex(), listIterator.nextIndex());
+ Assertions.assertEquals(expectedIterator.next(), listIterator.next());
+ listIterator.remove();
+ expectedIterator.remove();
+ }
+
+ Assertions.assertEquals(expectedResult, bitList);
+
+ while (listIterator.hasPrevious()) {
+ Assertions.assertEquals(expectedIterator.previousIndex(), listIterator.previousIndex());
+ Assertions.assertEquals(expectedIterator.previous(), listIterator.previous());
+ listIterator.remove();
+ expectedIterator.remove();
+ }
+
+ Assertions.assertEquals(expectedResult, bitList);
+ }
+
+ @Test
+ public void testListIterator8() {
+ List<String> list = Arrays.asList("A", "B", "C", "D", "E");
+ BitList<String> bitList = new BitList<>(list);
+ bitList.addAll(Arrays.asList("F", "G", "H", "I"));
+ List<String> expectedResult = new LinkedList<>(Arrays.asList("A", "B", "C", "D", "E", "F", "G", "H", "I"));
+
+ ListIterator<String> listIterator = bitList.listIterator(7);
+ ListIterator<String> expectedIterator = expectedResult.listIterator(7);
+
+ while (listIterator.hasPrevious()) {
+ Assertions.assertEquals(expectedIterator.previousIndex(), listIterator.previousIndex());
+ Assertions.assertEquals(expectedIterator.previous(), listIterator.previous());
+ listIterator.remove();
+ expectedIterator.remove();
+ }
+
+ Assertions.assertEquals(expectedResult, bitList);
+
+ while (listIterator.hasNext()) {
+ Assertions.assertEquals(expectedIterator.nextIndex(), listIterator.nextIndex());
+ Assertions.assertEquals(expectedIterator.next(), listIterator.next());
+ listIterator.remove();
+ expectedIterator.remove();
+ }
+
+ Assertions.assertEquals(expectedResult, bitList);
+ }
+
+ @Test
+ public void testClone1() {
+ List<String> list = Arrays.asList("A", "B", "C", "D", "E");
+ BitList<String> bitList = new BitList<>(list);
+
+ BitList<String> clone1 = bitList.clone();
+ Assertions.assertNotSame(bitList, clone1);
+ Assertions.assertEquals(bitList, clone1);
+
+ HashSet<Object> set = new HashSet<>();
+ set.add(bitList);
+ set.add(clone1);
+ Assertions.assertEquals(1, set.size());
+
+ set.add(new LinkedList<>());
+ Assertions.assertEquals(2, set.size());
+
+ set.add(new LinkedList<>(Arrays.asList("A", "B", "C", "D", "E")));
+ Assertions.assertEquals(2, set.size());
+ }
+
+ @Test
+ public void testClone2() {
+ List<String> list = Arrays.asList("A", "B", "C", "D", "E");
+ BitList<String> bitList = new BitList<>(list);
+ bitList.addAll(Arrays.asList("F", "G"));
+
+ BitList<String> clone1 = bitList.clone();
+ Assertions.assertNotSame(bitList, clone1);
+ Assertions.assertEquals(bitList, clone1);
+
+ HashSet<Object> set = new HashSet<>();
+ set.add(bitList);
+ set.add(clone1);
+ Assertions.assertEquals(1, set.size());
+
+ set.add(new LinkedList<>());
+ Assertions.assertEquals(2, set.size());
+
+ set.add(new LinkedList<>(Arrays.asList("A", "B", "C", "D", "E", "F", "G")));
+ Assertions.assertEquals(2, set.size());
+ }
}
diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/AbstractClusterInvokerTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/AbstractClusterInvokerTest.java
index 0c3dec7..c3bff9f 100644
--- a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/AbstractClusterInvokerTest.java
+++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/AbstractClusterInvokerTest.java
@@ -33,6 +33,8 @@ import org.apache.dubbo.rpc.cluster.filter.DemoService;
import org.apache.dubbo.rpc.cluster.loadbalance.LeastActiveLoadBalance;
import org.apache.dubbo.rpc.cluster.loadbalance.RandomLoadBalance;
import org.apache.dubbo.rpc.cluster.loadbalance.RoundRobinLoadBalance;
+
+import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
@@ -49,6 +51,7 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import static org.apache.dubbo.common.constants.CommonConstants.DUBBO;
+import static org.apache.dubbo.common.constants.CommonConstants.ENABLE_CONNECTIVITY_VALIDATION;
import static org.apache.dubbo.common.constants.CommonConstants.INTERFACE_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.MONITOR_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.PATH_KEY;
@@ -83,9 +86,9 @@ public class AbstractClusterInvokerTest {
Invoker<IHelloService> invoker5;
Invoker<IHelloService> mockedInvoker1;
-
@BeforeAll
public static void setUpBeforeClass() throws Exception {
+ System.setProperty(ENABLE_CONNECTIVITY_VALIDATION, "false");
}
@AfterEach
@@ -93,6 +96,11 @@ public class AbstractClusterInvokerTest {
RpcContext.removeContext();
}
+ @AfterAll
+ public static void afterClass() {
+ System.clearProperty(ENABLE_CONNECTIVITY_VALIDATION);
+ }
+
@SuppressWarnings({"unchecked"})
@BeforeEach
public void setUp() throws Exception {
@@ -487,6 +495,7 @@ public class AbstractClusterInvokerTest {
}
private void initDic() {
+ dic.notify(invokers);
dic.buildRouterChain();
}
diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/ConnectivityValidationTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/ConnectivityValidationTest.java
new file mode 100644
index 0000000..99b1684
--- /dev/null
+++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/ConnectivityValidationTest.java
@@ -0,0 +1,292 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.dubbo.rpc.cluster.support;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.constants.CommonConstants;
+import org.apache.dubbo.common.utils.CollectionUtils;
+import org.apache.dubbo.rpc.Invocation;
+import org.apache.dubbo.rpc.Invoker;
+import org.apache.dubbo.rpc.Result;
+import org.apache.dubbo.rpc.RpcException;
+import org.apache.dubbo.rpc.RpcInvocation;
+import org.apache.dubbo.rpc.cluster.Directory;
+import org.apache.dubbo.rpc.cluster.LoadBalance;
+import org.apache.dubbo.rpc.cluster.directory.StaticDirectory;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.util.Collections;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ThreadLocalRandom;
+
+import static org.mockito.Mockito.when;
+
+@SuppressWarnings("all")
+public class ConnectivityValidationTest {
+ private Invoker invoker1;
+ private Invoker invoker2;
+ private Invoker invoker3;
+ private Invoker invoker4;
+ private Invoker invoker5;
+ private Invoker invoker6;
+ private Invoker invoker7;
+ private Invoker invoker8;
+ private Invoker invoker9;
+ private Invoker invoker10;
+ private Invoker invoker11;
+ private Invoker invoker12;
+ private Invoker invoker13;
+ private Invoker invoker14;
+ private Invoker invoker15;
+
+ private List<Invoker> invokerList;
+
+ private StaticDirectory directory;
+ private ConnectivityClusterInvoker clusterInvoker;
+
+ @BeforeEach
+ public void setup() {
+ invoker1 = Mockito.mock(Invoker.class);
+ invoker2 = Mockito.mock(Invoker.class);
+ invoker3 = Mockito.mock(Invoker.class);
+ invoker4 = Mockito.mock(Invoker.class);
+ invoker5 = Mockito.mock(Invoker.class);
+ invoker6 = Mockito.mock(Invoker.class);
+ invoker7 = Mockito.mock(Invoker.class);
+ invoker8 = Mockito.mock(Invoker.class);
+ invoker9 = Mockito.mock(Invoker.class);
+ invoker10 = Mockito.mock(Invoker.class);
+ invoker11 = Mockito.mock(Invoker.class);
+ invoker12 = Mockito.mock(Invoker.class);
+ invoker13 = Mockito.mock(Invoker.class);
+ invoker14 = Mockito.mock(Invoker.class);
+ invoker15 = Mockito.mock(Invoker.class);
+
+ configInvoker(invoker1);
+ configInvoker(invoker2);
+ configInvoker(invoker3);
+ configInvoker(invoker4);
+ configInvoker(invoker5);
+ configInvoker(invoker6);
+ configInvoker(invoker7);
+ configInvoker(invoker8);
+ configInvoker(invoker9);
+ configInvoker(invoker10);
+ configInvoker(invoker11);
+ configInvoker(invoker12);
+ configInvoker(invoker13);
+ configInvoker(invoker14);
+ configInvoker(invoker15);
+
+ invokerList = new LinkedList<>();
+ invokerList.add(invoker1);
+ invokerList.add(invoker2);
+ invokerList.add(invoker3);
+ invokerList.add(invoker4);
+ invokerList.add(invoker5);
+
+ directory = new StaticDirectory(invokerList);
+ clusterInvoker = new ConnectivityClusterInvoker(directory);
+ }
+
+ private void configInvoker(Invoker invoker) {
+ when(invoker.getUrl()).thenReturn(URL.valueOf(""));
+ when(invoker.isAvailable()).thenReturn(true);
+ }
+
+ @BeforeAll
+ public static void setupClass() {
+ System.setProperty(CommonConstants.RECONNECT_TASK_PERIOD, "1");
+ }
+
+ @AfterAll
+ public static void clearAfterClass() {
+ System.clearProperty(CommonConstants.RECONNECT_TASK_PERIOD);
+ }
+
+ @Test
+ public void testBasic() throws InterruptedException {
+ Invocation invocation = new RpcInvocation();
+ LoadBalance loadBalance = new RandomLoadBalance();
+
+ Assertions.assertEquals(5, directory.list(invocation).size());
+
+ Assertions.assertNotNull(clusterInvoker.select(loadBalance, invocation, directory.list(invocation), Collections.emptyList()));
+
+ when(invoker1.isAvailable()).thenReturn(false);
+ when(invoker2.isAvailable()).thenReturn(false);
+ when(invoker3.isAvailable()).thenReturn(false);
+ when(invoker4.isAvailable()).thenReturn(false);
+ when(invoker5.isAvailable()).thenReturn(false);
+
+ clusterInvoker.select(loadBalance, invocation, directory.list(invocation), Collections.emptyList());
+ Assertions.assertEquals(0, directory.list(invocation).size());
+
+ when(invoker1.isAvailable()).thenReturn(true);
+ waitRefresh();
+ Assertions.assertEquals(1, directory.list(invocation).size());
+ Assertions.assertNotNull(clusterInvoker.select(loadBalance, invocation, directory.list(invocation), Collections.emptyList()));
+
+ when(invoker2.isAvailable()).thenReturn(true);
+ waitRefresh();
+ Assertions.assertEquals(2, directory.list(invocation).size());
+ Assertions.assertNotNull(clusterInvoker.select(loadBalance, invocation, directory.list(invocation), Collections.emptyList()));
+
+ invokerList.remove(invoker5);
+ directory.notify(invokerList);
+ when(invoker2.isAvailable()).thenReturn(true);
+ waitRefresh();
+ Assertions.assertEquals(2, directory.list(invocation).size());
+ Assertions.assertNotNull(clusterInvoker.select(loadBalance, invocation, directory.list(invocation), Collections.emptyList()));
+
+ when(invoker3.isAvailable()).thenReturn(true);
+ when(invoker4.isAvailable()).thenReturn(true);
+
+ waitRefresh();
+ Assertions.assertEquals(4, directory.list(invocation).size());
+ Assertions.assertNotNull(clusterInvoker.select(loadBalance, invocation, directory.list(invocation), Collections.emptyList()));
+ }
+
+ @Test
+ public void testRetry() throws InterruptedException {
+ Invocation invocation = new RpcInvocation();
+ LoadBalance loadBalance = new RandomLoadBalance();
+
+ invokerList.clear();
+ invokerList.add(invoker1);
+ invokerList.add(invoker2);
+ directory.notify(invokerList);
+
+ Assertions.assertEquals(2, directory.list(invocation).size());
+
+ when(invoker1.isAvailable()).thenReturn(false);
+ Assertions.assertEquals(invoker2, clusterInvoker.select(loadBalance, invocation, directory.list(invocation), Collections.singletonList(invoker2)));
+ Assertions.assertEquals(1, directory.list(invocation).size());
+
+ when(invoker1.isAvailable()).thenReturn(true);
+ waitRefresh();
+ Assertions.assertEquals(2, directory.list(invocation).size());
+ }
+
+ @Test
+ public void testRandomSelect() throws InterruptedException {
+ Invocation invocation = new RpcInvocation();
+ LoadBalance loadBalance = new RandomLoadBalance();
+
+ invokerList.add(invoker6);
+ invokerList.add(invoker7);
+ invokerList.add(invoker8);
+ invokerList.add(invoker9);
+ invokerList.add(invoker10);
+ invokerList.add(invoker11);
+ invokerList.add(invoker12);
+ invokerList.add(invoker13);
+ invokerList.add(invoker14);
+ invokerList.add(invoker15);
+
+ directory.notify(invokerList);
+
+ Assertions.assertEquals(15, directory.list(invocation).size());
+
+ when(invoker2.isAvailable()).thenReturn(false);
+ when(invoker3.isAvailable()).thenReturn(false);
+ when(invoker4.isAvailable()).thenReturn(false);
+ when(invoker5.isAvailable()).thenReturn(false);
+ when(invoker6.isAvailable()).thenReturn(false);
+ when(invoker7.isAvailable()).thenReturn(false);
+ when(invoker8.isAvailable()).thenReturn(false);
+ when(invoker9.isAvailable()).thenReturn(false);
+ when(invoker10.isAvailable()).thenReturn(false);
+ when(invoker11.isAvailable()).thenReturn(false);
+ when(invoker12.isAvailable()).thenReturn(false);
+ when(invoker13.isAvailable()).thenReturn(false);
+ when(invoker14.isAvailable()).thenReturn(false);
+ when(invoker15.isAvailable()).thenReturn(false);
+ for (int i = 0; i < 15; i++) {
+ clusterInvoker.select(loadBalance, invocation, directory.list(invocation), Collections.emptyList());
+ }
+ for (int i = 0; i < 5; i++) {
+ Assertions.assertEquals(invoker1, clusterInvoker.select(loadBalance, invocation, directory.list(invocation), Collections.emptyList()));
+ }
+
+ when(invoker1.isAvailable()).thenReturn(false);
+ clusterInvoker.select(loadBalance, invocation, directory.list(invocation), Collections.emptyList());
+ Assertions.assertEquals(0, directory.list(invocation).size());
+
+ when(invoker1.isAvailable()).thenReturn(true);
+ when(invoker2.isAvailable()).thenReturn(true);
+ when(invoker3.isAvailable()).thenReturn(true);
+ when(invoker4.isAvailable()).thenReturn(true);
+ when(invoker5.isAvailable()).thenReturn(true);
+ when(invoker6.isAvailable()).thenReturn(true);
+ when(invoker7.isAvailable()).thenReturn(true);
+ when(invoker8.isAvailable()).thenReturn(true);
+ when(invoker9.isAvailable()).thenReturn(true);
+ when(invoker10.isAvailable()).thenReturn(true);
+ when(invoker11.isAvailable()).thenReturn(true);
+ when(invoker12.isAvailable()).thenReturn(true);
+ when(invoker13.isAvailable()).thenReturn(true);
+ when(invoker14.isAvailable()).thenReturn(true);
+ when(invoker15.isAvailable()).thenReturn(true);
+
+ waitRefresh();
+ Assertions.assertTrue(directory.list(invocation).size() > 1);
+ }
+
+ private static class ConnectivityClusterInvoker<T> extends AbstractClusterInvoker<T> {
+ public ConnectivityClusterInvoker(Directory<T> directory) {
+ super(directory);
+ }
+
+ @Override
+ public Invoker<T> select(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {
+ return super.select(loadbalance, invocation, invokers, selected);
+ }
+
+ @Override
+ protected Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
+ return null;
+ }
+ }
+
+ private void waitRefresh() throws InterruptedException {
+ ScheduledFuture future = directory.getConnectivityCheckFuture();
+ while (!future.isDone()) {
+ Thread.sleep(10);
+ }
+ directory.checkConnectivity();
+ future = directory.getConnectivityCheckFuture();
+ while (!future.isDone()) {
+ Thread.sleep(10);
+ }
+ }
+
+ private static class RandomLoadBalance implements LoadBalance {
+ @Override
+ public <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) throws RpcException {
+ return CollectionUtils.isNotEmpty(invokers) ? invokers.get(ThreadLocalRandom.current().nextInt(invokers.size())) : null;
+ }
+ }
+}
diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/FailoverClusterInvokerTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/FailoverClusterInvokerTest.java
index 4ec9770..4b9fba5 100644
--- a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/FailoverClusterInvokerTest.java
+++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/FailoverClusterInvokerTest.java
@@ -27,6 +27,7 @@ import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.RpcInvocation;
import org.apache.dubbo.rpc.cluster.Directory;
import org.apache.dubbo.rpc.cluster.directory.StaticDirectory;
+import org.apache.dubbo.rpc.cluster.router.state.BitList;
import org.apache.dubbo.rpc.protocol.AbstractInvoker;
import org.junit.jupiter.api.BeforeEach;
@@ -285,6 +286,8 @@ public class FailoverClusterInvokerTest {
invokers.add(invoker1);
invokers.add(invoker2);
+ MockDirectory<Demo> dic = new MockDirectory<>(url, invokers);
+
Callable<Object> callable = () -> {
//Simulation: all invokers are destroyed
for (Invoker<Demo> invoker : invokers) {
@@ -294,6 +297,7 @@ public class FailoverClusterInvokerTest {
MockInvoker<Demo> invoker3 = new MockInvoker<>(Demo.class, url);
invoker3.setResult(AsyncRpcResult.newDefaultAsyncResult(null));
invokers.add(invoker3);
+ dic.notify(invokers);
return null;
};
invoker1.setCallable(callable);
@@ -302,8 +306,6 @@ public class FailoverClusterInvokerTest {
RpcInvocation inv = new RpcInvocation();
inv.setMethodName("test");
- Directory<Demo> dic = new MockDirectory<>(url, invokers);
-
FailoverClusterInvoker<Demo> clusterInvoker = new FailoverClusterInvoker<>(dic);
clusterInvoker.invoke(inv);
}
@@ -358,8 +360,8 @@ public class FailoverClusterInvokerTest {
}
@Override
- protected List<Invoker<T>> doList(Invocation invocation) throws RpcException {
- return new ArrayList<>(super.doList(invocation));
+ protected List<Invoker<T>> doList(BitList<Invoker<T>> invokers, Invocation invocation) throws RpcException {
+ return super.doList(invokers, invocation);
}
}
}
diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java b/dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java
index d430e9b..0701286 100644
--- a/dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java
+++ b/dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java
@@ -484,6 +484,19 @@ public interface CommonConstants {
String OS_WIN_PREFIX = "win";
- String DUBBO_INTERNAL_APPLICATION = "DUBBO_INTERNAL_APPLICATION";
+ String RECONNECT_TASK_TRY_COUNT = "dubbo.reconnect.reconnectTaskTryCount";
+
+ int DEFAULT_RECONNECT_TASK_TRY_COUNT = 10;
+
+ String RECONNECT_TASK_PERIOD = "dubbo.reconnect.reconnectTaskPeriod";
+
+ int DEFAULT_RECONNECT_TASK_PERIOD = 1000;
+ String RESELECT_COUNT = "dubbo.reselect.count";
+
+ int DEFAULT_RESELECT_COUNT = 10;
+
+ String ENABLE_CONNECTIVITY_VALIDATION = "dubbo.connectivity.validation";
+
+ String DUBBO_INTERNAL_APPLICATION = "DUBBO_INTERNAL_APPLICATION";
}
diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/manager/DefaultExecutorRepository.java b/dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/manager/DefaultExecutorRepository.java
index 0ed4ac5..21572a2 100644
--- a/dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/manager/DefaultExecutorRepository.java
+++ b/dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/manager/DefaultExecutorRepository.java
@@ -68,7 +68,7 @@ public class DefaultExecutorRepository implements ExecutorRepository, ExtensionA
private volatile ExecutorService serviceReferExecutor;
- private ScheduledExecutorService reconnectScheduledExecutor;
+ private ScheduledExecutorService connectivityScheduledExecutor;
public Ring<ScheduledExecutorService> registryNotificationExecutorRing = new Ring<>();
@@ -102,7 +102,7 @@ public class DefaultExecutorRepository implements ExecutorRepository, ExtensionA
, new ThreadPoolExecutor.AbortPolicy()));
}
-// reconnectScheduledExecutor = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("Dubbo-reconnect-scheduler"));
+ connectivityScheduledExecutor = Executors.newScheduledThreadPool(DEFAULT_SCHEDULER_SIZE, new NamedThreadFactory("Dubbo-connectivity-scheduler", true));
poolRouterExecutor = new ThreadPoolExecutor(1, 10, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(1024),
new NamedInternalThreadFactory("Dubbo-state-router-pool-router", true), new ThreadPoolExecutor.AbortPolicy());
@@ -365,6 +365,11 @@ public class DefaultExecutorRepository implements ExecutorRepository, ExtensionA
}
@Override
+ public ScheduledExecutorService getConnectivityScheduledExecutor() {
+ return connectivityScheduledExecutor;
+ }
+
+ @Override
public void destroyAll() {
logger.info("destroying executor repository ..");
shutdownExecutorService(poolRouterExecutor, "poolRouterExecutor");
@@ -395,6 +400,9 @@ public class DefaultExecutorRepository implements ExecutorRepository, ExtensionA
// executorServiceRing
shutdownExecutorServices(executorServiceRing.listItems(), "executorServiceRing");
+ // connectivityScheduledExecutor
+ shutdownExecutorService(connectivityScheduledExecutor, "connectivityScheduledExecutor");
+
// shutdown share executor
shutdownExecutorService(sharedExecutor, "sharedExecutor");
shutdownExecutorService(sharedScheduledExecutor, "sharedScheduledExecutor");
diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/manager/ExecutorRepository.java b/dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/manager/ExecutorRepository.java
index 4e78ea8..b129f50 100644
--- a/dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/manager/ExecutorRepository.java
+++ b/dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/manager/ExecutorRepository.java
@@ -101,6 +101,13 @@ public interface ExecutorRepository {
ExecutorService getPoolRouterExecutor();
/**
+ * Scheduled executor handle connectivity check task
+ *
+ * @return
+ */
+ ScheduledExecutorService getConnectivityScheduledExecutor();
+
+ /**
* Destroy all executors that are not in shutdown state
*/
void destroyAll();
diff --git a/dubbo-dependencies-bom/pom.xml b/dubbo-dependencies-bom/pom.xml
index 64c6797..539a7cb 100644
--- a/dubbo-dependencies-bom/pom.xml
+++ b/dubbo-dependencies-bom/pom.xml
@@ -160,7 +160,6 @@
<hessian_lite_version>3.2.11</hessian_lite_version>
<swagger_version>1.5.24</swagger_version>
- <roaringbitmap_version>0.9.0</roaringbitmap_version>
<bouncycastle-bcprov_version>1.68</bouncycastle-bcprov_version>
<metrics_version>2.0.1</metrics_version>
<sofa_registry_version>5.2.0</sofa_registry_version>
@@ -336,11 +335,6 @@
<version>${protobuf-java_version}</version>
</dependency>
<dependency>
- <groupId>org.roaringbitmap</groupId>
- <artifactId>RoaringBitmap</artifactId>
- <version>${roaringbitmap_version}</version>
- </dependency>
- <dependency>
<groupId>org.bouncycastle</groupId>
<artifactId>bcprov-jdk15on</artifactId>
<version>${bouncycastle-bcprov_version}</version>
diff --git a/dubbo-distribution/dubbo-all/pom.xml b/dubbo-distribution/dubbo-all/pom.xml
index 727aca7..a9d3190 100644
--- a/dubbo-distribution/dubbo-all/pom.xml
+++ b/dubbo-distribution/dubbo-all/pom.xml
@@ -356,10 +356,6 @@
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
</dependency>
- <dependency>
- <groupId>org.roaringbitmap</groupId>
- <artifactId>RoaringBitmap</artifactId>
- </dependency>
<!-- Temporarily add this part to exclude transitive dependency -->
<dependency>
diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistryDirectory.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistryDirectory.java
index 37c20ab..7f10291 100644
--- a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistryDirectory.java
+++ b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistryDirectory.java
@@ -35,6 +35,7 @@ import org.apache.dubbo.rpc.RpcContext;
import org.apache.dubbo.rpc.RpcServiceContext;
import org.apache.dubbo.rpc.cluster.Configurator;
import org.apache.dubbo.rpc.cluster.RouterChain;
+import org.apache.dubbo.rpc.cluster.router.state.BitList;
import org.apache.dubbo.rpc.model.ModuleModel;
import java.util.ArrayList;
@@ -127,22 +128,6 @@ public class ServiceDiscoveryRegistryDirectory<T> extends DynamicDirectory<T> {
}
@Override
- public boolean isAvailable() {
- if (isDestroyed() || this.forbidden) {
- return false;
- }
- Map<String, Invoker<T>> localUrlInvokerMap = urlInvokerMap;
- if (localUrlInvokerMap != null && localUrlInvokerMap.size() > 0) {
- for (Invoker<T> invoker : new ArrayList<>(localUrlInvokerMap.values())) {
- if (invoker.isAvailable()) {
- return true;
- }
- }
- }
- return false;
- }
-
- @Override
public synchronized void notify(List<URL> instanceUrls) {
if (isDestroyed()) {
return;
@@ -219,7 +204,8 @@ public class ServiceDiscoveryRegistryDirectory<T> extends DynamicDirectory<T> {
if (invokerUrls.size() == 0) {
logger.info("Received empty url list...");
this.forbidden = true; // Forbid to access
- this.invokers = Collections.emptyList();
+ this.invokers = BitList.emptyList();
+ this.validInvokers = BitList.emptyList();
routerChain.setInvokers(this.invokers);
destroyAllInvokers(); // Close all invokers
} else {
@@ -243,10 +229,9 @@ public class ServiceDiscoveryRegistryDirectory<T> extends DynamicDirectory<T> {
return;
}
List<Invoker<T>> newInvokers = Collections.unmodifiableList(new ArrayList<>(newUrlInvokerMap.values()));
- // pre-route and build cache, notice that route cache should build on original Invoker list.
- // toMergeMethodInvokerMap() will wrap some invokers having different groups, those wrapped invokers not should be routed.
- routerChain.setInvokers(newInvokers);
- this.invokers = multiGroup ? toMergeInvokerList(newInvokers) : newInvokers;
+ this.invokers = multiGroup ? new BitList<>(toMergeInvokerList(newInvokers)) : new BitList<>(newInvokers);
+ // pre-route and build cache
+ routerChain.setInvokers(this.invokers);
this.urlInvokerMap = newUrlInvokerMap;
if (oldUrlInvokerMap != null) {
@@ -366,6 +351,7 @@ public class ServiceDiscoveryRegistryDirectory<T> extends DynamicDirectory<T> {
this.urlInvokerMap = null;
this.invokers = null;
+ this.validInvokers = null;
}
/**
diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/DynamicDirectory.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/DynamicDirectory.java
index f7339d7..cc709e8 100644
--- a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/DynamicDirectory.java
+++ b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/DynamicDirectory.java
@@ -21,6 +21,7 @@ import org.apache.dubbo.common.Version;
import org.apache.dubbo.common.extension.ExtensionLoader;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.NetUtils;
import org.apache.dubbo.registry.AddressListener;
import org.apache.dubbo.registry.NotifyListener;
@@ -36,6 +37,7 @@ import org.apache.dubbo.rpc.cluster.Configurator;
import org.apache.dubbo.rpc.cluster.RouterChain;
import org.apache.dubbo.rpc.cluster.RouterFactory;
import org.apache.dubbo.rpc.cluster.directory.AbstractDirectory;
+import org.apache.dubbo.rpc.cluster.router.state.BitList;
import java.util.Collections;
import java.util.List;
@@ -104,8 +106,6 @@ public abstract class DynamicDirectory<T> extends AbstractDirectory<T> implement
*/
protected volatile List<Configurator> configurators;
- protected volatile List<Invoker<T>> invokers;
-
protected ServiceInstancesChangedListener serviceListener;
public DynamicDirectory(Class<T> serviceType, URL url) {
@@ -165,7 +165,7 @@ public abstract class DynamicDirectory<T> extends AbstractDirectory<T> implement
}
@Override
- public List<Invoker<T>> doList(Invocation invocation) {
+ public List<Invoker<T>> doList(BitList<Invoker<T>> invokers, Invocation invocation) {
if (forbidden) {
// 1. No service provider 2. Service providers are disabled
throw new RpcException(RpcException.FORBIDDEN_EXCEPTION, "No provider available from registry " +
@@ -175,18 +175,17 @@ public abstract class DynamicDirectory<T> extends AbstractDirectory<T> implement
}
if (multiGroup) {
- return this.invokers == null ? Collections.emptyList() : this.invokers;
+ return this.invokers == null ? BitList.emptyList() : this.invokers;
}
- List<Invoker<T>> invokers = null;
try {
// Get invokers from cache, only runtime routers will be executed.
- invokers = routerChain.route(getConsumerUrl(), invocation);
+ List<Invoker<T>> result = routerChain.route(getConsumerUrl(), invokers, invocation);
+ return result == null ? BitList.emptyList() : result;
} catch (Throwable t) {
logger.error("Failed to execute router: " + getUrl() + ", cause: " + t.getMessage(), t);
+ return BitList.emptyList();
}
-
- return invokers == null ? Collections.emptyList() : invokers;
}
@Override
@@ -259,6 +258,15 @@ public abstract class DynamicDirectory<T> extends AbstractDirectory<T> implement
}
@Override
+ public boolean isAvailable() {
+ if (isDestroyed() || this.forbidden) {
+ return false;
+ }
+ return CollectionUtils.isNotEmpty(validInvokers)
+ && validInvokers.stream().anyMatch(Invoker::isAvailable);
+ }
+
+ @Override
public void destroy() {
if (isDestroyed()) {
return;
@@ -323,6 +331,7 @@ public abstract class DynamicDirectory<T> extends AbstractDirectory<T> implement
}
protected synchronized void invokersChanged() {
+ refreshInvoker();
invokersChanged = true;
if (invokersChangedListener != null) {
invokersChangedListener.onChange();
diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/RegistryDirectory.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/RegistryDirectory.java
index 9711f14..1e24630 100644
--- a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/RegistryDirectory.java
+++ b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/RegistryDirectory.java
@@ -18,7 +18,6 @@ package org.apache.dubbo.registry.integration;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.URLBuilder;
-import org.apache.dubbo.common.Version;
import org.apache.dubbo.common.config.configcenter.DynamicConfiguration;
import org.apache.dubbo.common.extension.ExtensionLoader;
import org.apache.dubbo.common.logger.Logger;
@@ -33,13 +32,12 @@ import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.common.utils.UrlUtils;
import org.apache.dubbo.registry.AddressListener;
import org.apache.dubbo.remoting.Constants;
-import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.Protocol;
-import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.cluster.Configurator;
import org.apache.dubbo.rpc.cluster.Router;
import org.apache.dubbo.rpc.cluster.directory.StaticDirectory;
+import org.apache.dubbo.rpc.cluster.router.state.BitList;
import org.apache.dubbo.rpc.cluster.support.ClusterUtils;
import org.apache.dubbo.rpc.model.ApplicationModel;
import org.apache.dubbo.rpc.model.ModuleModel;
@@ -205,7 +203,8 @@ public class RegistryDirectory<T> extends DynamicDirectory<T> {
&& invokerUrls.get(0) != null
&& EMPTY_PROTOCOL.equals(invokerUrls.get(0).getProtocol())) {
this.forbidden = true; // Forbid to access
- this.invokers = Collections.emptyList();
+ this.invokers = BitList.emptyList();
+ this.validInvokers = BitList.emptyList();
routerChain.setInvokers(this.invokers);
destroyAllInvokers(); // Close all invokers
} else {
@@ -248,10 +247,9 @@ public class RegistryDirectory<T> extends DynamicDirectory<T> {
}
List<Invoker<T>> newInvokers = Collections.unmodifiableList(new ArrayList<>(newUrlInvokerMap.values()));
- // pre-route and build cache, notice that route cache should build on original Invoker list.
- // toMergeMethodInvokerMap() will wrap some invokers having different groups, those wrapped invokers not should be routed.
- routerChain.setInvokers(newInvokers);
- this.invokers = multiGroup ? toMergeInvokerList(newInvokers) : newInvokers;
+ this.invokers = multiGroup ? new BitList<>(toMergeInvokerList(newInvokers)) : new BitList<>(newInvokers);
+ // pre-route and build cache
+ routerChain.setInvokers(this.invokers);
this.urlInvokerMap = newUrlInvokerMap;
try {
@@ -488,6 +486,7 @@ public class RegistryDirectory<T> extends DynamicDirectory<T> {
localUrlInvokerMap.clear();
}
invokers = null;
+ validInvokers = null;
cachedInvokerUrls = null;
}
@@ -519,31 +518,6 @@ public class RegistryDirectory<T> extends DynamicDirectory<T> {
}
@Override
- public List<Invoker<T>> doList(Invocation invocation) {
- if (forbidden) {
- // 1. No service provider 2. Service providers are disabled
- throw new RpcException(RpcException.FORBIDDEN_EXCEPTION, "No provider available from registry " +
- getUrl().getAddress() + " for service " + getConsumerUrl().getServiceKey() + " on consumer " +
- NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion() +
- ", please check status of providers(disabled, not registered or in blacklist).");
- }
-
- if (multiGroup) {
- return this.invokers == null ? Collections.emptyList() : this.invokers;
- }
-
- List<Invoker<T>> invokers = null;
- try {
- // Get invokers from cache, only runtime routers will be executed.
- invokers = routerChain.route(getConsumerUrl(), invocation);
- } catch (Throwable t) {
- logger.error("Failed to execute router: " + getUrl() + ", cause: " + t.getMessage(), t);
- }
-
- return invokers == null ? Collections.emptyList() : invokers;
- }
-
- @Override
public Class<T> getInterface() {
return serviceType;
}
@@ -574,16 +548,6 @@ public class RegistryDirectory<T> extends DynamicDirectory<T> {
}
}
- @Override
- public boolean isAvailable() {
- if (isDestroyed() || this.forbidden) {
- return false;
- }
- Map<URL, Invoker<T>> localUrlInvokerMap = urlInvokerMap;
- return CollectionUtils.isNotEmptyMap(localUrlInvokerMap)
- && localUrlInvokerMap.values().stream().anyMatch(Invoker::isAvailable);
- }
-
/**
* Haomin: added for test purpose
*/