natas_hw
/
dubbo
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

[3.0] Feature/state router (#6844) 

* state router

* fix router chain

* fix router chain

* fix comment

* fix comment

* format files

* deal dependecy

* fix comment

* fix comment

* fix

* fix ut

* fix ut

* fix

* fix comment

* add code comments

* add code comments

* add code comments

* add code comments
This commit is contained in:
panxiaojun233 2021-05-18 17:13:50 +08:00 committed by GitHub
parent 7c8b658e40
commit 3531a7d681
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
20 changed files with 1370 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
dubbo-cluster/pom.xml
View File

@ -31,6 +31,10 @@
<skip_maven_deploy>false</skip_maven_deploy>
</properties>
<dependencies>
<dependency>
<groupId>org.roaringbitmap</groupId>
<artifactId>RoaringBitmap</artifactId>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-rpc-api</artifactId>

208
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/RouterChain.java
View File

@ -16,19 +16,32 @@
*/
package org.apache.dubbo.rpc.cluster;
import org.apache.dubbo.common.URL;
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.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
import org.apache.dubbo.common.URL;
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.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.RpcException;
import org.apache.dubbo.rpc.cluster.directory.StaticDirectory;
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;
/**
* Router chain
*/
@ -45,19 +58,50 @@ public class RouterChain<T> {
// instance will never delete or recreate.
private List<Router> builtinRouters = Collections.emptyList();
private List<StateRouter> builtinStateRouters = Collections.emptyList();
private List<StateRouter> stateRouters = Collections.emptyList();
private final ExecutorRepository executorRepository = ExtensionLoader.getExtensionLoader(ExecutorRepository.class)
.getDefaultExtension();
protected URL url;
protected AtomicReference<AddrCache<T>> cache = new AtomicReference<>();
private final Semaphore loopPermit = new Semaphore(1);
private final Semaphore loopPermitNotify = new Semaphore(1);
private final ExecutorService loopPool;
private boolean firstBuildCache = true;
private static final Logger logger = LoggerFactory.getLogger(StaticDirectory.class);
public static <T> RouterChain<T> buildChain(URL url) {
return new RouterChain<>(url);
}
private RouterChain(URL url) {
loopPool = executorRepository.nextExecutorExecutor();
List<RouterFactory> extensionFactories = ExtensionLoader.getExtensionLoader(RouterFactory.class)
.getActivateExtension(url, "router");
.getActivateExtension(url, "router");
List<Router> routers = extensionFactories.stream()
.map(factory -> factory.getRouter(url))
.collect(Collectors.toList());
.map(factory -> factory.getRouter(url))
.collect(Collectors.toList());
initWithRouters(routers);
List<StateRouterFactory> extensionStateRouterFactories = ExtensionLoader.getExtensionLoader(
StateRouterFactory.class)
.getActivateExtension(url, "stateRouter");
List<StateRouter> stateRouters = extensionStateRouterFactories.stream()
.map(factory -> factory.getRouter(url, this))
.sorted(StateRouter::compareTo)
.collect(Collectors.toList());
// init state routers
initWithStateRouters(stateRouters);
}
/**
@ -70,6 +114,11 @@ public class RouterChain<T> {
this.sort();
}
public void initWithStateRouters(List<StateRouter> builtinRouters) {
this.builtinStateRouters = builtinRouters;
this.stateRouters = new ArrayList<>(builtinRouters);
}
/**
* If we use route:// protocol in version before 2.7.0, each URL will generate a Router instance, so we should
* keep the routers up to date, that is, each time router URLs changes, we should update the routers list, only
@ -86,6 +135,14 @@ public class RouterChain<T> {
this.routers = newRouters;
}
public void addStateRouters(List<StateRouter> stateRouters) {
List<StateRouter> newStateRouters = new ArrayList<>();
newStateRouters.addAll(builtinStateRouters);
newStateRouters.addAll(stateRouters);
CollectionUtils.sort(newStateRouters);
this.stateRouters = newStateRouters;
}
public List<Router> getRouters() {
return routers;
}
@ -95,13 +152,35 @@ public class RouterChain<T> {
}
/**
*
* @param url
* @param invocation
* @return
*/
public List<Invoker<T>> route(URL url, Invocation invocation) {
List<Invoker<T>> finalInvokers = invokers;
AddrCache<T> cache = this.cache.get();
if (cache == null) {
throw new RpcException(RpcException.ROUTER_CACHE_NOT_BUILD, "Failed to invoke the method "
+ invocation.getMethodName() + " in the service " + url.getServiceInterface()
+ ". address cache not build "
+ " on the consumer " + NetUtils.getLocalHost()
+ " using the dubbo version " + Version.getVersion()
+ ".");
}
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);
}
}
List<Invoker<T>> finalInvokers = new ArrayList<>(finalBitListInvokers.size());
for(Invoker<T> invoker: finalBitListInvokers) {
finalInvokers.add(invoker);
}
for (Router router : routers) {
finalInvokers = router.route(finalInvokers, url, invocation);
}
@ -114,7 +193,99 @@ public class RouterChain<T> {
*/
public void setInvokers(List<Invoker<T>> invokers) {
this.invokers = (invokers == null ? Collections.emptyList() : invokers);
stateRouters.forEach(router -> router.notify(this.invokers));
routers.forEach(router -> router.notify(this.invokers));
loop(true);
}
/**
* Build the asynchronous address cache for stateRouter.
* @param notify Whether the addresses in registry has changed.
*/
private void buildCache(boolean notify) {
if (invokers == null || invokers.size() <= 0) {
return;
}
AddrCache<T> origin = cache.get();
List<Invoker<T>> copyInvokers = new ArrayList<>(this.invokers);
AddrCache<T> newCache = new AddrCache<T>();
newCache.setInvokers(invokers);
for (StateRouter stateRouter : stateRouters) {
RouterCache routerCache;
try {
routerCache = poolRouter(stateRouter, origin, copyInvokers, notify);
//file cache
newCache.getCache().put(stateRouter.getName(), routerCache);
} catch (Throwable t) {
logger.error("Failed to pool router: " + stateRouter.getUrl() + ", cause: " + t.getMessage(), t);
return;
}
}
this.cache.set(newCache);
}
/**
* Cache the address list for each StateRouter.
* @param router router
* @param orign The original address cache
* @param invokers The full address list
* @param notify Whether the addresses in registry has changed.
* @return
*/
private RouterCache poolRouter(StateRouter router, AddrCache<T> orign, List<Invoker<T>> invokers, boolean notify) {
String routerName = router.getName();
RouterCache routerCache;
if (isCacheMiss(orign, routerName) || router.shouldRePool() || notify) {
return router.pool(invokers);
} else {
routerCache = orign.getCache().get(routerName);
}
if (routerCache == null) {
return new RouterCache();
}
return routerCache;
}
private boolean isCacheMiss(AddrCache<T> cache, String routerName) {
return cache == null || cache.getCache() == null || cache.getInvokers() == null || cache.getCache().get(
routerName)
== null;
}
/***
* Build the asynchronous address cache for stateRouter.
* @param notify Whether the addresses in registry has changed.
*/
public void loop(boolean notify) {
if (firstBuildCache) {
firstBuildCache = false;
buildCache(notify);
}
if (notify) {
if (loopPermitNotify.tryAcquire()) {
loopPool.submit(new NotifyLoopRunnable(true));
}
} else {
if (loopPermit.tryAcquire()) {
loopPool.submit(new NotifyLoopRunnable(false));
}
}
}
class NotifyLoopRunnable implements Runnable {
private final boolean notify;
public NotifyLoopRunnable(boolean notify) {
this.notify = notify;
}
@Override
public void run() {
loopPermitNotify.release();
buildCache(notify);
}
}
public void destroy() {
@ -128,5 +299,16 @@ public class RouterChain<T> {
}
routers = Collections.emptyList();
builtinRouters = Collections.emptyList();
for (StateRouter router : stateRouters) {
try {
router.stop();
} catch (Exception e) {
LOGGER.error("Error trying to stop stateRouter " + router.getClass(), e);
}
}
stateRouters = Collections.emptyList();
builtinStateRouters = Collections.emptyList();
}
}

