This is an automated email from the ASF dual-hosted git repository.
albumenj pushed a commit to branch 3.2
in repository https://gitbox.apache.org/repos/asf/dubbo.git
The following commit(s) were added to refs/heads/3.2 by this push:
new b56687957e Fix invokers has changed when router chain switch (#11248)
b56687957e is described below
commit b56687957e3dbb36d49a3da3f8959a53a953b1ae
Author: Albumen Kevin <[email protected]>
AuthorDate: Sat Jan 14 14:34:50 2023 +0800
Fix invokers has changed when router chain switch (#11248)
---
.../org/apache/dubbo/rpc/cluster/RouterChain.java | 158 +++++++++++++--------
.../dubbo/rpc/cluster/SingleRouterChain.java | 52 ++++---
.../rpc/cluster/directory/AbstractDirectory.java | 109 +++++++++-----
.../rpc/cluster/directory/StaticDirectory.java | 18 +--
.../apache/dubbo/rpc/cluster/RouterChainTest.java | 19 +--
.../loadbalance/ConsistentHashLoadBalanceTest.java | 14 +-
.../support/FailoverClusterInvokerTest.java | 14 +-
.../client/ServiceDiscoveryRegistryDirectory.java | 32 ++---
.../registry/integration/DynamicDirectory.java | 12 +-
.../registry/integration/RegistryDirectory.java | 34 ++---
10 files changed, 270 insertions(+), 192 deletions(-)
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 a5fa88fc63..e0d15af1fe 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
@@ -16,8 +16,15 @@
*/
package org.apache.dubbo.rpc.cluster;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.stream.Collectors;
+
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.config.ConfigurationUtils;
+import org.apache.dubbo.common.constants.LoggerCodeConstants;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.rpc.Invocation;
@@ -29,12 +36,6 @@ import
org.apache.dubbo.rpc.cluster.router.state.StateRouterFactory;
import org.apache.dubbo.rpc.model.ModuleModel;
import org.apache.dubbo.rpc.model.ScopeModelUtil;
-import java.util.List;
-import java.util.concurrent.atomic.AtomicReference;
-import java.util.stream.Collectors;
-
-import static
org.apache.dubbo.common.constants.LoggerCodeConstants.INTERNAL_INTERRUPTED;
-import static
org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_ROUTER_WAIT_LONG;
import static org.apache.dubbo.rpc.cluster.Constants.ROUTER_KEY;
/**
@@ -43,8 +44,8 @@ import static
org.apache.dubbo.rpc.cluster.Constants.ROUTER_KEY;
public class RouterChain<T> {
private static final ErrorTypeAwareLogger logger =
LoggerFactory.getErrorTypeAwareLogger(RouterChain.class);
- private final SingleRouterChain<T> mainChain;
- private final SingleRouterChain<T> backupChain;
+ private volatile SingleRouterChain<T> mainChain;
+ private volatile SingleRouterChain<T> backupChain;
private volatile SingleRouterChain<T> currentChain;
@SuppressWarnings({"rawtypes", "unchecked"})
@@ -91,18 +92,34 @@ public class RouterChain<T> {
private final AtomicReference<BitList<Invoker<T>>> notifyingInvokers = new
AtomicReference<>();
- public List<Invoker<T>> route(URL url, BitList<Invoker<T>>
availableInvokers, Invocation invocation) {
- if (notifyingInvokers.get() != null) {
- if (availableInvokers.getOriginList() ==
notifyingInvokers.get().getOriginList()) {
- // notify switch chain
- switchToMainChain();
- }
+ private final ReadWriteLock lock = new ReentrantReadWriteLock();
+
+ public ReadWriteLock getLock() {
+ return lock;
+ }
+
+ public SingleRouterChain<T> getSingleChain(URL url, BitList<Invoker<T>>
availableInvokers, Invocation invocation) {
+ // If current is in:
+ // 1. `setInvokers` is in progress
+ // 2. Most of the invocation should use backup chain => currentChain
== backupChain
+ // 3. Main chain has been update success => notifyingInvokers.get() !=
null
+ // If `availableInvokers` is created from origin invokers => use
backup chain
+ // If `availableInvokers` is created from newly invokers => use
main chain
+ BitList<Invoker<T>> notifying = notifyingInvokers.get();
+ if (notifying != null &&
+ currentChain == backupChain &&
+ availableInvokers.getOriginList() == notifying.getOriginList()) {
+ return mainChain;
}
- return currentChain.route(url, availableInvokers, invocation);
+ return currentChain;
}
- private void switchToMainChain() {
- currentChain = mainChain;
+ /**
+ * @deprecated use {@link RouterChain#getSingleChain(URL, BitList,
Invocation)} and {@link SingleRouterChain#route(URL, BitList, Invocation)}
instead
+ */
+ @Deprecated
+ public List<Invoker<T>> route(URL url, BitList<Invoker<T>>
availableInvokers, Invocation invocation) {
+ return getSingleChain(url, availableInvokers, invocation).route(url,
availableInvokers, invocation);
}
/**
@@ -110,56 +127,78 @@ public class RouterChain<T> {
* Notify whenever addresses in registry change.
*/
public synchronized void setInvokers(BitList<Invoker<T>> invokers,
Runnable switchAction) {
- // 1. switch to backup chain
- currentChain = backupChain;
-
- // 2. wait main chain to finish routing
- waitChain(mainChain);
+ try {
+ // Lock to prevent directory continue list
+ lock.writeLock().lock();
+
+ // Switch to back up chain. Will update main chain first.
+ currentChain = backupChain;
+ } finally {
+ // Release lock to minimize the impact for each newly created
invocations as much as possible.
+ // Should not release lock until main chain update finished. Or
this may cause long hang.
+ lock.writeLock().unlock();
+ }
- // 3. notify main chain
- mainChain.setInvokers(invokers);
+ // Refresh main chain.
+ // No one can request to use main chain. `currentChain` is backup
chain. `route` method cannot access main chain.
+ try {
+ // Lock main chain to wait all invocation end
+ // To wait until no one is using main chain.
+ mainChain.getLock().writeLock().lock();
+
+ // refresh
+ mainChain.setInvokers(invokers);
+ } catch (Throwable t) {
+ logger.error(LoggerCodeConstants.INTERNAL_ERROR, "", "", "Error
occurred when refreshing router chain.", t);
+ throw t;
+ } finally {
+ // Unlock main chain
+ mainChain.getLock().writeLock().unlock();
+ }
- // 4. let `route` method check if the invokers are updated
+ // Set the reference of newly invokers to temp variable.
+ // Reason: The next step will switch the invokers reference in
directory, so we should check the `availableInvokers`
+ // argument when `route`. If the current invocation use newly
invokers, we should use main chain to route, and
+ // this can prevent use newly invokers to route backup chain,
which can only route origin invokers now.
notifyingInvokers.set(invokers);
- // 5. switch in directory
+ // Switch the invokers reference in directory.
+ // Cannot switch before update main chain or after backup chain update
success. Or that will cause state inconsistent.
switchAction.run();
- // 6. wait
try {
- Thread.sleep(1);
- } catch (InterruptedException e) {
- logger.error(INTERNAL_INTERRUPTED, "", "", "Interrupted when
waiting router chain.", e);
- Thread.currentThread().interrupt();
+ // Lock to prevent directory continue list
+ // The invokers reference in directory now should be the newly one
and should always use the newly one once lock released.
+ lock.writeLock().lock();
+
+ // Switch to main chain. Will update backup chain later.
+ currentChain = mainChain;
+
+ // Clean up temp variable.
+ // `availableInvokers` check is useless now, because `route`
method will no longer receive any `availableInvokers` related
+ // with the origin invokers. The getter of invokers reference in
directory is locked now, and will return newly invokers
+ // once lock released.
+ notifyingInvokers.set(null);
+ } finally {
+ // Release lock to minimize the impact for each newly created
invocations as much as possible.
+ // Will use newly invokers and main chain now.
+ lock.writeLock().unlock();
}
- // 7. switch chain
- switchToMainChain();
-
- // 8. clean up `route` method wait
- notifyingInvokers.set(null);
-
- // 10. wait backup chain
- waitChain(backupChain);
-
- // 11. notify backup chain
- backupChain.setInvokers(invokers);
- }
-
- private void waitChain(SingleRouterChain<T> oldChain) {
+ // Refresh main chain.
+ // No one can request to use main chain. `currentChain` is main chain.
`route` method cannot access backup chain.
try {
- Thread.sleep(1);
- int waitTime = 0;
- while (oldChain.getCurrentConcurrency() != 0) {
- if (waitTime++ == 1000) {
- logger.warn(REGISTRY_ROUTER_WAIT_LONG, "Wait router to
long", "", "Wait router invoke end exceed 1000ms, router may stuck in.");
- }
- // long time wait
- Thread.sleep(1);
- }
- } catch (InterruptedException t) {
- logger.error(INTERNAL_INTERRUPTED, "Wait router to interrupted",
"", "Wait router to interrupted.");
- Thread.currentThread().interrupt();
+ // Lock main chain to wait all invocation end
+ backupChain.getLock().writeLock().lock();
+
+ // refresh
+ backupChain.setInvokers(invokers);
+ } catch (Throwable t) {
+ logger.error(LoggerCodeConstants.INTERNAL_ERROR, "", "", "Error
occurred when refreshing router chain.", t);
+ throw t;
+ } finally {
+ // Unlock backup chain
+ backupChain.getLock().writeLock().unlock();
}
}
@@ -168,10 +207,9 @@ public class RouterChain<T> {
backupChain.destroy();
// 2. switch
+ lock.writeLock().lock();
currentChain = backupChain;
-
- // 3. wait
- waitChain(mainChain);
+ lock.writeLock().unlock();
// 4. destroy
mainChain.destroy();
diff --git
a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/SingleRouterChain.java
b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/SingleRouterChain.java
index e76aac3c4a..278722eded 100644
---
a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/SingleRouterChain.java
+++
b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/SingleRouterChain.java
@@ -16,6 +16,13 @@
*/
package org.apache.dubbo.rpc.cluster;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.Version;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
@@ -33,12 +40,6 @@ import org.apache.dubbo.rpc.cluster.router.state.BitList;
import org.apache.dubbo.rpc.cluster.router.state.StateRouter;
import org.apache.dubbo.rpc.cluster.router.state.TailStateRouter;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.concurrent.atomic.AtomicInteger;
-
import static
org.apache.dubbo.common.constants.LoggerCodeConstants.CLUSTER_FAILED_STOP;
import static
org.apache.dubbo.common.constants.LoggerCodeConstants.CLUSTER_NO_VALID_PROVIDER;
import static
org.apache.dubbo.common.constants.LoggerCodeConstants.INTERNAL_ERROR;
@@ -76,7 +77,7 @@ public class SingleRouterChain<T> {
private final RouterSnapshotSwitcher routerSnapshotSwitcher;
- private final AtomicInteger currentConcurrency = new AtomicInteger(0);
+ private final ReadWriteLock lock = new ReentrantReadWriteLock();
public SingleRouterChain(List<Router> routers, List<StateRouter<T>>
stateRouters, boolean shouldFailFast, RouterSnapshotSwitcher
routerSnapshotSwitcher) {
initWithRouters(routers);
@@ -132,21 +133,16 @@ public class SingleRouterChain<T> {
}
public List<Invoker<T>> route(URL url, BitList<Invoker<T>>
availableInvokers, Invocation invocation) {
- currentConcurrency.incrementAndGet();
- try {
- if (invokers.getOriginList() != availableInvokers.getOriginList())
{
- logger.error(INTERNAL_ERROR, "", "Router's invoker size: " +
invokers.getOriginList().size() +
- " Invocation's invoker size: " +
availableInvokers.getOriginList().size(),
- "Reject to route, because the invokers has changed.");
- throw new IllegalStateException("reject to route, because the
invokers has changed.");
- }
- if (RpcContext.getServiceContext().isNeedPrintRouterSnapshot()) {
- return routeAndPrint(url, availableInvokers, invocation);
- } else {
- return simpleRoute(url, availableInvokers, invocation);
- }
- } finally {
- currentConcurrency.decrementAndGet();
+ if (invokers.getOriginList() != availableInvokers.getOriginList()) {
+ logger.error(INTERNAL_ERROR, "", "Router's invoker size: " +
invokers.getOriginList().size() +
+ " Invocation's invoker size: " +
availableInvokers.getOriginList().size(),
+ "Reject to route, because the invokers has changed.");
+ throw new IllegalStateException("reject to route, because the
invokers has changed.");
+ }
+ if (RpcContext.getServiceContext().isNeedPrintRouterSnapshot()) {
+ return routeAndPrint(url, availableInvokers, invocation);
+ } else {
+ return simpleRoute(url, availableInvokers, invocation);
}
}
@@ -260,7 +256,7 @@ public class SingleRouterChain<T> {
// 3. set router chain output reverse
RouterSnapshotNode<T> currentNode = commonRouterNode;
- while (currentNode != null){
+ while (currentNode != null) {
RouterSnapshotNode<T> parent = currentNode.getParentNode();
if (parent != null) {
// common router only has one child invoke
@@ -282,7 +278,7 @@ public class SingleRouterChain<T> {
if (routerSnapshotSwitcher.isEnable()) {
routerSnapshotSwitcher.setSnapshot(message);
}
- logger.warn(CLUSTER_NO_VALID_PROVIDER,"No provider available
after route for the service","",message);
+ logger.warn(CLUSTER_NO_VALID_PROVIDER, "No provider available
after route for the service", "", message);
}
} else {
if (logger.isInfoEnabled()) {
@@ -324,8 +320,8 @@ public class SingleRouterChain<T> {
return stateRouters;
}
- public int getCurrentConcurrency() {
- return currentConcurrency.get();
+ public ReadWriteLock getLock() {
+ return lock;
}
public void destroy() {
@@ -334,7 +330,7 @@ public class SingleRouterChain<T> {
try {
router.stop();
} catch (Exception e) {
- logger.error(CLUSTER_FAILED_STOP,"route stop failed","","Error
trying to stop router " + router.getClass(),e);
+ logger.error(CLUSTER_FAILED_STOP, "route stop failed", "",
"Error trying to stop router " + router.getClass(), e);
}
}
routers = Collections.emptyList();
@@ -344,7 +340,7 @@ public class SingleRouterChain<T> {
try {
router.stop();
} catch (Exception e) {
- logger.error(CLUSTER_FAILED_STOP,"StateRouter stop
failed","","Error trying to stop StateRouter " + router.getClass(),e);
+ logger.error(CLUSTER_FAILED_STOP, "StateRouter stop failed",
"", "Error trying to stop StateRouter " + router.getClass(), e);
}
}
stateRouters = Collections.emptyList();
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 0fca01db7b..eb69309f24 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
@@ -16,10 +16,26 @@
*/
package org.apache.dubbo.rpc.cluster.directory;
+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.stream.Collectors;
+
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.LoggerCodeConstants;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.threadpool.manager.FrameworkExecutorRepository;
@@ -33,25 +49,11 @@ 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.SingleRouterChain;
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.stream.Collectors;
-
import static org.apache.dubbo.common.constants.CommonConstants.CONSUMER;
import static
org.apache.dubbo.common.constants.CommonConstants.DEFAULT_RECONNECT_TASK_PERIOD;
import static
org.apache.dubbo.common.constants.CommonConstants.DEFAULT_RECONNECT_TASK_TRY_COUNT;
@@ -186,27 +188,48 @@ public abstract class AbstractDirectory<T> implements
Directory<T> {
}
BitList<Invoker<T>> availableInvokers;
- // use clone to avoid being modified at doList().
- if (invokersInitialized) {
- availableInvokers = validInvokers.clone();
- } else {
- availableInvokers = invokers.clone();
- }
+ SingleRouterChain<T> singleChain = null;
+ try {
+ try {
+ if (routerChain != null) {
+ routerChain.getLock().readLock().lock();
+ }
+ // use clone to avoid being modified at doList().
+ if (invokersInitialized) {
+ availableInvokers = validInvokers.clone();
+ } else {
+ availableInvokers = invokers.clone();
+ }
+
+ if (routerChain != null) {
+ singleChain = routerChain.getSingleChain(getConsumerUrl(),
availableInvokers, invocation);
+ singleChain.getLock().readLock().lock();
+ }
+ } finally {
+ if (routerChain != null) {
+ routerChain.getLock().readLock().unlock();
+ }
+ }
- List<Invoker<T>> routedResult = doList(availableInvokers, invocation);
- if (routedResult.isEmpty()) {
- // 2-2 - No provider available.
-
- logger.warn(CLUSTER_NO_VALID_PROVIDER, "provider server or
registry center crashed", "",
- "No provider available after connectivity filter for the
service " + getConsumerUrl().getServiceKey()
- + " All validInvokers' size: " + validInvokers.size()
- + " All routed invokers' size: " + routedResult.size()
- + " All invokers' size: " + invokers.size()
- + " from registry " + getUrl().getAddress()
- + " on the consumer " + NetUtils.getLocalHost()
- + " using the dubbo version " + Version.getVersion() + ".");
+ List<Invoker<T>> routedResult = doList(singleChain,
availableInvokers, invocation);
+ if (routedResult.isEmpty()) {
+ // 2-2 - No provider available.
+
+ logger.warn(CLUSTER_NO_VALID_PROVIDER, "provider server or
registry center crashed", "",
+ "No provider available after connectivity filter for the
service " + getConsumerUrl().getServiceKey()
+ + " All validInvokers' size: " + validInvokers.size()
+ + " All routed invokers' size: " + routedResult.size()
+ + " All invokers' size: " + invokers.size()
+ + " from registry " + getUrl().getAddress()
+ + " on the consumer " + NetUtils.getLocalHost()
+ + " using the dubbo version " + Version.getVersion() +
".");
+ }
+ return Collections.unmodifiableList(routedResult);
+ } finally {
+ if (singleChain != null) {
+ singleChain.getLock().readLock().unlock();
+ }
}
- return Collections.unmodifiableList(routedResult);
}
@Override
@@ -377,6 +400,21 @@ public abstract class AbstractDirectory<T> implements
Directory<T> {
}
}
+ protected final void refreshRouter(BitList<Invoker<T>> newlyInvokers,
Runnable switchAction) {
+ try {
+ routerChain.setInvokers(newlyInvokers.clone(), switchAction);
+ } catch (Throwable t) {
+ logger.error(LoggerCodeConstants.INTERNAL_ERROR, "", "", "Error
occurred when refreshing router chain. " +
+ "The addresses from notification: " +
+ newlyInvokers.stream()
+ .map(Invoker::getUrl)
+ .map(URL::getAddress)
+ .collect(Collectors.joining(", ")), t);
+
+ throw t;
+ }
+ }
+
/**
* for ut only
*/
@@ -436,7 +474,8 @@ public abstract class AbstractDirectory<T> implements
Directory<T> {
}
}
- protected abstract List<Invoker<T>> doList(BitList<Invoker<T>> invokers,
Invocation invocation) throws RpcException;
+ protected abstract List<Invoker<T>> doList(SingleRouterChain<T>
singleRouterChain,
+ BitList<Invoker<T>> invokers,
Invocation invocation) throws RpcException;
protected String joinValidInvokerAddresses() {
BitList<Invoker<T>> validInvokers = getValidInvokers().clone();
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 2a87e13d31..d8a022f3d2 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
@@ -16,6 +16,8 @@
*/
package org.apache.dubbo.rpc.cluster.directory;
+import java.util.List;
+
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
@@ -24,10 +26,9 @@ 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.SingleRouterChain;
import org.apache.dubbo.rpc.cluster.router.state.BitList;
-import java.util.List;
-
import static
org.apache.dubbo.common.constants.LoggerCodeConstants.CLUSTER_FAILED_SITE_SELECTION;
/**
@@ -92,27 +93,28 @@ public class StaticDirectory<T> extends
AbstractDirectory<T> {
public void buildRouterChain() {
RouterChain<T> routerChain = RouterChain.buildChain(getInterface(),
getUrl());
- routerChain.setInvokers(getInvokers(), ()->{});
+ routerChain.setInvokers(getInvokers(), () -> {
+ });
this.setRouterChain(routerChain);
}
public void notify(List<Invoker<T>> invokers) {
BitList<Invoker<T>> bitList = new BitList<>(invokers);
if (routerChain != null) {
- routerChain.setInvokers(bitList.clone(), () ->
this.setInvokers(bitList));
+ refreshRouter(bitList.clone(), () -> this.setInvokers(bitList));
} else {
this.setInvokers(bitList);
}
}
@Override
- protected List<Invoker<T>> doList(BitList<Invoker<T>> invokers, Invocation
invocation) throws RpcException {
- if (routerChain != null) {
+ protected List<Invoker<T>> doList(SingleRouterChain<T> singleRouterChain,
BitList<Invoker<T>> invokers, Invocation invocation) throws RpcException {
+ if (singleRouterChain != null) {
try {
- List<Invoker<T>> finalInvokers =
routerChain.route(getConsumerUrl(), invokers, invocation);
+ List<Invoker<T>> finalInvokers =
singleRouterChain.route(getConsumerUrl(), invokers, invocation);
return finalInvokers == null ? BitList.emptyList() :
finalInvokers;
} catch (Throwable t) {
- logger.error(CLUSTER_FAILED_SITE_SELECTION,"Failed to execute
router","","Failed to execute router: " + getUrl() + ", cause: " +
t.getMessage(),t);
+ logger.error(CLUSTER_FAILED_SITE_SELECTION, "Failed to execute
router", "", "Failed to execute router: " + getUrl() + ", cause: " +
t.getMessage(), t);
return BitList.emptyList();
}
}
diff --git
a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/RouterChainTest.java
b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/RouterChainTest.java
index b5edc67ed5..19409b6602 100644
---
a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/RouterChainTest.java
+++
b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/RouterChainTest.java
@@ -17,6 +17,12 @@
package org.apache.dubbo.rpc.cluster;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.config.configcenter.ConfigChangeType;
import org.apache.dubbo.common.config.configcenter.ConfigChangedEvent;
@@ -35,17 +41,10 @@ import
org.apache.dubbo.rpc.cluster.router.mesh.route.MeshAppRuleListener;
import org.apache.dubbo.rpc.cluster.router.mesh.route.MeshRuleManager;
import org.apache.dubbo.rpc.cluster.router.state.BitList;
import org.apache.dubbo.rpc.model.FrameworkModel;
-
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-
import static org.apache.dubbo.common.constants.CommonConstants.INTERFACE_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.TAG_KEY;
import static
org.apache.dubbo.rpc.cluster.router.mesh.route.MeshRuleConstants.MESH_RULE_DATA_ID_SUFFIX;
@@ -153,7 +152,8 @@ class RouterChainTest {
RouterSnapshotSwitcher routerSnapshotSwitcher =
FrameworkModel.defaultModel().getBeanFactory().getBean(RouterSnapshotSwitcher.class);
routerSnapshotSwitcher.addEnabledService("org.apache.dubbo.demo.DemoService");
// route
- List<Invoker<DemoService>> result = routerChain.route(consumerUrl,
invokers, rpcInvocation);
+ List<Invoker<DemoService>> result =
routerChain.getSingleChain(consumerUrl, invokers, rpcInvocation)
+ .route(consumerUrl, invokers, rpcInvocation);
Assertions.assertEquals(result.size(), 1);
Assertions.assertTrue(result.contains(invoker5));
@@ -170,7 +170,8 @@ class RouterChainTest {
Assertions.assertTrue(snapshot[0].contains(snapshotLog));
RpcContext.getServiceContext().setNeedPrintRouterSnapshot(false);
- result = routerChain.route(consumerUrl, invokers, rpcInvocation);
+ result = routerChain.getSingleChain(consumerUrl, invokers,
rpcInvocation)
+ .route(consumerUrl, invokers, rpcInvocation);
Assertions.assertEquals(result.size(), 1);
Assertions.assertTrue(result.contains(invoker5));
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 b5ae59fad6..4e4b1e7b63 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
@@ -16,20 +16,19 @@
*/
package org.apache.dubbo.rpc.cluster.loadbalance;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
+
import org.apache.dubbo.common.URL;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.cluster.LoadBalance;
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;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.atomic.AtomicLong;
-
@SuppressWarnings("rawtypes")
class ConsistentHashLoadBalanceTest extends LoadBalanceBaseTest {
@@ -94,7 +93,8 @@ class ConsistentHashLoadBalanceTest extends
LoadBalanceBaseTest {
for (int i = 0; i < 100; i++) {
routerChain.setInvokers(new BitList<>(invokers), () -> {});
- List<Invoker<LoadBalanceBaseTest>> routeInvokers =
routerChain.route(url, new BitList<>(invokers), invocation);
+ List<Invoker<LoadBalanceBaseTest>> routeInvokers =
routerChain.getSingleChain(url, new BitList<>(invokers), invocation)
+ .route(url, new BitList<>(invokers), invocation);
Invoker<LoadBalanceBaseTest> finalInvoker =
lb.select(routeInvokers, url, invocation);
Assertions.assertEquals(result, finalInvoker);
}
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 c9b0e99178..736d5a71d2 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
@@ -16,6 +16,10 @@
*/
package org.apache.dubbo.rpc.cluster.support;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Callable;
+
import org.apache.dubbo.common.URL;
import org.apache.dubbo.rpc.AppResponse;
import org.apache.dubbo.rpc.AsyncRpcResult;
@@ -26,17 +30,13 @@ 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;
+import org.apache.dubbo.rpc.cluster.SingleRouterChain;
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;
import org.junit.jupiter.api.Test;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.Callable;
-
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertSame;
@@ -360,8 +360,8 @@ class FailoverClusterInvokerTest {
}
@Override
- protected List<Invoker<T>> doList(BitList<Invoker<T>> invokers,
Invocation invocation) throws RpcException {
- return super.doList(invokers, invocation);
+ protected List<Invoker<T>> doList(SingleRouterChain<T>
singleRouterChain, BitList<Invoker<T>> invokers, Invocation invocation) throws
RpcException {
+ return super.doList(singleRouterChain, invokers, invocation);
}
}
}
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 e93d090873..8cc86720cc 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
@@ -16,6 +16,18 @@
*/
package org.apache.dubbo.registry.client;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+
import org.apache.dubbo.common.ProtocolServiceKey;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.config.configcenter.DynamicConfiguration;
@@ -46,18 +58,6 @@ import
org.apache.dubbo.rpc.cluster.directory.StaticDirectory;
import org.apache.dubbo.rpc.cluster.router.state.BitList;
import org.apache.dubbo.rpc.model.ModuleModel;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.stream.Collectors;
-
import static org.apache.dubbo.common.constants.CommonConstants.DISABLED_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.ENABLED_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.PROTOCOL_KEY;
@@ -232,9 +232,9 @@ public class ServiceDiscoveryRegistryDirectory<T> extends
DynamicDirectory<T> {
if (invokerUrls.size() == 1 &&
EMPTY_PROTOCOL.equals(invokerUrls.get(0).getProtocol())) {
logger.warn(PROTOCOL_UNSUPPORTED, "", "", "Received url with EMPTY
protocol, will clear all available addresses.");
- routerChain.setInvokers(BitList.emptyList(), () -> {
- this.forbidden = true; // Forbid to access
- });
+ refreshRouter(BitList.emptyList(), () ->
+ this.forbidden = true // Forbid to access
+ );
destroyAllInvokers(); // Close all invokers
} else {
this.forbidden = false; // Allow accessing
@@ -262,7 +262,7 @@ public class ServiceDiscoveryRegistryDirectory<T> extends
DynamicDirectory<T> {
List<Invoker<T>> newInvokers = Collections.unmodifiableList(new
ArrayList<>(newUrlInvokerMap.values()));
BitList<Invoker<T>> finalInvokers = multiGroup ? new
BitList<>(toMergeInvokerList(newInvokers)) : new BitList<>(newInvokers);
// pre-route and build cache
- routerChain.setInvokers(finalInvokers.clone(), () ->
this.setInvokers(finalInvokers));
+ refreshRouter(finalInvokers.clone(), () ->
this.setInvokers(finalInvokers));
this.urlInvokerMap = newUrlInvokerMap;
if (oldUrlInvokerMap != 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 167cdd2105..12f2dcef34 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
@@ -16,6 +16,8 @@
*/
package org.apache.dubbo.registry.integration;
+import java.util.List;
+
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.Version;
import org.apache.dubbo.common.config.ConfigurationUtils;
@@ -39,16 +41,15 @@ import org.apache.dubbo.rpc.cluster.Configurator;
import org.apache.dubbo.rpc.cluster.Constants;
import org.apache.dubbo.rpc.cluster.RouterChain;
import org.apache.dubbo.rpc.cluster.RouterFactory;
+import org.apache.dubbo.rpc.cluster.SingleRouterChain;
import org.apache.dubbo.rpc.cluster.directory.AbstractDirectory;
import org.apache.dubbo.rpc.cluster.router.state.BitList;
import org.apache.dubbo.rpc.model.ModuleModel;
-import java.util.List;
-
import static org.apache.dubbo.common.constants.CommonConstants.ANY_VALUE;
+import static
org.apache.dubbo.common.constants.LoggerCodeConstants.CLUSTER_FAILED_SITE_SELECTION;
import static
org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_FAILED_DESTROY_SERVICE;
import static
org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_FAILED_DESTROY_UNREGISTER_URL;
-import static
org.apache.dubbo.common.constants.LoggerCodeConstants.CLUSTER_FAILED_SITE_SELECTION;
import static org.apache.dubbo.common.constants.RegistryConstants.CATEGORY_KEY;
import static
org.apache.dubbo.common.constants.RegistryConstants.CONSUMERS_CATEGORY;
import static org.apache.dubbo.registry.Constants.REGISTER_KEY;
@@ -189,7 +190,8 @@ public abstract class DynamicDirectory<T> extends
AbstractDirectory<T> implement
}
@Override
- public List<Invoker<T>> doList(BitList<Invoker<T>> invokers, Invocation
invocation) {
+ public List<Invoker<T>> doList(SingleRouterChain<T> singleRouterChain,
+ BitList<Invoker<T>> invokers, Invocation
invocation) {
if (forbidden && shouldFailFast) {
// 1. No service provider 2. Service providers are disabled
throw new RpcException(RpcException.FORBIDDEN_EXCEPTION, "No
provider available from registry " +
@@ -204,7 +206,7 @@ public abstract class DynamicDirectory<T> extends
AbstractDirectory<T> implement
try {
// Get invokers from cache, only runtime routers will be executed.
- List<Invoker<T>> result = routerChain.route(getConsumerUrl(),
invokers, invocation);
+ List<Invoker<T>> result =
singleRouterChain.route(getConsumerUrl(), invokers, invocation);
return result == null ? BitList.emptyList() : result;
} catch (Throwable t) {
// 2-1 - Failed to execute routing.
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 2f371b0701..a994f1f1c7 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
@@ -16,6 +16,19 @@
*/
package org.apache.dubbo.registry.integration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.URLBuilder;
import org.apache.dubbo.common.config.configcenter.DynamicConfiguration;
@@ -42,19 +55,6 @@ import org.apache.dubbo.rpc.cluster.router.state.BitList;
import org.apache.dubbo.rpc.cluster.support.ClusterUtils;
import org.apache.dubbo.rpc.model.ModuleModel;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Optional;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.stream.Collectors;
-
import static
org.apache.dubbo.common.constants.CommonConstants.APPLICATION_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.DISABLED_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.DUBBO_PROTOCOL;
@@ -220,9 +220,9 @@ public class RegistryDirectory<T> extends
DynamicDirectory<T> {
if (invokerUrls.size() == 1
&& invokerUrls.get(0) != null
&& EMPTY_PROTOCOL.equals(invokerUrls.get(0).getProtocol())) {
- routerChain.setInvokers(BitList.emptyList(), () -> {
- this.forbidden = true; // Forbid to access
- });
+ refreshRouter(BitList.emptyList(), () ->
+ this.forbidden = true // Forbid to access
+ );
destroyAllInvokers(); // Close all invokers
} else {
this.forbidden = false; // Allow to access
@@ -285,7 +285,7 @@ public class RegistryDirectory<T> extends
DynamicDirectory<T> {
List<Invoker<T>> newInvokers = Collections.unmodifiableList(new
ArrayList<>(newUrlInvokerMap.values()));
BitList<Invoker<T>> finalInvokers = multiGroup ? new
BitList<>(toMergeInvokerList(newInvokers)) : new BitList<>(newInvokers);
// pre-route and build cache
- routerChain.setInvokers(finalInvokers.clone(), () ->
this.setInvokers(finalInvokers));
+ refreshRouter(finalInvokers.clone(), () ->
this.setInvokers(finalInvokers));
this.urlInvokerMap = newUrlInvokerMap;
try {
