diff --git a/dubbo-cluster/pom.xml b/dubbo-cluster/pom.xml index bb588497f6..3055454d9e 100644 --- a/dubbo-cluster/pom.xml +++ b/dubbo-cluster/pom.xml @@ -31,6 +31,10 @@ false + + org.roaringbitmap + RoaringBitmap + org.apache.dubbo dubbo-rpc-api 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 a16b2aa256..eb92945440 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,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 { // instance will never delete or recreate. private List builtinRouters = Collections.emptyList(); + private List builtinStateRouters = Collections.emptyList(); + private List stateRouters = Collections.emptyList(); + private final ExecutorRepository executorRepository = ExtensionLoader.getExtensionLoader(ExecutorRepository.class) + .getDefaultExtension(); + + protected URL url; + + protected AtomicReference> 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 RouterChain buildChain(URL url) { return new RouterChain<>(url); } private RouterChain(URL url) { + loopPool = executorRepository.nextExecutorExecutor(); List extensionFactories = ExtensionLoader.getExtensionLoader(RouterFactory.class) - .getActivateExtension(url, "router"); + .getActivateExtension(url, "router"); List routers = extensionFactories.stream() - .map(factory -> factory.getRouter(url)) - .collect(Collectors.toList()); + .map(factory -> factory.getRouter(url)) + .collect(Collectors.toList()); initWithRouters(routers); + + List extensionStateRouterFactories = ExtensionLoader.getExtensionLoader( + StateRouterFactory.class) + .getActivateExtension(url, "stateRouter"); + + List 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 { this.sort(); } + public void initWithStateRouters(List 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 { this.routers = newRouters; } + public void addStateRouters(List stateRouters) { + List newStateRouters = new ArrayList<>(); + newStateRouters.addAll(builtinStateRouters); + newStateRouters.addAll(stateRouters); + CollectionUtils.sort(newStateRouters); + this.stateRouters = newStateRouters; + } + public List getRouters() { return routers; } @@ -95,13 +152,35 @@ public class RouterChain { } /** - * * @param url * @param invocation * @return */ public List> route(URL url, Invocation invocation) { - List> finalInvokers = invokers; + + AddrCache 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> finalBitListInvokers = new BitList<>(invokers, false); + for (StateRouter stateRouter : stateRouters) { + if (stateRouter.isEnable()) { + RouterCache routerCache = cache.getCache().get(stateRouter.getName()); + finalBitListInvokers = stateRouter.route(finalBitListInvokers, routerCache, url, invocation); + } + } + + List> finalInvokers = new ArrayList<>(finalBitListInvokers.size()); + + for(Invoker invoker: finalBitListInvokers) { + finalInvokers.add(invoker); + } + for (Router router : routers) { finalInvokers = router.route(finalInvokers, url, invocation); } @@ -114,7 +193,99 @@ public class RouterChain { */ public void setInvokers(List> 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 origin = cache.get(); + List> copyInvokers = new ArrayList<>(this.invokers); + AddrCache newCache = new AddrCache(); + 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 orign, List> 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 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 { } 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(); } + } 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 db1b202483..50110aa6ab 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 @@ -94,6 +94,7 @@ public class StaticDirectory extends AbstractDirectory { public void buildRouterChain() { RouterChain routerChain = RouterChain.buildChain(getUrl()); routerChain.setInvokers(invokers); + routerChain.loop(true); this.setRouterChain(routerChain); } 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 new file mode 100644 index 0000000000..fc5e3624ff --- /dev/null +++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/AbstractStateRouter.java @@ -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 invokers; + protected AtomicReference 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 void notify(List> 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 BitList> route(BitList> invokers, RouterCache cache, URL url, + Invocation invocation) throws RpcException { + + List tags = getTags(url, invocation); + + if (tags == null) { + return invokers; + } + for (String tag : tags) { + BitList> tagInvokers = cache.getAddrPool().get(tag); + if (tagMatchFail(tagInvokers)) { + continue; + } + return tagInvokers.intersect(invokers, invokers.getUnmodifiableList()); + } + + return invokers; + } + + protected List getTags(URL url, Invocation invocation) { + return new ArrayList(); + } + + public Boolean tagMatchFail(BitList> invokers) { + return invokers == null || invokers.isEmpty(); + } + + @Override + public void pool() { + chain.loop(false); + } +} diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/AddrCache.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/AddrCache.java new file mode 100644 index 0000000000..9d1e9b3346 --- /dev/null +++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/AddrCache.java @@ -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 + * @since 3.0 + */ +public class AddrCache { + private final static ConcurrentHashMap EMPTY_MAP = new ConcurrentHashMap<>(); + + protected List> invokers; + protected ConcurrentMap> cache = EMPTY_MAP; + + public List> getInvokers() { + return invokers; + } + + public void setInvokers(List> invokers) { + this.invokers = invokers; + } + + public ConcurrentMap> getCache() { + return cache; + } + + public void setCache(ConcurrentHashMap> cache) { + this.cache = cache; + } +} 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 new file mode 100644 index 0000000000..71456cb640 --- /dev/null +++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/BitList.java @@ -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 + * @since 3.0 + */ +public class BitList implements List { + private final RoaringBitmap rootMap; + private final List unmodifiableList; + + public BitList(List unmodifiableList, boolean empty) { + this.unmodifiableList = unmodifiableList; + this.rootMap = new RoaringBitmap(); + if (!empty) { + this.rootMap.add(0L, unmodifiableList.size()); + } + } + + public BitList(List unmodifiableList, RoaringBitmap rootMap) { + this.unmodifiableList = unmodifiableList; + this.rootMap = rootMap; + } + + public BitList(List unmodifiableList) { + this(unmodifiableList, false); + } + + public List getUnmodifiableList() { + return unmodifiableList; + } + + public void addIndex(int index) { + this.rootMap.add(index); + } + + public BitList intersect(List b, List 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 iterator() { + return new Iterator() { + 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[] 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 c) { + throw new UnsupportedOperationException(); + } + + @Override + public boolean addAll(int index, Collection 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 listIterator() { + throw new UnsupportedOperationException(); + } + + @Override + public ListIterator listIterator(int index) { + throw new UnsupportedOperationException(); + } + + @Override + public List subList(int fromIndex, int toIndex) { + throw new UnsupportedOperationException(); + } + +} diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/CacheableStateRouterFactory.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/CacheableStateRouterFactory.java new file mode 100644 index 0000000000..5f38ccb99a --- /dev/null +++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/CacheableStateRouterFactory.java @@ -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 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); +} diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/RouterCache.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/RouterCache.java new file mode 100644 index 0000000000..0b7b499ce8 --- /dev/null +++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/RouterCache.java @@ -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 + * @since 3.0 + */ +public class RouterCache { + private final static ConcurrentHashMap EMPTY_MAP = new ConcurrentHashMap<>(); + protected ConcurrentMap>> addrPool = EMPTY_MAP; + protected Object addrMetadata; + + public ConcurrentMap>> getAddrPool() { + return addrPool; + } + + public void setAddrPool(ConcurrentHashMap>> addrPool) { + this.addrPool = addrPool; + } + + public Object getAddrMetadata() { + return addrMetadata; + } + + public void setAddrMetadata(Object addrMetadata) { + this.addrMetadata = addrMetadata; + } +} \ No newline at end of file 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 new file mode 100644 index 0000000000..ee28459837 --- /dev/null +++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/StateRouter.java @@ -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) + *

+ * Routing + * + * @see org.apache.dubbo.rpc.cluster.Cluster#join(Directory) + * @see Directory#list(Invocation) + * @since 3.0 + */ +public interface StateRouter extends Comparable { + + 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 + * @return routed invokers + * @throws RpcException + * @Since 3.0 + */ + BitList> route(BitList> invokers, RouterCache cache, URL url, Invocation invocation) + throws + RpcException; + + default void notify(List> 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(); + + RouterCache pool(List> invokers); + + void pool(); + + default void stop() { + //do nothing by default + } +} diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/StateRouterFactory.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/StateRouterFactory.java new file mode 100644 index 0000000000..9780c367c4 --- /dev/null +++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/StateRouterFactory.java @@ -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") + StateRouter getRouter(URL url, RouterChain chain); +} 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 new file mode 100644 index 0000000000..231aa2c85b --- /dev/null +++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/tag/TagDynamicStateRouter.java @@ -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 BitList> route(BitList> invokers, RouterCache 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>> addrPool = cache.getAddrPool(); + + if (StringUtils.isEmpty(tag)) { + return invokers.intersect(addrPool.get(NO_TAG), invokers.getUnmodifiableList()); + } else { + BitList> 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 RouterCache pool(List> invokers) { + + RouterCache routerCache = new RouterCache<>(); + ConcurrentHashMap>> addrPool = new ConcurrentHashMap<>(); + + final TagRouterRule tagRouterRuleCopy = tagRouterRule; + + + if (tagRouterRuleCopy == null || !tagRouterRuleCopy.isValid() || !tagRouterRuleCopy.isEnabled()) { + BitList> 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 tagNames = tagRouterRuleCopy.getTagNames(); + Map> tagnameToAddresses = tagRouterRuleCopy.getTagnameToAddresses(); + + for (String tag : tagNames) { + List addresses = tagnameToAddresses.get(tag); + BitList> list = new BitList<>(invokers, true); + + if (CollectionUtils.isEmpty(addresses)) { + list.addAll(invokers); + } else { + for (int index = 0; index < invokers.size(); index++) { + Invoker invoker = invokers.get(index); + if (addressMatches(invoker.getUrl(), addresses)) { + list.addIndex(index); + } + } + } + + addrPool.put(tag, list); + } + + List addresses = tagRouterRuleCopy.getAddresses(); + BitList> noTagList = new BitList<>(invokers, true); + + for (int index = 0; index < invokers.size(); index++) { + Invoker 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 addresses) { + return addresses != null && checkAddressMatch(addresses, url.getHost(), url.getPort()); + } + + private boolean addressNotMatches(URL url, List addresses) { + return addresses == null || !checkAddressMatch(addresses, url.getHost(), url.getPort()); + } + + private boolean checkAddressMatch(List 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 void notify(List> invokers) { + if (CollectionUtils.isEmpty(invokers)) { + return; + } + + Invoker 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); + } + +} diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/tag/TagDynamicStateRouterFactory.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/tag/TagDynamicStateRouterFactory.java new file mode 100644 index 0000000000..cdcdb8408c --- /dev/null +++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/tag/TagDynamicStateRouterFactory.java @@ -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); + } +} 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 new file mode 100644 index 0000000000..d15745f7e5 --- /dev/null +++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/tag/TagStaticStateRouter.java @@ -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 BitList> route(BitList> invokers, RouterCache 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>> pool = routerCache.getAddrPool(); + BitList> res = pool.get(tag); + if (res == null) { + return invokers; + } + return invokers.intersect(res, invokers.getUnmodifiableList()); + } + + @Override + protected List getTags(URL url, Invocation invocation) { + List 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 RouterCache pool(List> invokers) { + + RouterCache routerCache = new RouterCache<>(); + ConcurrentHashMap>> addrPool = new ConcurrentHashMap<>(); + + for (int index = 0; index < invokers.size(); index++) { + Invoker invoker = invokers.get(index); + String tag = invoker.getUrl().getParameter(TAG_KEY); + if (StringUtils.isEmpty(tag)) { + BitList> noTagList = addrPool.putIfAbsent(NO_TAG, new BitList<>(invokers, true)); + if (noTagList == null) { + noTagList = addrPool.get(NO_TAG); + } + noTagList.addIndex(index); + } else { + BitList> 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 void notify(List> invokers) { + if (CollectionUtils.isEmpty(invokers)) { + return; + } + + pool(invokers); + } + +} diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/tag/TagStaticStateRouterFactory.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/tag/TagStaticStateRouterFactory.java new file mode 100644 index 0000000000..21fe348682 --- /dev/null +++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/tag/TagStaticStateRouterFactory.java @@ -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); + } +} diff --git a/dubbo-cluster/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.router.state.StateRouterFactory b/dubbo-cluster/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.router.state.StateRouterFactory new file mode 100644 index 0000000000..c69b3abdfb --- /dev/null +++ b/dubbo-cluster/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.router.state.StateRouterFactory @@ -0,0 +1,2 @@ +tag-dynamic=org.apache.dubbo.rpc.cluster.router.tag.TagDynamicStateRouterFactory +tag-static=org.apache.dubbo.rpc.cluster.router.tag.TagStaticStateRouterFactory \ No newline at end of file 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 b8eb41ee05..02bd6c977b 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 @@ -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> data = new ConcurrentHashMap<>(); + private ExecutorService poolRouterExecutor; + + private static Ring executorServiceRing = new Ring(); + 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(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(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; + } } 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 d00a7bb4e2..0a8b145ca8 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 @@ -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(); + } diff --git a/dubbo-dependencies-bom/pom.xml b/dubbo-dependencies-bom/pom.xml index 675b7c74ec..5d59d8de3e 100644 --- a/dubbo-dependencies-bom/pom.xml +++ b/dubbo-dependencies-bom/pom.xml @@ -160,6 +160,7 @@ 3.2.8 1.5.19 + 0.9.0 2.0.1 5.2.0 2.8.5 @@ -338,6 +339,11 @@ protobuf-java-util ${protobuf-java_version} + + org.roaringbitmap + RoaringBitmap + ${roaringbitmap_version} + javax.annotation diff --git a/dubbo-distribution/dubbo-all/pom.xml b/dubbo-distribution/dubbo-all/pom.xml index 519fbc781d..d3d4a184f9 100644 --- a/dubbo-distribution/dubbo-all/pom.xml +++ b/dubbo-distribution/dubbo-all/pom.xml @@ -331,6 +331,10 @@ com.alibaba fastjson + + org.roaringbitmap + RoaringBitmap + diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcException.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcException.java index 3228debf92..d57815158d 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcException.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcException.java @@ -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