1
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/directory/StaticDirectory.java
View File

@ -94,6 +94,7 @@ public class StaticDirectory<T> extends AbstractDirectory<T> {
public void buildRouterChain() {
RouterChain<T> routerChain = RouterChain.buildChain(getUrl());
routerChain.setInvokers(invokers);
routerChain.loop(true);
this.setRouterChain(routerChain);
}

119
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/AbstractStateRouter.java Normal file
View File

@ -0,0 +1,119 @@
/*
* 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;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.extension.ExtensionLoader;
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.governance.GovernanceRuleRepository;
/***
* The abstract class of StateRoute.
* @since 3.0
*/
public abstract class AbstractStateRouter implements StateRouter {
final protected RouterChain chain;
protected int priority = DEFAULT_PRIORITY;
protected boolean force = false;
protected URL url;
protected List<Invoker> invokers;
protected AtomicReference<AddrCache> cache;
protected GovernanceRuleRepository ruleRepository;
public AbstractStateRouter(URL url, RouterChain chain) {
this.ruleRepository = ExtensionLoader.getExtensionLoader(GovernanceRuleRepository.class).getDefaultExtension();
this.chain = chain;
this.url = url;
}
@Override
public <T> void notify(List<Invoker<T>> invokers) {
this.invokers = (List)invokers;
}
@Override
public URL getUrl() {
return url;
}
public void setUrl(URL url) {
this.url = url;
}
@Override
public boolean isRuntime() {
return true;
}
@Override
public boolean isForce() {
return force;
}
public void setForce(boolean force) {
this.force = force;
}
@Override
public int getPriority() {
return priority;
}
public void setPriority(int priority) {
this.priority = priority;
}
@Override
public <T> BitList<Invoker<T>> route(BitList<Invoker<T>> invokers, RouterCache<T> cache, URL url,
Invocation invocation) throws RpcException {
List<String> tags = getTags(url, invocation);
if (tags == null) {
return invokers;
}
for (String tag : tags) {
BitList<Invoker<T>> tagInvokers = cache.getAddrPool().get(tag);
if (tagMatchFail(tagInvokers)) {
continue;
}
return tagInvokers.intersect(invokers, invokers.getUnmodifiableList());
}
return invokers;
}
protected List<String> getTags(URL url, Invocation invocation) {
return new ArrayList<String>();
}
public <T> Boolean tagMatchFail(BitList<Invoker<T>> invokers) {
return invokers == null || invokers.isEmpty();
}
@Override
public void pool() {
chain.loop(false);
}
}

52
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/AddrCache.java Normal file
View File

@ -0,0 +1,52 @@
/*
* 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;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.apache.dubbo.rpc.Invoker;
/***
* Address cache,
* used to cache the results of the StaterRouter's asynchronous address list calculations.
* @param <T>
* @since 3.0
*/
public class AddrCache<T> {
private final static ConcurrentHashMap EMPTY_MAP = new ConcurrentHashMap<>();
protected List<Invoker<T>> invokers;
protected ConcurrentMap<String, RouterCache<T>> cache = EMPTY_MAP;
public List<Invoker<T>> getInvokers() {
return invokers;
}
public void setInvokers(List<Invoker<T>> invokers) {
this.invokers = invokers;
}
public ConcurrentMap<String, RouterCache<T>> getCache() {
return cache;
}
public void setCache(ConcurrentHashMap<String, RouterCache<T>> cache) {
this.cache = cache;
}
}

232
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/BitList.java Normal file
View File

@ -0,0 +1,232 @@
/*
* 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;
import java.util.Arrays;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.ListIterator;
import org.roaringbitmap.IntIterator;
import org.roaringbitmap.RoaringBitmap;
/**
* BitList based on BitMap implementation.
* @param <E>
* @since 3.0
*/
public class BitList<E> implements List<E> {
private final RoaringBitmap rootMap;
private final List<E> unmodifiableList;
public BitList(List<E> unmodifiableList, boolean empty) {
this.unmodifiableList = unmodifiableList;
this.rootMap = new RoaringBitmap();
if (!empty) {
this.rootMap.add(0L, unmodifiableList.size());
}
}
public BitList(List<E> unmodifiableList, RoaringBitmap rootMap) {
this.unmodifiableList = unmodifiableList;
this.rootMap = rootMap;
}
public BitList(List<E> unmodifiableList) {
this(unmodifiableList, false);
}
public List<E> getUnmodifiableList() {
return unmodifiableList;
}
public void addIndex(int index) {
this.rootMap.add(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);
}
@Override
public int size() {
return rootMap.getCardinality();
}
@Override
public boolean isEmpty() {
return rootMap.isEmpty();
}
@Override
public boolean contains(Object o) {
int idx = unmodifiableList.indexOf(o);
return idx >= 0 && rootMap.contains(idx);
}
@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);
}
};
}
@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;
}
@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;
}
@Override
public boolean add(E e) {
throw new UnsupportedOperationException();
}
@Override
public boolean remove(Object o) {
int idx = unmodifiableList.indexOf(o);
if (idx > -1) {
rootMap.remove(idx);
}
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);
}
return true;
}
@Override
public boolean retainAll(Collection<?> c) {
return false;
}
@Override
public void clear() {
rootMap.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();
}
@Override
public E remove(int index) {
rootMap.remove(index);
return null;
}
@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;
}
st++;
}
return -1;
}
@Override
public int lastIndexOf(Object o) {
throw new UnsupportedOperationException();
}
@Override
public ListIterator<E> listIterator() {
throw new UnsupportedOperationException();
}
@Override
public ListIterator<E> listIterator(int index) {
throw new UnsupportedOperationException();
}
@Override
public List<E> subList(int fromIndex, int toIndex) {
throw new UnsupportedOperationException();
}
}

38
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/CacheableStateRouterFactory.java Normal file
View File

@ -0,0 +1,38 @@
/*
* 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;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.rpc.cluster.RouterChain;
/**
* If you want to provide a router implementation based on design of v2.7.0, please extend from this abstract class.
* For 2.6.x style router, please implement and use RouterFactory directly.
*/
public abstract class CacheableStateRouterFactory implements StateRouterFactory {
private final ConcurrentMap<String, StateRouter> routerMap = new ConcurrentHashMap<>();
@Override
public StateRouter getRouter(URL url, RouterChain chain) {
return routerMap.computeIfAbsent(url.getServiceKey(), k -> createRouter(url, chain));
}
protected abstract StateRouter createRouter(URL url, RouterChain chain);
}

49
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/RouterCache.java Normal file
View File

@ -0,0 +1,49 @@
/*
* 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;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.apache.dubbo.rpc.Invoker;
/***
* Cache the address list for each Router.
* @param <T>
* @since 3.0
*/
public class RouterCache<T> {
private final static ConcurrentHashMap EMPTY_MAP = new ConcurrentHashMap<>();
protected ConcurrentMap<String, BitList<Invoker<T>>> addrPool = EMPTY_MAP;
protected Object addrMetadata;
public ConcurrentMap<String, BitList<Invoker<T>>> getAddrPool() {
return addrPool;
}
public void setAddrPool(ConcurrentHashMap<String, BitList<Invoker<T>>> addrPool) {
this.addrPool = addrPool;
}
public Object getAddrMetadata() {
return addrMetadata;
}
public void setAddrMetadata(Object addrMetadata) {
this.addrMetadata = addrMetadata;
}
}

100
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/StateRouter.java Normal file
View File

@ -0,0 +1,100 @@
/*
* 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;
import java.util.List;
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.Directory;
/**
* State Router. (SPI, Prototype, ThreadSafe)
* <p>
* <a href="http://en.wikipedia.org/wiki/Routing">Routing</a>
*
* @see org.apache.dubbo.rpc.cluster.Cluster#join(Directory)
* @see Directory#list(Invocation)
* @since 3.0
*/
public interface StateRouter extends Comparable<StateRouter> {
int DEFAULT_PRIORITY = Integer.MAX_VALUE;
/**
* Get the router url.
*
* @return url
*/
URL getUrl();
/***
* 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 <T>
* @return routed invokers
* @throws RpcException
* @Since 3.0
*/
<T> BitList<Invoker<T>> route(BitList<Invoker<T>> invokers, RouterCache<T> cache, URL url, Invocation invocation)
throws
RpcException;
default <T> void notify(List<Invoker<T>> invokers) {
}
/**
* To decide whether this router need to execute every time an RPC comes or should only execute when addresses or
* rule change.
*
* @return true if the router need to execute every time.
*/
boolean isRuntime();
boolean isEnable();
boolean isForce();
int getPriority();
@Override
default int compareTo(StateRouter o) {
if (o == null) {
throw new IllegalArgumentException();
}
return Integer.compare(this.getPriority(), o.getPriority());
}
String getName();
boolean shouldRePool();
<T> RouterCache<T> pool(List<Invoker<T>> invokers);
void pool();
default void stop() {
//do nothing by default
}
}

35
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/StateRouterFactory.java Normal file
View File

@ -0,0 +1,35 @@
/*
* 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;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.extension.Adaptive;
import org.apache.dubbo.common.extension.SPI;
import org.apache.dubbo.rpc.cluster.RouterChain;
@SPI
public interface StateRouterFactory {
/**
* Create state router.
*
* @param url url
* @return router instance
* @since 3.0
*/
@Adaptive("protocol")
<T> StateRouter getRouter(URL url, RouterChain<T> chain);
}

267
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/tag/TagDynamicStateRouter.java Normal file
View File

@ -0,0 +1,267 @@
/*
* 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.tag;
import java.net.UnknownHostException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.config.configcenter.ConfigChangeType;
import org.apache.dubbo.common.config.configcenter.ConfigChangedEvent;
import org.apache.dubbo.common.config.configcenter.ConfigurationListener;
import org.apache.dubbo.common.config.configcenter.DynamicConfiguration;
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.common.utils.StringUtils;
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.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.tag.model.TagRouterRule;
import org.apache.dubbo.rpc.cluster.router.tag.model.TagRuleParser;
import static org.apache.dubbo.common.constants.CommonConstants.ANYHOST_VALUE;
import static org.apache.dubbo.common.constants.CommonConstants.TAG_KEY;
import static org.apache.dubbo.rpc.Constants.FORCE_USE_TAG;
/**
* TagDynamicStateRouter, "application.tag-router"
*/
public class TagDynamicStateRouter extends AbstractStateRouter implements ConfigurationListener {
public static final String NAME = "TAG_ROUTER";
private static final int TAG_ROUTER_DEFAULT_PRIORITY = 100;
private static final Logger logger = LoggerFactory.getLogger(TagDynamicStateRouter.class);
private static final String RULE_SUFFIX = ".tag-router";
private static final String NO_TAG = "noTag";
private TagRouterRule tagRouterRule;
private String application;
public TagDynamicStateRouter(URL url, RouterChain chain) {
super(url, chain);
this.priority = TAG_ROUTER_DEFAULT_PRIORITY;
}
@Override
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());
}
try {
if (event.getChangeType().equals(ConfigChangeType.DELETED)) {
this.tagRouterRule = null;
} else {
this.tagRouterRule = TagRuleParser.parse(event.getContent());
}
} 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);
}
}
@Override
public URL getUrl() {
return url;
}
@Override
public <T> BitList<Invoker<T>> route(BitList<Invoker<T>> invokers, RouterCache<T> cache, URL url,
Invocation invocation) throws RpcException {
final TagRouterRule tagRouterRuleCopy = (TagRouterRule)cache.getAddrMetadata();
String tag = StringUtils.isEmpty(invocation.getAttachment(TAG_KEY)) ? url.getParameter(TAG_KEY) :
invocation.getAttachment(TAG_KEY);
ConcurrentMap<String, BitList<Invoker<T>>> addrPool = cache.getAddrPool();
if (StringUtils.isEmpty(tag)) {
return invokers.intersect(addrPool.get(NO_TAG), invokers.getUnmodifiableList());
} else {
BitList<Invoker<T>> result = addrPool.get(tag);
if (CollectionUtils.isNotEmpty(result) || (tagRouterRuleCopy != null && tagRouterRuleCopy.isForce())
|| isForceUseTag(invocation)) {
return invokers.intersect(result, invokers.getUnmodifiableList());
} else {
invocation.setAttachment(TAG_KEY, NO_TAG);
return invokers;
}
}
}
private boolean isForceUseTag(Invocation invocation) {
return Boolean.valueOf(invocation.getAttachment(FORCE_USE_TAG, url.getParameter(FORCE_USE_TAG, "false")));
}
@Override
public boolean isRuntime() {
return tagRouterRule != null && tagRouterRule.isRuntime();
}
@Override
public boolean isEnable() {
return true;
}
@Override
public boolean isForce() {
return tagRouterRule != null && tagRouterRule.isForce();
}
@Override
public String getName() {
return "TagDynamic";
}
@Override
public boolean shouldRePool() {
return false;
}
@Override
public <T> RouterCache<T> pool(List<Invoker<T>> invokers) {
RouterCache<T> routerCache = new RouterCache<>();
ConcurrentHashMap<String, BitList<Invoker<T>>> addrPool = new ConcurrentHashMap<>();
final TagRouterRule tagRouterRuleCopy = tagRouterRule;
if (tagRouterRuleCopy == null || !tagRouterRuleCopy.isValid() || !tagRouterRuleCopy.isEnabled()) {
BitList<Invoker<T>> noTagList = new BitList<>(invokers, true);
for (int index = 0; index < invokers.size(); index++) {
noTagList.addIndex(index);
}
addrPool.put(NO_TAG, noTagList);
routerCache.setAddrPool(addrPool);
return routerCache;
}
List<String> tagNames = tagRouterRuleCopy.getTagNames();
Map<String, List<String>> tagnameToAddresses = tagRouterRuleCopy.getTagnameToAddresses();
for (String tag : tagNames) {
List<String> addresses = tagnameToAddresses.get(tag);
BitList<Invoker<T>> list = new BitList<>(invokers, true);
if (CollectionUtils.isEmpty(addresses)) {
list.addAll(invokers);
} else {
for (int index = 0; index < invokers.size(); index++) {
Invoker<T> invoker = invokers.get(index);
if (addressMatches(invoker.getUrl(), addresses)) {
list.addIndex(index);
}
}
}
addrPool.put(tag, list);
}
List<String> addresses = tagRouterRuleCopy.getAddresses();
BitList<Invoker<T>> noTagList = new BitList<>(invokers, true);
for (int index = 0; index < invokers.size(); index++) {
Invoker<T> invoker = invokers.get(index);
if (addressNotMatches(invoker.getUrl(), addresses)) {
noTagList.addIndex(index);
}
}
addrPool.put(NO_TAG, noTagList);
routerCache.setAddrPool(addrPool);
routerCache.setAddrMetadata(tagRouterRuleCopy);
return routerCache;
}
private boolean addressMatches(URL url, List<String> addresses) {
return addresses != null && checkAddressMatch(addresses, url.getHost(), url.getPort());
}
private boolean addressNotMatches(URL url, List<String> addresses) {
return addresses == null || !checkAddressMatch(addresses, url.getHost(), url.getPort());
}
private boolean checkAddressMatch(List<String> addresses, String host, int port) {
for (String address : addresses) {
try {
if (NetUtils.matchIpExpression(address, host, port)) {
return true;
}
if ((ANYHOST_VALUE + ":" + port).equals(address)) {
return true;
}
} catch (UnknownHostException e) {
logger.error("The format of ip address is invalid in tag route. Address :" + address, e);
} catch (Exception e) {
logger.error("The format of ip address is invalid in tag route. Address :" + address, e);
}
}
return false;
}
public void setApplication(String app) {
this.application = app;
}
@Override
public <T> void notify(List<Invoker<T>> invokers) {
if (CollectionUtils.isEmpty(invokers)) {
return;
}
Invoker<T> invoker = invokers.get(0);
URL url = invoker.getUrl();
String providerApplication = url.getRemoteApplication();
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.");
return;
}
synchronized (this) {
if (!providerApplication.equals(application)) {
if (!StringUtils.isEmpty(application)) {
ruleRepository.removeListener(application + RULE_SUFFIX, this);
}
String key = providerApplication + RULE_SUFFIX;
ruleRepository.addListener(key, this);
application = providerApplication;
String rawRule = ruleRepository.getRule(key, DynamicConfiguration.DEFAULT_GROUP);
if (StringUtils.isNotEmpty(rawRule)) {
this.process(new ConfigChangedEvent(key, DynamicConfiguration.DEFAULT_GROUP, rawRule));
}
}
}
pool(invokers);
}
}

37
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/tag/TagDynamicStateRouterFactory.java Normal file
View File

@ -0,0 +1,37 @@
/*
* 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.tag;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.rpc.cluster.RouterChain;
import org.apache.dubbo.rpc.cluster.router.state.CacheableStateRouterFactory;
import org.apache.dubbo.rpc.cluster.router.state.StateRouter;
/**
* Tag router factory
*/
@Activate(order = 99)
public class TagDynamicStateRouterFactory extends CacheableStateRouterFactory {
public static final String NAME = "tag-dynamic";
@Override
protected StateRouter createRouter(URL url, RouterChain chain) {
return new TagDynamicStateRouter(url, chain);
}
}

155
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/tag/TagStaticStateRouter.java Normal file
View File

@ -0,0 +1,155 @@
/*
* 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.tag;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.apache.dubbo.common.URL;
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.StringUtils;
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.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.tag.model.TagRouterRule;
import static org.apache.dubbo.common.constants.CommonConstants.TAG_KEY;
/**
* TagStaticStateRouter, "application.tag-router"
*/
public class TagStaticStateRouter extends AbstractStateRouter {
public static final String NAME = "TAG_ROUTER";
private static final int TAG_ROUTER_DEFAULT_PRIORITY = 100;
private static final Logger logger = LoggerFactory.getLogger(TagStaticStateRouter.class);
private static final String NO_TAG = "noTag";
private TagRouterRule tagRouterRule;
public TagStaticStateRouter(URL url, RouterChain chain) {
super(url, chain);
this.priority = TAG_ROUTER_DEFAULT_PRIORITY;
}
@Override
public URL getUrl() {
return url;
}
@Override
public <T> BitList<Invoker<T>> route(BitList<Invoker<T>> invokers, RouterCache<T> routerCache, URL url, Invocation invocation)
throws RpcException {
String tag = StringUtils.isEmpty(invocation.getAttachment(TAG_KEY)) ? url.getParameter(TAG_KEY) :
invocation.getAttachment(TAG_KEY);
if (StringUtils.isEmpty(tag)) {
tag = NO_TAG;
}
ConcurrentMap<String, BitList<Invoker<T>>> pool = routerCache.getAddrPool();
BitList<Invoker<T>> res = pool.get(tag);
if (res == null) {
return invokers;
}
return invokers.intersect(res, invokers.getUnmodifiableList());
}
@Override
protected List<String> getTags(URL url, Invocation invocation) {
List<String> tags = new ArrayList<>();
String tag = StringUtils.isEmpty(invocation.getAttachment(TAG_KEY)) ? url.getParameter(TAG_KEY) :
invocation.getAttachment(TAG_KEY);
if (StringUtils.isEmpty(tag)) {
tag = NO_TAG;
}
tags.add(tag);
return tags;
}
@Override
public boolean isRuntime() {
return tagRouterRule != null && tagRouterRule.isRuntime();
}
@Override
public boolean isEnable() {
return true;
}
@Override
public boolean isForce() {
// FIXME
return false;
}
@Override
public String getName() {
return "TagStatic";
}
@Override
public boolean shouldRePool() {
return false;
}
@Override
public <T> RouterCache<T> pool(List<Invoker<T>> invokers) {
RouterCache<T> routerCache = new RouterCache<>();
ConcurrentHashMap<String, BitList<Invoker<T>>> addrPool = new ConcurrentHashMap<>();
for (int index = 0; index < invokers.size(); index++) {
Invoker<T> invoker = invokers.get(index);
String tag = invoker.getUrl().getParameter(TAG_KEY);
if (StringUtils.isEmpty(tag)) {
BitList<Invoker<T>> noTagList = addrPool.putIfAbsent(NO_TAG, new BitList<>(invokers, true));
if (noTagList == null) {
noTagList = addrPool.get(NO_TAG);
}
noTagList.addIndex(index);
} else {
BitList<Invoker<T>> list = addrPool.putIfAbsent(tag, new BitList<>(invokers, true));
if (list == null) {
list = addrPool.get(tag);
}
list.addIndex(index);
}
}
routerCache.setAddrPool(addrPool);
return routerCache;
}
@Override
public <T> void notify(List<Invoker<T>> invokers) {
if (CollectionUtils.isEmpty(invokers)) {
return;
}
pool(invokers);
}
}

37
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/tag/TagStaticStateRouterFactory.java Normal file
View File

@ -0,0 +1,37 @@
/*
* 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.tag;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.rpc.cluster.RouterChain;
import org.apache.dubbo.rpc.cluster.router.state.CacheableStateRouterFactory;
import org.apache.dubbo.rpc.cluster.router.state.StateRouter;
/**
* Tag router factory
*/
@Activate(order = 100)
public class TagStaticStateRouterFactory extends CacheableStateRouterFactory {
public static final String NAME = "tag-static";
@Override
protected StateRouter createRouter(URL url, RouterChain chain) {
return new TagStaticStateRouter(url, chain);
}
}

2
dubbo-cluster/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.router.state.StateRouterFactory Normal file
View File

@ -0,0 +1,2 @@
tag-dynamic=org.apache.dubbo.rpc.cluster.router.tag.TagDynamicStateRouterFactory
tag-static=org.apache.dubbo.rpc.cluster.router.tag.TagStaticStateRouterFactory

40
dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/manager/DefaultExecutorRepository.java
View File

@ -16,20 +16,23 @@
*/
package org.apache.dubbo.common.threadpool.manager;
import org.apache.dubbo.common.URL;
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.threadpool.ThreadPool;
import org.apache.dubbo.common.utils.NamedThreadFactory;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.dubbo.common.URL;
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.threadlocal.NamedInternalThreadFactory;
import org.apache.dubbo.common.threadpool.ThreadPool;
import org.apache.dubbo.common.utils.NamedThreadFactory;
import static org.apache.dubbo.common.constants.CommonConstants.CONSUMER_SIDE;
import static org.apache.dubbo.common.constants.CommonConstants.EXECUTOR_SERVICE_COMPONENT_KEY;
@ -59,13 +62,25 @@ public class DefaultExecutorRepository implements ExecutorRepository {
private ConcurrentMap<String, ConcurrentMap<Integer, ExecutorService>> data = new ConcurrentHashMap<>();
private ExecutorService poolRouterExecutor;
private static Ring<ExecutorService> executorServiceRing = new Ring<ExecutorService>();
public DefaultExecutorRepository() {
for (int i = 0; i < DEFAULT_SCHEDULER_SIZE; i++) {
ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("Dubbo-framework-scheduler"));
ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(
new NamedThreadFactory("Dubbo-framework-scheduler"));
scheduledExecutors.addItem(scheduler);
executorServiceRing.addItem(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>(1024), new NamedInternalThreadFactory("Dubbo-state-router-loop", true)
, new ThreadPoolExecutor.AbortPolicy()));
}
//
// reconnectScheduledExecutor = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("Dubbo-reconnect-scheduler"));
poolRouterExecutor = new ThreadPoolExecutor(1, 10, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(1024),
new NamedInternalThreadFactory("Dubbo-state-router-pool-router", true), new ThreadPoolExecutor.AbortPolicy());
serviceExporterExecutor = Executors.newScheduledThreadPool(1, new NamedThreadFactory("Dubbo-exporter-scheduler"));
serviceDiscveryAddressNotificationExecutor = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("Dubbo-SD-address-refresh"));
registryNotificationExecutor = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("Dubbo-registry-notification"));
@ -157,6 +172,11 @@ public class DefaultExecutorRepository implements ExecutorRepository {
return scheduledExecutors.pollItem();
}
@Override
public ExecutorService nextExecutorExecutor() {
return executorServiceRing.pollItem();
}
@Override
public ScheduledExecutorService getServiceExporterExecutor() {
return serviceExporterExecutor;
@ -185,4 +205,8 @@ public class DefaultExecutorRepository implements ExecutorRepository {
return (ExecutorService) ExtensionLoader.getExtensionLoader(ThreadPool.class).getAdaptiveExtension().getExecutor(url);
}
@Override
public ExecutorService getPoolRouterExecutor() {
return poolRouterExecutor;
}
}

5
dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/manager/ExecutorRepository.java
View File

@ -55,6 +55,8 @@ public interface ExecutorRepository {
*/
ScheduledExecutorService nextScheduledExecutor();
ExecutorService nextExecutorExecutor();
ScheduledExecutorService getServiceExporterExecutor();
ScheduledExecutorService getServiceDiscoveryAddressNotificationExecutor();
@ -68,7 +70,6 @@ public interface ExecutorRepository {
*/
ScheduledExecutorService getRegistryNotificationExecutor();
/**
* Get the default shared threadpool.
*
@ -76,4 +77,6 @@ public interface ExecutorRepository {
*/
ExecutorService getSharedExecutor();
ExecutorService getPoolRouterExecutor();
}

6
dubbo-dependencies-bom/pom.xml
View File

@ -160,6 +160,7 @@
<hessian_lite_version>3.2.8</hessian_lite_version>
<swagger_version>1.5.19</swagger_version>
<roaringbitmap_version>0.9.0</roaringbitmap_version>
<metrics_version>2.0.1</metrics_version>
<sofa_registry_version>5.2.0</sofa_registry_version>
<gson_version>2.8.5</gson_version>
@ -338,6 +339,11 @@
<artifactId>protobuf-java-util</artifactId>
<version>${protobuf-java_version}</version>
</dependency>
<dependency>
<groupId>org.roaringbitmap</groupId>
<artifactId>RoaringBitmap</artifactId>
<version>${roaringbitmap_version}</version>
</dependency>
<!-- Common Annotations API -->
<dependency>
<groupId>javax.annotation</groupId>

4
dubbo-distribution/dubbo-all/pom.xml
View File

@ -331,6 +331,10 @@
<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>

1
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcException.java
View File

@ -38,6 +38,7 @@ public /**final**/ class RpcException extends RuntimeException {
public static final int LIMIT_EXCEEDED_EXCEPTION = 7;
public static final int TIMEOUT_TERMINATE = 8;
public static final int REGISTRY_EXCEPTION = 9;
public static final int ROUTER_CACHE_NOT_BUILD = 10;
private static final long serialVersionUID = 7815426752583648734L;
/**
* RpcException cannot be extended, use error code for exception type to keep compatibility