This is an automated email from the ASF dual-hosted git repository.
albumenj pushed a commit to branch 3.2
in repository https://gitbox.apache.org/repos/asf/dubbo.git
The following commit(s) were added to refs/heads/3.2 by this push:
new d2eb0ef458 [ISSUE #11294] Optimize ConcurrentHashMap#computeIfAbsent
have performance problem in jdk1.8 (#11326)
d2eb0ef458 is described below
commit d2eb0ef4589b956d1648d6809cffd0a9966f5eb6
Author: mxsm <[email protected]>
AuthorDate: Thu Jan 19 23:07:12 2023 +0800
[ISSUE #11294] Optimize ConcurrentHashMap#computeIfAbsent have performance
problem in jdk1.8 (#11326)
---
.../dubbo/rpc/cluster/CacheableRouterFactory.java | 3 +-
.../cluster/loadbalance/RoundRobinLoadBalance.java | 9 ++--
.../loadbalance/ShortestResponseLoadBalance.java | 3 +-
.../router/mesh/util/MeshRuleDispatcher.java | 8 +--
.../cluster/router/script/ScriptStateRouter.java | 15 +++---
.../router/state/CacheableStateRouterFactory.java | 3 +-
.../common/beans/factory/ScopeBeanFactory.java | 6 +--
.../org/apache/dubbo/common/bytecode/Wrapper.java | 16 +++---
.../dubbo/common/cache/FileCacheStoreFactory.java | 8 +--
.../AbstractDynamicConfigurationFactory.java | 6 +--
.../apache/dubbo/common/convert/ConverterUtil.java | 9 ++--
.../apache/dubbo/common/logger/LoggerFactory.java | 9 ++--
.../collector/stat/MetricsStatComposite.java | 49 +++++++++++--------
.../common/store/support/SimpleDataStore.java | 5 +-
.../manager/DefaultExecutorRepository.java | 5 +-
.../common/url/component/ServiceConfigURL.java | 24 ++++-----
.../dubbo/common/utils/ConcurrentHashMapUtils.java | 57 ++++++++++++++++++++++
.../org/apache/dubbo/config/AbstractConfig.java | 10 ++--
.../rpc/model/FrameworkServiceRepository.java | 5 +-
.../dubbo/rpc/model/ModuleServiceRepository.java | 9 ++--
.../common/utils/ConcurrentHashMapUtilsTest.java | 31 ++++++------
.../dubbo/config/bootstrap/DubboBootstrap.java | 12 +++--
.../dubbo/config/utils/SimpleReferenceCache.java | 13 ++---
.../spring/reference/ReferenceBeanManager.java | 12 +++--
.../config/spring/AbstractRegistryService.java | 7 +--
.../dubbo/config/spring/SimpleRegistryService.java | 21 ++++----
.../support/nacos/NacosDynamicConfiguration.java | 7 +--
.../support/zookeeper/CacheListener.java | 6 ++-
.../store/zookeeper/ZookeeperMetadataReport.java | 6 ++-
.../collector/AggregateMetricsCollector.java | 39 +++++++--------
.../dubbo/monitor/support/MonitorFilter.java | 3 +-
.../apache/dubbo/monitor/dubbo/DubboMonitor.java | 3 +-
.../client/AbstractServiceDiscoveryFactory.java | 3 +-
.../client/ReflectionBasedServiceDiscovery.java | 3 +-
.../registry/client/ServiceDiscoveryRegistry.java | 3 +-
.../client/ServiceDiscoveryRegistryDirectory.java | 6 ++-
.../DefaultMigrationAddressComparator.java | 6 ++-
.../client/migration/MigrationRuleListener.java | 6 ++-
.../support/CacheableFailbackRegistry.java | 14 +++---
.../multiple/MultipleServiceDiscovery.java | 6 ++-
.../registry/nacos/NacosNamingServiceWrapper.java | 16 +++---
.../registry/zookeeper/ZookeeperRegistry.java | 9 ++--
.../MultiplexProtocolConnectionManager.java | 3 +-
.../exchange/PortUnificationExchanger.java | 5 +-
.../dubbo/remoting/transport/CodecSupport.java | 6 ++-
.../zookeeper/AbstractZookeeperClient.java | 9 ++--
.../java/org/apache/dubbo/rpc/AdaptiveMetrics.java | 32 ++++++------
.../main/java/org/apache/dubbo/rpc/RpcStatus.java | 11 +++--
.../apache/dubbo/rpc/filter/AccessLogFilter.java | 8 +--
.../dubbo/rpc/protocol/AbstractProtocol.java | 3 +-
.../dubbo/rpc/protocol/grpc/GrpcProtocol.java | 21 ++++----
.../dubbo/rpc/protocol/rest/RestProtocol.java | 8 +--
.../rpc/protocol/tri/SingleProtobufUtils.java | 4 +-
.../xds/util/protocol/AbstractProtocol.java | 3 +-
.../rpc/cluster/router/xds/EdsEndpointManager.java | 5 +-
.../cluster/router/xds/RdsRouteRuleManager.java | 8 +--
56 files changed, 378 insertions(+), 239 deletions(-)
diff --git
a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/CacheableRouterFactory.java
b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/CacheableRouterFactory.java
index 4d4b9eb5e0..c3a69374df 100644
---
a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/CacheableRouterFactory.java
+++
b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/CacheableRouterFactory.java
@@ -17,6 +17,7 @@
package org.apache.dubbo.rpc.cluster;
import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@@ -30,7 +31,7 @@ public abstract class CacheableRouterFactory implements
RouterFactory {
@Override
public Router getRouter(URL url) {
- return routerMap.computeIfAbsent(url.getServiceKey(), k ->
createRouter(url));
+ return ConcurrentHashMapUtils.computeIfAbsent(routerMap,
url.getServiceKey(), k -> createRouter(url));
}
protected abstract Router createRouter(URL url);
diff --git
a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/loadbalance/RoundRobinLoadBalance.java
b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/loadbalance/RoundRobinLoadBalance.java
index b3e5adb9e6..4d4f7bfbf2 100644
---
a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/loadbalance/RoundRobinLoadBalance.java
+++
b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/loadbalance/RoundRobinLoadBalance.java
@@ -17,6 +17,7 @@
package org.apache.dubbo.rpc.cluster.loadbalance;
import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
@@ -35,6 +36,8 @@ public class RoundRobinLoadBalance extends
AbstractLoadBalance {
private static final int RECYCLE_PERIOD = 60000;
+ private ConcurrentMap<String, ConcurrentMap<String, WeightedRoundRobin>>
methodWeightMap = new ConcurrentHashMap<>();
+
protected static class WeightedRoundRobin {
private int weight;
private AtomicLong current = new AtomicLong(0);
@@ -66,8 +69,6 @@ public class RoundRobinLoadBalance extends
AbstractLoadBalance {
}
}
- private ConcurrentMap<String, ConcurrentMap<String, WeightedRoundRobin>>
methodWeightMap = new ConcurrentHashMap<String, ConcurrentMap<String,
WeightedRoundRobin>>();
-
/**
* get invoker addr list cached for specified invocation
* <p>
@@ -89,7 +90,7 @@ public class RoundRobinLoadBalance extends
AbstractLoadBalance {
@Override
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url,
Invocation invocation) {
String key = invokers.get(0).getUrl().getServiceKey() + "." +
invocation.getMethodName();
- ConcurrentMap<String, WeightedRoundRobin> map =
methodWeightMap.computeIfAbsent(key, k -> new ConcurrentHashMap<>());
+ ConcurrentMap<String, WeightedRoundRobin> map =
ConcurrentHashMapUtils.computeIfAbsent(methodWeightMap, key, k -> new
ConcurrentHashMap<>());
int totalWeight = 0;
long maxCurrent = Long.MIN_VALUE;
long now = System.currentTimeMillis();
@@ -98,7 +99,7 @@ public class RoundRobinLoadBalance extends
AbstractLoadBalance {
for (Invoker<T> invoker : invokers) {
String identifyString = invoker.getUrl().toIdentityString();
int weight = getWeight(invoker, invocation);
- WeightedRoundRobin weightedRoundRobin =
map.computeIfAbsent(identifyString, k -> {
+ WeightedRoundRobin weightedRoundRobin =
ConcurrentHashMapUtils.computeIfAbsent(map, identifyString, k -> {
WeightedRoundRobin wrr = new WeightedRoundRobin();
wrr.setWeight(weight);
return wrr;
diff --git
a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/loadbalance/ShortestResponseLoadBalance.java
b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/loadbalance/ShortestResponseLoadBalance.java
index 37717daec2..906c49642f 100644
---
a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/loadbalance/ShortestResponseLoadBalance.java
+++
b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/loadbalance/ShortestResponseLoadBalance.java
@@ -18,6 +18,7 @@ package org.apache.dubbo.rpc.cluster.loadbalance;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.threadpool.manager.FrameworkExecutorRepository;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.RpcStatus;
@@ -116,7 +117,7 @@ public class ShortestResponseLoadBalance extends
AbstractLoadBalance implements
for (int i = 0; i < length; i++) {
Invoker<T> invoker = invokers.get(i);
RpcStatus rpcStatus = RpcStatus.getStatus(invoker.getUrl(),
invocation.getMethodName());
- SlideWindowData slideWindowData =
methodMap.computeIfAbsent(rpcStatus, SlideWindowData::new);
+ SlideWindowData slideWindowData =
ConcurrentHashMapUtils.computeIfAbsent(methodMap, rpcStatus,
SlideWindowData::new);
// Calculate the estimated response time from the product of
active connections and succeeded average elapsed time.
long estimateResponse = slideWindowData.getEstimateResponse();
diff --git
a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mesh/util/MeshRuleDispatcher.java
b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mesh/util/MeshRuleDispatcher.java
index 03f68ddbab..713c44bd1b 100644
---
a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mesh/util/MeshRuleDispatcher.java
+++
b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mesh/util/MeshRuleDispatcher.java
@@ -20,12 +20,14 @@ package org.apache.dubbo.rpc.cluster.router.mesh.util;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.CollectionUtils;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.ConcurrentHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import static
org.apache.dubbo.common.constants.LoggerCodeConstants.CLUSTER_NO_RULE_LISTENER;
@@ -34,7 +36,7 @@ public class MeshRuleDispatcher {
private static final ErrorTypeAwareLogger logger =
LoggerFactory.getErrorTypeAwareLogger(MeshRuleDispatcher.class);
private final String appName;
- private final Map<String, Set<MeshRuleListener>> listenerMap = new
ConcurrentHashMap<>();
+ private final ConcurrentMap<String, Set<MeshRuleListener>> listenerMap =
new ConcurrentHashMap<>();
public MeshRuleDispatcher(String appName) {
this.appName = appName;
@@ -57,7 +59,7 @@ public class MeshRuleDispatcher {
listener.onRuleChange(appName, entry.getValue());
}
} else {
- logger.warn(CLUSTER_NO_RULE_LISTENER,"Receive mesh rule
but none of listener has been registered","","Receive rule but none of listener
has been registered. Maybe type not matched. Rule Type: " + ruleType);
+ logger.warn(CLUSTER_NO_RULE_LISTENER, "Receive mesh rule
but none of listener has been registered", "", "Receive rule but none of
listener has been registered. Maybe type not matched. Rule Type: " + ruleType);
}
}
// clear rule listener not being notified in this time
@@ -75,7 +77,7 @@ public class MeshRuleDispatcher {
if (listener == null) {
return;
}
- listenerMap.computeIfAbsent(listener.ruleSuffix(), (k) -> new
ConcurrentHashSet<>())
+ ConcurrentHashMapUtils.computeIfAbsent(listenerMap,
listener.ruleSuffix(), (k) -> new ConcurrentHashSet<>())
.add(listener);
}
diff --git
a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/script/ScriptStateRouter.java
b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/script/ScriptStateRouter.java
index 174ecb9e0d..ac01380be1 100644
---
a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/script/ScriptStateRouter.java
+++
b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/script/ScriptStateRouter.java
@@ -19,6 +19,7 @@ package org.apache.dubbo.rpc.cluster.router.script;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.Holder;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.rpc.Invocation;
@@ -45,8 +46,8 @@ import java.security.cert.Certificate;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
-import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import java.util.stream.Collectors;
import static
org.apache.dubbo.common.constants.LoggerCodeConstants.CLUSTER_SCRIPT_EXCEPTION;
@@ -64,7 +65,7 @@ public class ScriptStateRouter<T> extends
AbstractStateRouter<T> {
private static final int SCRIPT_ROUTER_DEFAULT_PRIORITY = 0;
private static final ErrorTypeAwareLogger logger =
LoggerFactory.getErrorTypeAwareLogger(ScriptStateRouter.class);
- private static final Map<String, ScriptEngine> ENGINES = new
ConcurrentHashMap<>();
+ private static final ConcurrentMap<String, ScriptEngine> ENGINES = new
ConcurrentHashMap<>();
private final ScriptEngine engine;
@@ -93,8 +94,8 @@ public class ScriptStateRouter<T> extends
AbstractStateRouter<T> {
Compilable compilable = (Compilable) engine;
function = compilable.compile(rule);
} catch (ScriptException e) {
- logger.error(CLUSTER_SCRIPT_EXCEPTION,"script route rule
invalid","","script route error, rule has been ignored. rule: " + rule +
- ", url: " + RpcContext.getServiceContext().getUrl(),e);
+ logger.error(CLUSTER_SCRIPT_EXCEPTION, "script route rule
invalid", "", "script route error, rule has been ignored. rule: " + rule +
+ ", url: " + RpcContext.getServiceContext().getUrl(), e);
}
}
@@ -115,7 +116,7 @@ public class ScriptStateRouter<T> extends
AbstractStateRouter<T> {
private ScriptEngine getEngine(URL url) {
String type = url.getParameter(TYPE_KEY, DEFAULT_SCRIPT_TYPE_KEY);
- return ENGINES.computeIfAbsent(type, t -> {
+ return ConcurrentHashMapUtils.computeIfAbsent(ENGINES, type, t -> {
ScriptEngine scriptEngine = new
ScriptEngineManager().getEngineByName(type);
if (scriptEngine == null) {
throw new IllegalStateException("unsupported route engine
type: " + type);
@@ -137,8 +138,8 @@ public class ScriptStateRouter<T> extends
AbstractStateRouter<T> {
try {
return function.eval(bindings);
} catch (ScriptException e) {
- logger.error(CLUSTER_SCRIPT_EXCEPTION,"Scriptrouter exec
script error","","Script route error, rule has been ignored. rule: " + rule +
", method:" +
- invocation.getMethodName() + ", url: " +
RpcContext.getContext().getUrl(),e);
+ logger.error(CLUSTER_SCRIPT_EXCEPTION, "Scriptrouter exec
script error", "", "Script route error, rule has been ignored. rule: " + rule +
", method:" +
+ invocation.getMethodName() + ", url: " +
RpcContext.getContext().getUrl(), e);
return invokers;
}
}, accessControlContext));
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
index 422c77e635..91d28bbbbc 100644
---
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
@@ -17,6 +17,7 @@
package org.apache.dubbo.rpc.cluster.router.state;
import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@@ -31,7 +32,7 @@ public abstract class CacheableStateRouterFactory implements
StateRouterFactory
@Override
public <T> StateRouter<T> getRouter(Class<T> interfaceClass, URL url) {
- return routerMap.computeIfAbsent(url.getServiceKey(), k ->
createRouter(interfaceClass, url));
+ return ConcurrentHashMapUtils.computeIfAbsent(routerMap,
url.getServiceKey(), k -> createRouter(interfaceClass, url));
}
protected abstract <T> StateRouter<T> createRouter(Class<T>
interfaceClass, URL url);
diff --git
a/dubbo-common/src/main/java/org/apache/dubbo/common/beans/factory/ScopeBeanFactory.java
b/dubbo-common/src/main/java/org/apache/dubbo/common/beans/factory/ScopeBeanFactory.java
index 04448401a1..e549016703 100644
---
a/dubbo-common/src/main/java/org/apache/dubbo/common/beans/factory/ScopeBeanFactory.java
+++
b/dubbo-common/src/main/java/org/apache/dubbo/common/beans/factory/ScopeBeanFactory.java
@@ -24,12 +24,12 @@ import
org.apache.dubbo.common.extension.ExtensionPostProcessor;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.resource.Disposable;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.rpc.model.ScopeModelAccessor;
import java.util.ArrayList;
import java.util.List;
-import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -49,7 +49,7 @@ public class ScopeBeanFactory {
private final ScopeBeanFactory parent;
private ExtensionAccessor extensionAccessor;
private List<ExtensionPostProcessor> extensionPostProcessors;
- private Map<Class, AtomicInteger> beanNameIdCounterMap = new
ConcurrentHashMap<>();
+ private ConcurrentHashMap<Class, AtomicInteger> beanNameIdCounterMap = new
ConcurrentHashMap<>();
private List<BeanInfo> registeredBeanInfos = new CopyOnWriteArrayList<>();
private InstantiationStrategy instantiationStrategy;
private AtomicBoolean destroyed = new AtomicBoolean();
@@ -183,7 +183,7 @@ public class ScopeBeanFactory {
}
private int getNextId(Class<?> beanClass) {
- return beanNameIdCounterMap.computeIfAbsent(beanClass, key -> new
AtomicInteger()).incrementAndGet();
+ return ConcurrentHashMapUtils.computeIfAbsent(beanNameIdCounterMap,
beanClass, key -> new AtomicInteger()).incrementAndGet();
}
public <T> T getBean(Class<T> type) {
diff --git
a/dubbo-common/src/main/java/org/apache/dubbo/common/bytecode/Wrapper.java
b/dubbo-common/src/main/java/org/apache/dubbo/common/bytecode/Wrapper.java
index c1b6699574..bde471afae 100644
--- a/dubbo-common/src/main/java/org/apache/dubbo/common/bytecode/Wrapper.java
+++ b/dubbo-common/src/main/java/org/apache/dubbo/common/bytecode/Wrapper.java
@@ -17,6 +17,7 @@
package org.apache.dubbo.common.bytecode;
import org.apache.dubbo.common.utils.ClassUtils;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.ReflectUtils;
import javassist.ClassPool;
@@ -34,6 +35,7 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.regex.Matcher;
import java.util.stream.Collectors;
@@ -42,7 +44,7 @@ import java.util.stream.Collectors;
* Wrapper.
*/
public abstract class Wrapper {
- private static final Map<Class<?>, Wrapper> WRAPPER_MAP = new
ConcurrentHashMap<Class<?>, Wrapper>(); //class wrapper map
+ private static final ConcurrentMap<Class<?>, Wrapper> WRAPPER_MAP = new
ConcurrentHashMap<Class<?>, Wrapper>(); //class wrapper map
private static final String[] EMPTY_STRING_ARRAY = new String[0];
private static final String[] OBJECT_METHODS = new String[]{"getClass",
"hashCode", "toString", "equals"};
private static final Wrapper OBJECT_WRAPPER = new Wrapper() {
@@ -119,7 +121,7 @@ public abstract class Wrapper {
return OBJECT_WRAPPER;
}
- return WRAPPER_MAP.computeIfAbsent(c, Wrapper::makeWrapper);
+ return ConcurrentHashMapUtils.computeIfAbsent(WRAPPER_MAP, c,
Wrapper::makeWrapper);
}
private static Wrapper makeWrapper(Class<?> c) {
@@ -171,16 +173,16 @@ public abstract class Wrapper {
}
Method[] methods = Arrays.stream(c.getMethods())
- .filter(method ->
allMethod.contains(ReflectUtils.getDesc(method)))
- .collect(Collectors.toList())
- .toArray(new Method[] {});
+ .filter(method -> allMethod.contains(ReflectUtils.getDesc(method)))
+ .collect(Collectors.toList())
+ .toArray(new Method[]{});
// get all public method.
boolean hasMethod = ClassUtils.hasMethods(methods);
if (hasMethod) {
Map<String, Integer> sameNameMethodCount = new HashMap<>((int)
(methods.length / 0.75f) + 1);
for (Method m : methods) {
sameNameMethodCount.compute(m.getName(),
- (key, oldValue) -> oldValue == null ? 1 : oldValue +
1);
+ (key, oldValue) -> oldValue == null ? 1 : oldValue + 1);
}
c3.append(" try{");
@@ -200,7 +202,7 @@ public abstract class Wrapper {
if (len > 0) {
for (int l = 0; l < len; l++) {
c3.append(" && ").append("
$3[").append(l).append("].getName().equals(\"")
-
.append(m.getParameterTypes()[l].getName()).append("\")");
+
.append(m.getParameterTypes()[l].getName()).append("\")");
}
}
}
diff --git
a/dubbo-common/src/main/java/org/apache/dubbo/common/cache/FileCacheStoreFactory.java
b/dubbo-common/src/main/java/org/apache/dubbo/common/cache/FileCacheStoreFactory.java
index 773be9f546..154e59658b 100644
---
a/dubbo-common/src/main/java/org/apache/dubbo/common/cache/FileCacheStoreFactory.java
+++
b/dubbo-common/src/main/java/org/apache/dubbo/common/cache/FileCacheStoreFactory.java
@@ -19,6 +19,7 @@ package org.apache.dubbo.common.cache;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import java.io.File;
import java.io.IOException;
@@ -33,6 +34,7 @@ import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import static
org.apache.dubbo.common.constants.LoggerCodeConstants.COMMON_CACHE_PATH_INACCESSIBLE;
@@ -50,11 +52,11 @@ public final class FileCacheStoreFactory {
}
private static final ErrorTypeAwareLogger logger =
LoggerFactory.getErrorTypeAwareLogger(FileCacheStoreFactory.class);
- private static final Map<String, FileCacheStore> cacheMap = new
ConcurrentHashMap<>();
+ private static final ConcurrentMap<String, FileCacheStore> cacheMap = new
ConcurrentHashMap<>();
private static final String SUFFIX = ".dubbo.cache";
private static final char ESCAPE_MARK = '%';
- private static final Set<Character> LEGAL_CHARACTERS =
Collections.unmodifiableSet(new HashSet<Character>(){{
+ private static final Set<Character> LEGAL_CHARACTERS =
Collections.unmodifiableSet(new HashSet<Character>() {{
// - $ . _ 0-9 a-z A-Z
add('-');
add('$');
@@ -108,7 +110,7 @@ public final class FileCacheStoreFactory {
String cacheFilePath = basePath + File.separator + cacheName;
- return cacheMap.computeIfAbsent(cacheFilePath, k -> getFile(k,
enableFileCache));
+ return ConcurrentHashMapUtils.computeIfAbsent(cacheMap, cacheFilePath,
k -> getFile(k, enableFileCache));
}
/**
diff --git
a/dubbo-common/src/main/java/org/apache/dubbo/common/config/configcenter/AbstractDynamicConfigurationFactory.java
b/dubbo-common/src/main/java/org/apache/dubbo/common/config/configcenter/AbstractDynamicConfigurationFactory.java
index db53e7317f..5ebf0ce867 100644
---
a/dubbo-common/src/main/java/org/apache/dubbo/common/config/configcenter/AbstractDynamicConfigurationFactory.java
+++
b/dubbo-common/src/main/java/org/apache/dubbo/common/config/configcenter/AbstractDynamicConfigurationFactory.java
@@ -17,8 +17,8 @@
package org.apache.dubbo.common.config.configcenter;
import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
-import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_KEY;
@@ -31,12 +31,12 @@ import static
org.apache.dubbo.common.constants.CommonConstants.DEFAULT_KEY;
*/
public abstract class AbstractDynamicConfigurationFactory implements
DynamicConfigurationFactory {
- private volatile Map<String, DynamicConfiguration> dynamicConfigurations =
new ConcurrentHashMap<>();
+ private volatile ConcurrentHashMap<String, DynamicConfiguration>
dynamicConfigurations = new ConcurrentHashMap<>();
@Override
public final DynamicConfiguration getDynamicConfiguration(URL url) {
String key = url == null ? DEFAULT_KEY : url.toServiceString();
- return dynamicConfigurations.computeIfAbsent(key, k ->
createDynamicConfiguration(url));
+ return ConcurrentHashMapUtils.computeIfAbsent(dynamicConfigurations,
key, k -> createDynamicConfiguration(url));
}
protected abstract DynamicConfiguration createDynamicConfiguration(URL
url);
diff --git
a/dubbo-common/src/main/java/org/apache/dubbo/common/convert/ConverterUtil.java
b/dubbo-common/src/main/java/org/apache/dubbo/common/convert/ConverterUtil.java
index 260c980daf..7e5c8b65fc 100644
---
a/dubbo-common/src/main/java/org/apache/dubbo/common/convert/ConverterUtil.java
+++
b/dubbo-common/src/main/java/org/apache/dubbo/common/convert/ConverterUtil.java
@@ -17,16 +17,17 @@
package org.apache.dubbo.common.convert;
import org.apache.dubbo.common.extension.ExtensionLoader;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.rpc.model.FrameworkModel;
import java.util.List;
-import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import java.util.stream.Collectors;
public class ConverterUtil {
private final FrameworkModel frameworkModel;
- private final Map<Class<?>, Map<Class<?>, List<Converter>>> converterCache
= new ConcurrentHashMap<>();
+ private final ConcurrentMap<Class<?>, ConcurrentMap<Class<?>,
List<Converter>>> converterCache = new ConcurrentHashMap<>();
public ConverterUtil(FrameworkModel frameworkModel) {
this.frameworkModel = frameworkModel;
@@ -41,8 +42,8 @@ public class ConverterUtil {
* @see ExtensionLoader#getSupportedExtensionInstances()
*/
public Converter<?, ?> getConverter(Class<?> sourceType, Class<?>
targetType) {
- Map<Class<?>, List<Converter>> toTargetMap =
converterCache.computeIfAbsent(sourceType, (k) -> new ConcurrentHashMap<>());
- List<Converter> converters = toTargetMap.computeIfAbsent(targetType,
(k) -> frameworkModel.getExtensionLoader(Converter.class)
+ ConcurrentMap<Class<?>, List<Converter>> toTargetMap =
ConcurrentHashMapUtils.computeIfAbsent(converterCache, sourceType, (k) -> new
ConcurrentHashMap<>());
+ List<Converter> converters =
ConcurrentHashMapUtils.computeIfAbsent(toTargetMap, targetType, (k) ->
frameworkModel.getExtensionLoader(Converter.class)
.getSupportedExtensionInstances()
.stream()
.filter(converter -> converter.accept(sourceType, targetType))
diff --git
a/dubbo-common/src/main/java/org/apache/dubbo/common/logger/LoggerFactory.java
b/dubbo-common/src/main/java/org/apache/dubbo/common/logger/LoggerFactory.java
index 56b2ddb74c..c718ae9e79 100644
---
a/dubbo-common/src/main/java/org/apache/dubbo/common/logger/LoggerFactory.java
+++
b/dubbo-common/src/main/java/org/apache/dubbo/common/logger/LoggerFactory.java
@@ -23,6 +23,7 @@ import
org.apache.dubbo.common.logger.log4j2.Log4j2LoggerAdapter;
import org.apache.dubbo.common.logger.slf4j.Slf4jLoggerAdapter;
import org.apache.dubbo.common.logger.support.FailsafeErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.support.FailsafeLogger;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.rpc.model.FrameworkModel;
import java.io.File;
@@ -145,7 +146,7 @@ public class LoggerFactory {
* @return logger
*/
public static Logger getLogger(Class<?> key) {
- return LOGGERS.computeIfAbsent(key.getName(), name -> new
FailsafeLogger(loggerAdapter.getLogger(name)));
+ return ConcurrentHashMapUtils.computeIfAbsent(LOGGERS, key.getName(),
name -> new FailsafeLogger(loggerAdapter.getLogger(name)));
}
/**
@@ -155,7 +156,7 @@ public class LoggerFactory {
* @return logger provider
*/
public static Logger getLogger(String key) {
- return LOGGERS.computeIfAbsent(key, k -> new
FailsafeLogger(loggerAdapter.getLogger(k)));
+ return ConcurrentHashMapUtils.computeIfAbsent(LOGGERS, key, k -> new
FailsafeLogger(loggerAdapter.getLogger(k)));
}
/**
@@ -165,7 +166,7 @@ public class LoggerFactory {
* @return error type aware logger
*/
public static ErrorTypeAwareLogger getErrorTypeAwareLogger(Class<?> key) {
- return ERROR_TYPE_AWARE_LOGGERS.computeIfAbsent(key.getName(), name ->
new FailsafeErrorTypeAwareLogger(loggerAdapter.getLogger(name)));
+ return
ConcurrentHashMapUtils.computeIfAbsent(ERROR_TYPE_AWARE_LOGGERS, key.getName(),
name -> new FailsafeErrorTypeAwareLogger(loggerAdapter.getLogger(name)));
}
/**
@@ -175,7 +176,7 @@ public class LoggerFactory {
* @return error type aware logger
*/
public static ErrorTypeAwareLogger getErrorTypeAwareLogger(String key) {
- return ERROR_TYPE_AWARE_LOGGERS.computeIfAbsent(key, k -> new
FailsafeErrorTypeAwareLogger(loggerAdapter.getLogger(k)));
+ return
ConcurrentHashMapUtils.computeIfAbsent(ERROR_TYPE_AWARE_LOGGERS, key, k -> new
FailsafeErrorTypeAwareLogger(loggerAdapter.getLogger(k)));
}
/**
diff --git
a/dubbo-common/src/main/java/org/apache/dubbo/common/metrics/collector/stat/MetricsStatComposite.java
b/dubbo-common/src/main/java/org/apache/dubbo/common/metrics/collector/stat/MetricsStatComposite.java
index baed0ad4cc..dc75962b52 100644
---
a/dubbo-common/src/main/java/org/apache/dubbo/common/metrics/collector/stat/MetricsStatComposite.java
+++
b/dubbo-common/src/main/java/org/apache/dubbo/common/metrics/collector/stat/MetricsStatComposite.java
@@ -20,6 +20,7 @@ package org.apache.dubbo.common.metrics.collector.stat;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.LongAccumulator;
@@ -29,20 +30,22 @@ import org.apache.dubbo.common.metrics.event.RTEvent;
import org.apache.dubbo.common.metrics.event.RequestEvent;
import org.apache.dubbo.common.metrics.listener.MetricsListener;
import org.apache.dubbo.common.metrics.model.MethodMetric;
-public class MetricsStatComposite{
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
+
+public class MetricsStatComposite {
public Map<RequestEvent.Type, MetricsStatHandler> stats = new
ConcurrentHashMap<>();
- private final Map<MethodMetric, AtomicLong> lastRT = new
ConcurrentHashMap<>();
- private final Map<MethodMetric, LongAccumulator> minRT = new
ConcurrentHashMap<>();
- private final Map<MethodMetric, LongAccumulator> maxRT = new
ConcurrentHashMap<>();
- private final Map<MethodMetric, AtomicLong> avgRT = new
ConcurrentHashMap<>();
- private final Map<MethodMetric, AtomicLong> totalRT = new
ConcurrentHashMap<>();
- private final Map<MethodMetric, AtomicLong> rtCount = new
ConcurrentHashMap<>();
+ private final ConcurrentMap<MethodMetric, AtomicLong> lastRT = new
ConcurrentHashMap<>();
+ private final ConcurrentMap<MethodMetric, LongAccumulator> minRT = new
ConcurrentHashMap<>();
+ private final ConcurrentMap<MethodMetric, LongAccumulator> maxRT = new
ConcurrentHashMap<>();
+ private final ConcurrentMap<MethodMetric, AtomicLong> avgRT = new
ConcurrentHashMap<>();
+ private final ConcurrentMap<MethodMetric, AtomicLong> totalRT = new
ConcurrentHashMap<>();
+ private final ConcurrentMap<MethodMetric, AtomicLong> rtCount = new
ConcurrentHashMap<>();
private final String applicationName;
private final List<MetricsListener> listeners;
private DefaultMetricsCollector collector;
- public MetricsStatComposite(String applicationName,
DefaultMetricsCollector collector){
+ public MetricsStatComposite(String applicationName,
DefaultMetricsCollector collector) {
this.applicationName = applicationName;
this.listeners = collector.getListener();
this.collector = collector;
@@ -53,23 +56,27 @@ public class MetricsStatComposite{
return stats.get(statType);
}
- public Map<MethodMetric, AtomicLong> getLastRT(){
+ public Map<MethodMetric, AtomicLong> getLastRT() {
return this.lastRT;
}
- public Map<MethodMetric, LongAccumulator> getMinRT(){
+
+ public Map<MethodMetric, LongAccumulator> getMinRT() {
return this.minRT;
}
- public Map<MethodMetric, LongAccumulator> getMaxRT(){
+ public Map<MethodMetric, LongAccumulator> getMaxRT() {
return this.maxRT;
}
- public Map<MethodMetric, AtomicLong> getAvgRT(){
+
+ public Map<MethodMetric, AtomicLong> getAvgRT() {
return this.avgRT;
}
- public Map<MethodMetric, AtomicLong> getTotalRT(){
+
+ public Map<MethodMetric, AtomicLong> getTotalRT() {
return this.totalRT;
}
- public Map<MethodMetric, AtomicLong> getRtCount(){
+
+ public Map<MethodMetric, AtomicLong> getRtCount() {
return this.rtCount;
}
@@ -77,29 +84,29 @@ public class MetricsStatComposite{
if (collector.isCollectEnabled()) {
MethodMetric metric = new MethodMetric(applicationName,
interfaceName, methodName, group, version);
- AtomicLong last = lastRT.computeIfAbsent(metric, k -> new
AtomicLong());
+ AtomicLong last = ConcurrentHashMapUtils.computeIfAbsent(lastRT,
metric, k -> new AtomicLong());
last.set(responseTime);
- LongAccumulator min = minRT.computeIfAbsent(metric, k -> new
LongAccumulator(Long::min, Long.MAX_VALUE));
+ LongAccumulator min =
ConcurrentHashMapUtils.computeIfAbsent(minRT, metric, k -> new
LongAccumulator(Long::min, Long.MAX_VALUE));
min.accumulate(responseTime);
- LongAccumulator max = maxRT.computeIfAbsent(metric, k -> new
LongAccumulator(Long::max, Long.MIN_VALUE));
+ LongAccumulator max =
ConcurrentHashMapUtils.computeIfAbsent(maxRT, metric, k -> new
LongAccumulator(Long::max, Long.MIN_VALUE));
max.accumulate(responseTime);
- AtomicLong total = totalRT.computeIfAbsent(metric, k -> new
AtomicLong());
+ AtomicLong total = ConcurrentHashMapUtils.computeIfAbsent(totalRT,
metric, k -> new AtomicLong());
total.addAndGet(responseTime);
- AtomicLong count = rtCount.computeIfAbsent(metric, k -> new
AtomicLong());
+ AtomicLong count = ConcurrentHashMapUtils.computeIfAbsent(rtCount,
metric, k -> new AtomicLong());
count.incrementAndGet();
- avgRT.computeIfAbsent(metric, k -> new AtomicLong());
+ ConcurrentHashMapUtils.computeIfAbsent(avgRT, metric, k -> new
AtomicLong());
publishEvent(new RTEvent(metric, responseTime));
}
}
private void init() {
- stats.put(RequestEvent.Type.TOTAL, new
DefaultMetricsStatHandler(applicationName){
+ stats.put(RequestEvent.Type.TOTAL, new
DefaultMetricsStatHandler(applicationName) {
@Override
public void doNotify(MethodMetric metric) {
publishEvent(new RequestEvent(metric,
RequestEvent.Type.TOTAL));
diff --git
a/dubbo-common/src/main/java/org/apache/dubbo/common/store/support/SimpleDataStore.java
b/dubbo-common/src/main/java/org/apache/dubbo/common/store/support/SimpleDataStore.java
index e9709b0416..d511ebd513 100644
---
a/dubbo-common/src/main/java/org/apache/dubbo/common/store/support/SimpleDataStore.java
+++
b/dubbo-common/src/main/java/org/apache/dubbo/common/store/support/SimpleDataStore.java
@@ -18,6 +18,7 @@
package org.apache.dubbo.common.store.support;
import org.apache.dubbo.common.store.DataStore;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import java.util.HashMap;
import java.util.Map;
@@ -28,7 +29,7 @@ public class SimpleDataStore implements DataStore {
// <component name or id, <data-name, data-value>>
private ConcurrentMap<String, ConcurrentMap<String, Object>> data =
- new ConcurrentHashMap<String, ConcurrentMap<String, Object>>();
+ new ConcurrentHashMap<String, ConcurrentMap<String, Object>>();
@Override
public Map<String, Object> get(String componentName) {
@@ -50,7 +51,7 @@ public class SimpleDataStore implements DataStore {
@Override
public void put(String componentName, String key, Object value) {
- Map<String, Object> componentData =
data.computeIfAbsent(componentName, k -> new ConcurrentHashMap<>());
+ Map<String, Object> componentData =
ConcurrentHashMapUtils.computeIfAbsent(data, componentName, k -> new
ConcurrentHashMap<>());
componentData.put(key, value);
}
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 5abc993d01..fd2c9aacb2 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
@@ -23,6 +23,7 @@ import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.store.DataStore;
import org.apache.dubbo.common.threadpool.ThreadPool;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.ExecutorUtil;
import org.apache.dubbo.common.utils.NamedThreadFactory;
import org.apache.dubbo.common.utils.StringUtils;
@@ -93,14 +94,14 @@ public class DefaultExecutorRepository implements
ExecutorRepository, ExtensionA
@Override
public synchronized ExecutorService createExecutorIfAbsent(URL url) {
String executorKey = getExecutorKey(url);
- Map<String, ExecutorService> executors =
data.computeIfAbsent(executorKey, k -> new ConcurrentHashMap<>());
+ ConcurrentMap<String, ExecutorService> executors =
ConcurrentHashMapUtils.computeIfAbsent(data, executorKey, k -> new
ConcurrentHashMap<>());
String executorCacheKey = getExecutorSecondKey(url);
url = setThreadNameIfAbsent(url, executorCacheKey);
URL finalUrl = url;
- ExecutorService executor = executors.computeIfAbsent(executorCacheKey,
k -> createExecutor(finalUrl));
+ ExecutorService executor =
ConcurrentHashMapUtils.computeIfAbsent(executors, executorCacheKey, k ->
createExecutor(finalUrl));
// If executor has been shut down, create a new one
if (executor.isShutdown() || executor.isTerminated()) {
executors.remove(executorCacheKey);
diff --git
a/dubbo-common/src/main/java/org/apache/dubbo/common/url/component/ServiceConfigURL.java
b/dubbo-common/src/main/java/org/apache/dubbo/common/url/component/ServiceConfigURL.java
index 95bdb75110..9b73daf1c6 100644
---
a/dubbo-common/src/main/java/org/apache/dubbo/common/url/component/ServiceConfigURL.java
+++
b/dubbo-common/src/main/java/org/apache/dubbo/common/url/component/ServiceConfigURL.java
@@ -18,18 +18,20 @@ package org.apache.dubbo.common.url.component;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.utils.CollectionUtils;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.StringUtils;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
public class ServiceConfigURL extends URL {
- private volatile transient Map<String, URL> urls;
- private volatile transient Map<String, Number> numbers;
- private volatile transient Map<String, Map<String, Number>> methodNumbers;
+ private volatile transient ConcurrentMap<String, URL> urls;
+ private volatile transient ConcurrentMap<String, Number> numbers;
+ private volatile transient ConcurrentMap<String, Map<String, Number>>
methodNumbers;
private volatile transient String full;
private volatile transient String string;
private volatile transient String identity;
@@ -77,12 +79,12 @@ public class ServiceConfigURL extends URL {
}
public ServiceConfigURL(String protocol,
- String username,
- String password,
- String host,
- int port,
- String path,
- Map<String, String> parameters) {
+ String username,
+ String password,
+ String host,
+ int port,
+ String path,
+ Map<String, String> parameters) {
this(new PathURLAddress(protocol, username, password, path, host,
port), URLParam.parse(parameters), null);
}
@@ -551,11 +553,11 @@ public class ServiceConfigURL extends URL {
}
private void updateCachedNumber(String method, String key, Number n) {
- Map<String, Number> keyNumber =
getMethodNumbers().computeIfAbsent(method, m -> new HashMap<>());
+ Map<String, Number> keyNumber =
ConcurrentHashMapUtils.computeIfAbsent(getMethodNumbers(), method, m -> new
HashMap<>());
keyNumber.put(key, n);
}
- protected Map<String, Map<String, Number>> getMethodNumbers() {
+ protected ConcurrentMap<String, Map<String, Number>> getMethodNumbers() {
if (methodNumbers == null) { // concurrent initialization is tolerant
methodNumbers = new ConcurrentHashMap<>();
}
diff --git
a/dubbo-common/src/main/java/org/apache/dubbo/common/utils/ConcurrentHashMapUtils.java
b/dubbo-common/src/main/java/org/apache/dubbo/common/utils/ConcurrentHashMapUtils.java
new file mode 100644
index 0000000000..0b71d673d6
--- /dev/null
+++
b/dubbo-common/src/main/java/org/apache/dubbo/common/utils/ConcurrentHashMapUtils.java
@@ -0,0 +1,57 @@
+/*
+ * 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.common.utils;
+
+import java.util.concurrent.ConcurrentMap;
+import java.util.function.Function;
+
+/**
+ * ConcurrentHashMap util
+ */
+
+public class ConcurrentHashMapUtils {
+
+ private static boolean IS_JAVA8;
+
+ static {
+ try {
+ IS_JAVA8 = System.getProperty("java.version").startsWith("1.8.");
+ } catch (Exception ignore) {
+ // exception is ignored
+ IS_JAVA8 = true;
+ }
+ }
+
+ /**
+ * A temporary workaround for Java 8 ConcurrentHashMap#computeIfAbsent
specific performance issue: JDK-8161372.</br>
+ * @see <a
href="https://bugs.openjdk.java.net/browse/JDK-8161372";>https://bugs.openjdk.java.net/browse/JDK-8161372</a>
+ *
+ */
+ public static <K, V> V computeIfAbsent(ConcurrentMap<K, V> map, K key,
Function<? super K, ? extends V> func) {
+ if (IS_JAVA8) {
+ V v = map.get(key);
+ if (null == v) {
+ v = map.computeIfAbsent(key, func);
+ }
+ return v;
+ } else {
+ return map.computeIfAbsent(key, func);
+ }
+ }
+
+}
diff --git
a/dubbo-common/src/main/java/org/apache/dubbo/config/AbstractConfig.java
b/dubbo-common/src/main/java/org/apache/dubbo/config/AbstractConfig.java
index 413521cf91..6ee6307560 100644
--- a/dubbo-common/src/main/java/org/apache/dubbo/config/AbstractConfig.java
+++ b/dubbo-common/src/main/java/org/apache/dubbo/config/AbstractConfig.java
@@ -36,6 +36,7 @@ import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.dubbo.common.URL;
@@ -48,6 +49,7 @@ import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.ClassUtils;
import org.apache.dubbo.common.utils.CollectionUtils;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.FieldUtils;
import org.apache.dubbo.common.utils.MethodUtils;
import org.apache.dubbo.common.utils.ReflectUtils;
@@ -81,12 +83,12 @@ public abstract class AbstractConfig implements
Serializable {
/**
* tag name cache, speed up get tag name frequently
*/
- private static final Map<Class, String> tagNameCache = new
ConcurrentHashMap<>();
+ private static final ConcurrentMap<Class, String> tagNameCache = new
ConcurrentHashMap<>();
/**
* attributed getter method cache for equals(), hashCode() and toString()
*/
- private static final Map<Class, List<Method>> attributedMethodCache = new
ConcurrentHashMap<>();
+ private static final ConcurrentMap<Class, List<Method>>
attributedMethodCache = new ConcurrentHashMap<>();
/**
* The suffix container
@@ -122,7 +124,7 @@ public abstract class AbstractConfig implements
Serializable {
}
public static String getTagName(Class<?> cls) {
- return tagNameCache.computeIfAbsent(cls, (key) -> {
+ return ConcurrentHashMapUtils.computeIfAbsent(tagNameCache, cls, (key)
-> {
String tag = cls.getSimpleName();
for (String suffix : SUFFIXES) {
if (tag.endsWith(suffix)) {
@@ -1031,7 +1033,7 @@ public abstract class AbstractConfig implements
Serializable {
private List<Method> getAttributedMethods() {
Class<? extends AbstractConfig> cls = this.getClass();
- return attributedMethodCache.computeIfAbsent(cls, (key) ->
computeAttributedMethods());
+ return ConcurrentHashMapUtils.computeIfAbsent(attributedMethodCache,
cls, (key) -> computeAttributedMethods());
}
/**
diff --git
a/dubbo-common/src/main/java/org/apache/dubbo/rpc/model/FrameworkServiceRepository.java
b/dubbo-common/src/main/java/org/apache/dubbo/rpc/model/FrameworkServiceRepository.java
index d858fc1398..9f217ccaab 100644
---
a/dubbo-common/src/main/java/org/apache/dubbo/rpc/model/FrameworkServiceRepository.java
+++
b/dubbo-common/src/main/java/org/apache/dubbo/rpc/model/FrameworkServiceRepository.java
@@ -17,6 +17,7 @@
package org.apache.dubbo.rpc.model;
import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.StringUtils;
import java.util.ArrayList;
@@ -58,7 +59,7 @@ public class FrameworkServiceRepository {
// throw new IllegalStateException("Register duplicate provider
for key: " + key);
}
String keyWithoutGroup = keyWithoutGroup(key);
- providersWithoutGroup.computeIfAbsent(keyWithoutGroup, (k) -> new
CopyOnWriteArrayList<>()).add(providerModel);
+ ConcurrentHashMapUtils.computeIfAbsent(providersWithoutGroup,
keyWithoutGroup, (k) -> new CopyOnWriteArrayList<>()).add(providerModel);
}
public void unregisterProvider(ProviderModel providerModel) {
@@ -82,7 +83,7 @@ public class FrameworkServiceRepository {
}
public void registerProviderUrl(URL url) {
-
providerUrlsWithoutGroup.computeIfAbsent(keyWithoutGroup(url.getServiceKey()),
(k) -> new CopyOnWriteArrayList<>()).add(url);
+ ConcurrentHashMapUtils.computeIfAbsent(providerUrlsWithoutGroup,
keyWithoutGroup(url.getServiceKey()), (k) -> new
CopyOnWriteArrayList<>()).add(url);
}
public ProviderModel lookupExportedService(String serviceKey) {
diff --git
a/dubbo-common/src/main/java/org/apache/dubbo/rpc/model/ModuleServiceRepository.java
b/dubbo-common/src/main/java/org/apache/dubbo/rpc/model/ModuleServiceRepository.java
index 748b21deef..c3b1d446c7 100644
---
a/dubbo-common/src/main/java/org/apache/dubbo/rpc/model/ModuleServiceRepository.java
+++
b/dubbo-common/src/main/java/org/apache/dubbo/rpc/model/ModuleServiceRepository.java
@@ -17,6 +17,7 @@
package org.apache.dubbo.rpc.model;
import org.apache.dubbo.common.utils.CollectionUtils;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.config.ReferenceConfigBase;
import org.apache.dubbo.config.ServiceConfigBase;
@@ -81,7 +82,7 @@ public class ModuleServiceRepository {
}
public void registerConsumer(ConsumerModel consumerModel) {
- consumers.computeIfAbsent(consumerModel.getServiceKey(), (serviceKey)
-> new CopyOnWriteArrayList<>()).add(consumerModel);
+ ConcurrentHashMapUtils.computeIfAbsent(consumers,
consumerModel.getServiceKey(), (serviceKey) -> new
CopyOnWriteArrayList<>()).add(consumerModel);
}
/**
@@ -119,7 +120,7 @@ public class ModuleServiceRepository {
}
public ServiceDescriptor registerService(Class<?> interfaceClazz,
ServiceDescriptor serviceDescriptor) {
- List<ServiceDescriptor> serviceDescriptors =
services.computeIfAbsent(interfaceClazz.getName(),
+ List<ServiceDescriptor> serviceDescriptors =
ConcurrentHashMapUtils.computeIfAbsent(services, interfaceClazz.getName(),
k -> new CopyOnWriteArrayList<>());
synchronized (serviceDescriptors) {
Optional<ServiceDescriptor> previous = serviceDescriptors.stream()
@@ -149,7 +150,7 @@ public class ModuleServiceRepository {
ServiceDescriptor serviceDescriptor = registerService(interfaceClass);
// if path is different with interface name, add extra path mapping
if (!interfaceClass.getName().equals(path)) {
- List<ServiceDescriptor> serviceDescriptors =
services.computeIfAbsent(path,
+ List<ServiceDescriptor> serviceDescriptors =
ConcurrentHashMapUtils.computeIfAbsent(services, path,
_k -> new CopyOnWriteArrayList<>());
synchronized (serviceDescriptors) {
Optional<ServiceDescriptor> previous =
serviceDescriptors.stream()
@@ -179,7 +180,7 @@ public class ModuleServiceRepository {
public void reRegisterConsumer(String newServiceKey, String serviceKey) {
List<ConsumerModel> consumerModel = this.consumers.get(serviceKey);
consumerModel.forEach(c -> c.setServiceKey(newServiceKey));
- this.consumers.computeIfAbsent(newServiceKey, (k) -> new
CopyOnWriteArrayList<>()).addAll(consumerModel);
+ ConcurrentHashMapUtils.computeIfAbsent(this.consumers, newServiceKey,
(k) -> new CopyOnWriteArrayList<>()).addAll(consumerModel);
this.consumers.remove(serviceKey);
}
diff --git
a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/CacheableRouterFactory.java
b/dubbo-common/src/test/java/org/apache/dubbo/common/utils/ConcurrentHashMapUtilsTest.java
similarity index 54%
copy from
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/CacheableRouterFactory.java
copy to
dubbo-common/src/test/java/org/apache/dubbo/common/utils/ConcurrentHashMapUtilsTest.java
index 4d4b9eb5e0..3fc895497f 100644
---
a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/CacheableRouterFactory.java
+++
b/dubbo-common/src/test/java/org/apache/dubbo/common/utils/ConcurrentHashMapUtilsTest.java
@@ -14,24 +14,27 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.dubbo.rpc.cluster;
-import org.apache.dubbo.common.URL;
+package org.apache.dubbo.common.utils;
+
+import org.junit.jupiter.api.Test;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-/**
- * 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 CacheableRouterFactory implements RouterFactory {
- private ConcurrentMap<String, Router> routerMap = new
ConcurrentHashMap<>();
+import static org.junit.jupiter.api.Assertions.assertEquals;
- @Override
- public Router getRouter(URL url) {
- return routerMap.computeIfAbsent(url.getServiceKey(), k ->
createRouter(url));
- }
- protected abstract Router createRouter(URL url);
+class ConcurrentHashMapUtilsTest {
+
+ @Test
+ public void testComputeIfAbsent() {
+ ConcurrentHashMap<String, String> map = new ConcurrentHashMap<>();
+ String ifAbsent = ConcurrentHashMapUtils.computeIfAbsent(map, "mxsm",
k -> "mxsm");
+ assertEquals("mxsm", ifAbsent);
+ ifAbsent = ConcurrentHashMapUtils.computeIfAbsent(map, "mxsm", k ->
"mxsm1");
+ assertEquals("mxsm", ifAbsent);
+ map.remove("mxsm");
+ ifAbsent = ConcurrentHashMapUtils.computeIfAbsent(map, "mxsm", k ->
"mxsm1");
+ assertEquals("mxsm1", ifAbsent);
+ }
}
diff --git
a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/bootstrap/DubboBootstrap.java
b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/bootstrap/DubboBootstrap.java
index ea88fd99d7..8eda1a9845 100644
---
a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/bootstrap/DubboBootstrap.java
+++
b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/bootstrap/DubboBootstrap.java
@@ -25,6 +25,7 @@ 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.ConcurrentHashMapUtils;
import org.apache.dubbo.config.ApplicationConfig;
import org.apache.dubbo.config.ConfigCenterConfig;
import org.apache.dubbo.config.ConsumerConfig;
@@ -53,8 +54,8 @@ import org.apache.dubbo.rpc.model.FrameworkModel;
import org.apache.dubbo.rpc.model.ModuleModel;
import java.util.List;
-import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Condition;
@@ -80,7 +81,7 @@ public final class DubboBootstrap {
private static final Logger logger =
LoggerFactory.getLogger(DubboBootstrap.class);
- private static volatile Map<ApplicationModel, DubboBootstrap> instanceMap
= new ConcurrentHashMap<>();
+ private static volatile ConcurrentMap<ApplicationModel, DubboBootstrap>
instanceMap = new ConcurrentHashMap<>();
private static volatile DubboBootstrap instance;
private final AtomicBoolean awaited = new AtomicBoolean(false);
@@ -116,7 +117,7 @@ public final class DubboBootstrap {
}
public static DubboBootstrap getInstance(ApplicationModel
applicationModel) {
- return instanceMap.computeIfAbsent(applicationModel, _k -> new
DubboBootstrap(applicationModel));
+ return ConcurrentHashMapUtils.computeIfAbsent(instanceMap,
applicationModel, _k -> new DubboBootstrap(applicationModel));
}
public static DubboBootstrap newInstance() {
@@ -217,6 +218,7 @@ public final class DubboBootstrap {
/**
* Start dubbo application
+ *
* @param wait If true, wait for startup to complete, or else no waiting.
* @return
*/
@@ -234,6 +236,7 @@ public final class DubboBootstrap {
/**
* Start dubbo application but no wait for finish.
+ *
* @return the future object
*/
public Future asyncStart() {
@@ -242,6 +245,7 @@ public final class DubboBootstrap {
/**
* Stop dubbo application
+ *
* @return
* @throws IllegalStateException
*/
@@ -369,7 +373,7 @@ public final class DubboBootstrap {
return metadataReport(null, consumerBuilder);
}
- public DubboBootstrap metadataReport(String
id,Consumer<MetadataReportBuilder> consumerBuilder) {
+ public DubboBootstrap metadataReport(String id,
Consumer<MetadataReportBuilder> consumerBuilder) {
MetadataReportBuilder metadataReportBuilder =
createMetadataReportBuilder(id);
consumerBuilder.accept(metadataReportBuilder);
return this;
diff --git
a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/utils/SimpleReferenceCache.java
b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/utils/SimpleReferenceCache.java
index 65389f95b5..29d3514127 100644
---
a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/utils/SimpleReferenceCache.java
+++
b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/utils/SimpleReferenceCache.java
@@ -21,6 +21,7 @@ import org.apache.dubbo.common.config.ReferenceCache;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.CollectionUtils;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.config.ReferenceConfigBase;
import org.apache.dubbo.rpc.service.Destroyable;
@@ -70,8 +71,8 @@ public class SimpleReferenceCache implements ReferenceCache {
private final String name;
private final KeyGenerator generator;
- private final Map<String, List<ReferenceConfigBase<?>>> referenceKeyMap =
new ConcurrentHashMap<>();
- private final Map<Class<?>, List<ReferenceConfigBase<?>>> referenceTypeMap
= new ConcurrentHashMap<>();
+ private final ConcurrentMap<String, List<ReferenceConfigBase<?>>>
referenceKeyMap = new ConcurrentHashMap<>();
+ private final ConcurrentMap<Class<?>, List<ReferenceConfigBase<?>>>
referenceTypeMap = new ConcurrentHashMap<>();
private final Map<ReferenceConfigBase<?>, Object> references = new
ConcurrentHashMap<>();
protected SimpleReferenceCache(String name, KeyGenerator generator) {
@@ -104,7 +105,7 @@ public class SimpleReferenceCache implements ReferenceCache
{
* Create cache if not existed yet.
*/
public static SimpleReferenceCache getCache(String name, KeyGenerator
keyGenerator) {
- return CACHE_HOLDER.computeIfAbsent(name, k -> new
SimpleReferenceCache(k, keyGenerator));
+ return ConcurrentHashMapUtils.computeIfAbsent(CACHE_HOLDER, name, k ->
new SimpleReferenceCache(k, keyGenerator));
}
@Override
@@ -124,9 +125,9 @@ public class SimpleReferenceCache implements ReferenceCache
{
}
if (proxy == null) {
- List<ReferenceConfigBase<?>> referencesOfType =
referenceTypeMap.computeIfAbsent(type, _t -> Collections.synchronizedList(new
ArrayList<>()));
+ List<ReferenceConfigBase<?>> referencesOfType =
ConcurrentHashMapUtils.computeIfAbsent(referenceTypeMap, type, _t ->
Collections.synchronizedList(new ArrayList<>()));
referencesOfType.add(rc);
- List<ReferenceConfigBase<?>> referenceConfigList =
referenceKeyMap.computeIfAbsent(key, _k -> Collections.synchronizedList(new
ArrayList<>()));
+ List<ReferenceConfigBase<?>> referenceConfigList =
ConcurrentHashMapUtils.computeIfAbsent(referenceKeyMap, key, _k ->
Collections.synchronizedList(new ArrayList<>()));
referenceConfigList.add(rc);
proxy = rc.get();
}
@@ -262,7 +263,7 @@ public class SimpleReferenceCache implements ReferenceCache
{
private void destroyReference(ReferenceConfigBase<?> rc) {
Destroyable proxy = (Destroyable) rc.get();
- if (proxy != null){
+ if (proxy != null) {
proxy.$destroy();
}
rc.destroy();
diff --git
a/dubbo-config/dubbo-config-spring/src/main/java/org/apache/dubbo/config/spring/reference/ReferenceBeanManager.java
b/dubbo-config/dubbo-config-spring/src/main/java/org/apache/dubbo/config/spring/reference/ReferenceBeanManager.java
index 578accb7b9..dcef96eeea 100644
---
a/dubbo-config/dubbo-config-spring/src/main/java/org/apache/dubbo/config/spring/reference/ReferenceBeanManager.java
+++
b/dubbo-config/dubbo-config-spring/src/main/java/org/apache/dubbo/config/spring/reference/ReferenceBeanManager.java
@@ -19,6 +19,7 @@ package org.apache.dubbo.config.spring.reference;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.Assert;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.config.ReferenceConfig;
import org.apache.dubbo.config.spring.ReferenceBean;
@@ -37,6 +38,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import static
org.apache.dubbo.common.constants.LoggerCodeConstants.CONFIG_DUBBO_BEAN_INITIALIZER;
@@ -46,16 +48,16 @@ public class ReferenceBeanManager implements
ApplicationContextAware {
private final ErrorTypeAwareLogger logger =
LoggerFactory.getErrorTypeAwareLogger(getClass());
//reference key -> reference bean names
- private Map<String, List<String>> referenceKeyMap = new
ConcurrentHashMap<>();
+ private ConcurrentMap<String, List<String>> referenceKeyMap = new
ConcurrentHashMap<>();
// reference alias -> reference bean name
- private Map<String, String> referenceAliasMap = new ConcurrentHashMap<>();
+ private ConcurrentMap<String, String> referenceAliasMap = new
ConcurrentHashMap<>();
//reference bean name -> ReferenceBean
- private Map<String, ReferenceBean> referenceBeanMap = new
ConcurrentHashMap<>();
+ private ConcurrentMap<String, ReferenceBean> referenceBeanMap = new
ConcurrentHashMap<>();
//reference key -> ReferenceConfig instance
- private Map<String, ReferenceConfig> referenceConfigMap = new
ConcurrentHashMap<>();
+ private ConcurrentMap<String, ReferenceConfig> referenceConfigMap = new
ConcurrentHashMap<>();
private ApplicationContext applicationContext;
private volatile boolean initialized = false;
@@ -105,7 +107,7 @@ public class ReferenceBeanManager implements
ApplicationContextAware {
}
public void registerReferenceKeyAndBeanName(String referenceKey, String
referenceBeanNameOrAlias) {
- List<String> list = referenceKeyMap.computeIfAbsent(referenceKey,
(key) -> new ArrayList<>());
+ List<String> list =
ConcurrentHashMapUtils.computeIfAbsent(referenceKeyMap, referenceKey, (key) ->
new ArrayList<>());
if (!list.contains(referenceBeanNameOrAlias)) {
list.add(referenceBeanNameOrAlias);
// register bean name as alias
diff --git
a/dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/AbstractRegistryService.java
b/dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/AbstractRegistryService.java
index 9ac517c37b..4a7e85410e 100644
---
a/dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/AbstractRegistryService.java
+++
b/dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/AbstractRegistryService.java
@@ -20,6 +20,7 @@ import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.CollectionUtils;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.registry.NotifyListener;
import org.apache.dubbo.registry.RegistryService;
@@ -102,7 +103,7 @@ public abstract class AbstractRegistryService implements
RegistryService {
if (url == null) {
throw new IllegalArgumentException("url == null");
}
- List<URL> urls = registered.computeIfAbsent(service, k -> new
CopyOnWriteArrayList<>());
+ List<URL> urls = ConcurrentHashMapUtils.computeIfAbsent(registered,
service, k -> new CopyOnWriteArrayList<>());
if (!urls.contains(url)) {
urls.add(url);
}
@@ -162,7 +163,7 @@ public abstract class AbstractRegistryService implements
RegistryService {
if (listener == null) {
return;
}
- List<NotifyListener> listeners =
notifyListeners.computeIfAbsent(service, k -> new CopyOnWriteArrayList<>());
+ List<NotifyListener> listeners =
ConcurrentHashMapUtils.computeIfAbsent(notifyListeners, service, k -> new
CopyOnWriteArrayList<>());
if (!listeners.contains(listener)) {
listeners.add(listener);
}
@@ -202,7 +203,7 @@ public abstract class AbstractRegistryService implements
RegistryService {
protected final void notify(String service, List<URL> urls) {
if (StringUtils.isEmpty(service)
- || CollectionUtils.isEmpty(urls)) {
+ || CollectionUtils.isEmpty(urls)) {
return;
}
doNotify(service, urls);
diff --git
a/dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/SimpleRegistryService.java
b/dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/SimpleRegistryService.java
index f4168654a7..a8e8d74575 100644
---
a/dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/SimpleRegistryService.java
+++
b/dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/SimpleRegistryService.java
@@ -21,6 +21,7 @@ import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.url.component.ServiceConfigURL;
import org.apache.dubbo.common.utils.CollectionUtils;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.NetUtils;
import org.apache.dubbo.common.utils.UrlUtils;
import org.apache.dubbo.registry.NotifyListener;
@@ -46,7 +47,7 @@ public class SimpleRegistryService extends
AbstractRegistryService {
public void register(String service, URL url) {
super.register(service, url);
String client =
RpcContext.getServiceContext().getRemoteAddressString();
- Map<String, URL> urls = remoteRegistered.computeIfAbsent(client, k ->
new ConcurrentHashMap<>());
+ Map<String, URL> urls =
ConcurrentHashMapUtils.computeIfAbsent(remoteRegistered, client, k -> new
ConcurrentHashMap<>());
urls.put(service, url);
notify(service, getRegistered().get(service));
}
@@ -70,12 +71,12 @@ public class SimpleRegistryService extends
AbstractRegistryService {
}
List<URL> urls = getRegistered().get(service);
if ((RegistryService.class.getName() + ":0.0.0").equals(service)
- && CollectionUtils.isEmpty(urls)) {
+ && CollectionUtils.isEmpty(urls)) {
register(service, new ServiceConfigURL("dubbo",
- NetUtils.getLocalHost(),
- RpcContext.getServiceContext().getLocalPort(),
- RegistryService.class.getName(),
- url.getParameters()));
+ NetUtils.getLocalHost(),
+ RpcContext.getServiceContext().getLocalPort(),
+ RegistryService.class.getName(),
+ url.getParameters()));
List<String> rs = registries;
if (rs != null && rs.size() > 0) {
for (String registry : rs) {
@@ -85,7 +86,7 @@ public class SimpleRegistryService extends
AbstractRegistryService {
}
super.subscribe(service, url, listener);
- Map<String, NotifyListener> listeners =
remoteListeners.computeIfAbsent(client, k -> new ConcurrentHashMap<>());
+ Map<String, NotifyListener> listeners =
ConcurrentHashMapUtils.computeIfAbsent(remoteListeners, client, k -> new
ConcurrentHashMap<>());
listeners.put(service, listener);
urls = getRegistered().get(service);
if (urls != null && urls.size() > 0) {
@@ -125,9 +126,9 @@ public class SimpleRegistryService extends
AbstractRegistryService {
for (Map.Entry<String, NotifyListener> entry :
listeners.entrySet()) {
String service = entry.getKey();
super.unsubscribe(service, new ServiceConfigURL("subscribe",
- RpcContext.getServiceContext().getRemoteHost(),
- RpcContext.getServiceContext().getRemotePort(),
- RegistryService.class.getName(),
getSubscribed(service)), entry.getValue());
+ RpcContext.getServiceContext().getRemoteHost(),
+ RpcContext.getServiceContext().getRemotePort(),
+ RegistryService.class.getName(), getSubscribed(service)),
entry.getValue());
}
}
}
diff --git
a/dubbo-configcenter/dubbo-configcenter-nacos/src/main/java/org/apache/dubbo/configcenter/support/nacos/NacosDynamicConfiguration.java
b/dubbo-configcenter/dubbo-configcenter-nacos/src/main/java/org/apache/dubbo/configcenter/support/nacos/NacosDynamicConfiguration.java
index 5a09e03d06..55bd14b40c 100644
---
a/dubbo-configcenter/dubbo-configcenter-nacos/src/main/java/org/apache/dubbo/configcenter/support/nacos/NacosDynamicConfiguration.java
+++
b/dubbo-configcenter/dubbo-configcenter-nacos/src/main/java/org/apache/dubbo/configcenter/support/nacos/NacosDynamicConfiguration.java
@@ -21,6 +21,7 @@ import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.Executor;
@@ -33,6 +34,7 @@ import
org.apache.dubbo.common.config.configcenter.DynamicConfiguration;
import org.apache.dubbo.common.constants.LoggerCodeConstants;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.MD5Utils;
import org.apache.dubbo.common.utils.StringUtils;
@@ -81,7 +83,7 @@ public class NacosDynamicConfiguration implements
DynamicConfiguration {
/**
* The map store the key to {@link NacosConfigListener} mapping
*/
- private final Map<String, NacosConfigListener> watchListenerMap;
+ private final ConcurrentMap<String, NacosConfigListener> watchListenerMap;
private MD5Utils md5Utils = new MD5Utils();
@@ -209,8 +211,7 @@ public class NacosDynamicConfiguration implements
DynamicConfiguration {
@Override
public void addListener(String key, String group, ConfigurationListener
listener) {
String listenerKey = buildListenerKey(key, group);
- NacosConfigListener nacosConfigListener =
- watchListenerMap.computeIfAbsent(listenerKey, k ->
createTargetListener(key, group));
+ NacosConfigListener nacosConfigListener =
ConcurrentHashMapUtils.computeIfAbsent(watchListenerMap, listenerKey, k ->
createTargetListener(key, group));
nacosConfigListener.addListener(listener);
try {
configService.addListener(key, group, nacosConfigListener);
diff --git
a/dubbo-configcenter/dubbo-configcenter-zookeeper/src/main/java/org/apache/dubbo/configcenter/support/zookeeper/CacheListener.java
b/dubbo-configcenter/dubbo-configcenter-zookeeper/src/main/java/org/apache/dubbo/configcenter/support/zookeeper/CacheListener.java
index 7c2bdf21fb..4f565d0ae0 100644
---
a/dubbo-configcenter/dubbo-configcenter-zookeeper/src/main/java/org/apache/dubbo/configcenter/support/zookeeper/CacheListener.java
+++
b/dubbo-configcenter/dubbo-configcenter-zookeeper/src/main/java/org/apache/dubbo/configcenter/support/zookeeper/CacheListener.java
@@ -18,22 +18,24 @@ package org.apache.dubbo.configcenter.support.zookeeper;
import org.apache.dubbo.common.config.configcenter.ConfigurationListener;
import org.apache.dubbo.common.utils.CollectionUtils;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
/**
* one path has one zookeeperDataListener
*/
public class CacheListener {
- private Map<String, ZookeeperDataListener> pathKeyListeners = new
ConcurrentHashMap<>();
+ private ConcurrentMap<String, ZookeeperDataListener> pathKeyListeners =
new ConcurrentHashMap<>();
public CacheListener() {
}
public ZookeeperDataListener addListener(String pathKey,
ConfigurationListener configurationListener, String key, String group) {
- ZookeeperDataListener zookeeperDataListener =
pathKeyListeners.computeIfAbsent(pathKey,
+ ZookeeperDataListener zookeeperDataListener =
ConcurrentHashMapUtils.computeIfAbsent(pathKeyListeners, pathKey,
_pathKey -> new ZookeeperDataListener(_pathKey, key, group));
zookeeperDataListener.addListener(configurationListener);
return zookeeperDataListener;
diff --git
a/dubbo-metadata/dubbo-metadata-report-zookeeper/src/main/java/org/apache/dubbo/metadata/store/zookeeper/ZookeeperMetadataReport.java
b/dubbo-metadata/dubbo-metadata-report-zookeeper/src/main/java/org/apache/dubbo/metadata/store/zookeeper/ZookeeperMetadataReport.java
index 2bfa020093..4231966aaa 100644
---
a/dubbo-metadata/dubbo-metadata-report-zookeeper/src/main/java/org/apache/dubbo/metadata/store/zookeeper/ZookeeperMetadataReport.java
+++
b/dubbo-metadata/dubbo-metadata-report-zookeeper/src/main/java/org/apache/dubbo/metadata/store/zookeeper/ZookeeperMetadataReport.java
@@ -20,6 +20,7 @@ import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.config.configcenter.ConfigItem;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.JsonUtils;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.metadata.MappingChangedEvent;
@@ -45,6 +46,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import static org.apache.dubbo.common.constants.CommonConstants.PATH_SEPARATOR;
import static
org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_ZOOKEEPER_EXCEPTION;
@@ -62,7 +64,7 @@ public class ZookeeperMetadataReport extends
AbstractMetadataReport {
ZookeeperClient zkClient;
- private Map<String, MappingDataListener> casListenerMap = new
ConcurrentHashMap<>();
+ private ConcurrentMap<String, MappingDataListener> casListenerMap = new
ConcurrentHashMap<>();
public ZookeeperMetadataReport(URL url, ZookeeperTransporter
zookeeperTransporter) {
@@ -162,7 +164,7 @@ public class ZookeeperMetadataReport extends
AbstractMetadataReport {
@Override
public Set<String> getServiceAppMapping(String serviceKey, MappingListener
listener, URL url) {
String path = buildPathKey(DEFAULT_MAPPING_GROUP, serviceKey);
- MappingDataListener mappingDataListener =
casListenerMap.computeIfAbsent(path, _k -> {
+ MappingDataListener mappingDataListener =
ConcurrentHashMapUtils.computeIfAbsent(casListenerMap, path, _k -> {
MappingDataListener newMappingListener = new
MappingDataListener(serviceKey, path);
zkClient.addDataListener(path, newMappingListener);
return newMappingListener;
diff --git
a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/collector/AggregateMetricsCollector.java
b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/collector/AggregateMetricsCollector.java
index 1dfaab9ecd..1def3cdd60 100644
---
a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/collector/AggregateMetricsCollector.java
+++
b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/collector/AggregateMetricsCollector.java
@@ -27,6 +27,7 @@ import org.apache.dubbo.common.metrics.model.MethodMetric;
import org.apache.dubbo.common.metrics.model.MetricsKey;
import org.apache.dubbo.common.metrics.model.sample.GaugeMetricSample;
import org.apache.dubbo.common.metrics.model.sample.MetricSample;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.config.MetricsConfig;
import org.apache.dubbo.config.context.ConfigManager;
import org.apache.dubbo.config.nested.AggregationConfig;
@@ -36,8 +37,8 @@ import org.apache.dubbo.rpc.model.ApplicationModel;
import java.util.ArrayList;
import java.util.List;
-import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import static org.apache.dubbo.common.metrics.model.MetricsCategory.REQUESTS;
import static org.apache.dubbo.common.metrics.model.MetricsCategory.QPS;
@@ -51,15 +52,15 @@ public class AggregateMetricsCollector implements
MetricsCollector, MetricsListe
private int bucketNum;
private int timeWindowSeconds;
- private final Map<MethodMetric, TimeWindowCounter> totalRequests = new
ConcurrentHashMap<>();
- private final Map<MethodMetric, TimeWindowCounter> succeedRequests = new
ConcurrentHashMap<>();
- private final Map<MethodMetric, TimeWindowCounter> failedRequests = new
ConcurrentHashMap<>();
- private final Map<MethodMetric, TimeWindowCounter> businessFailedRequests
= new ConcurrentHashMap<>();
- private final Map<MethodMetric, TimeWindowCounter> timeoutRequests = new
ConcurrentHashMap<>();
- private final Map<MethodMetric, TimeWindowCounter> limitRequests = new
ConcurrentHashMap<>();
- private final Map<MethodMetric, TimeWindowCounter> totalFailedRequests =
new ConcurrentHashMap<>();
- private final Map<MethodMetric, TimeWindowCounter> qps = new
ConcurrentHashMap<>();
- private final Map<MethodMetric, TimeWindowQuantile> rt = new
ConcurrentHashMap<>();
+ private final ConcurrentMap<MethodMetric, TimeWindowCounter> totalRequests
= new ConcurrentHashMap<>();
+ private final ConcurrentMap<MethodMetric, TimeWindowCounter>
succeedRequests = new ConcurrentHashMap<>();
+ private final ConcurrentMap<MethodMetric, TimeWindowCounter>
failedRequests = new ConcurrentHashMap<>();
+ private final ConcurrentMap<MethodMetric, TimeWindowCounter>
businessFailedRequests = new ConcurrentHashMap<>();
+ private final ConcurrentMap<MethodMetric, TimeWindowCounter>
timeoutRequests = new ConcurrentHashMap<>();
+ private final ConcurrentMap<MethodMetric, TimeWindowCounter> limitRequests
= new ConcurrentHashMap<>();
+ private final ConcurrentMap<MethodMetric, TimeWindowCounter>
totalFailedRequests = new ConcurrentHashMap<>();
+ private final ConcurrentMap<MethodMetric, TimeWindowCounter> qps = new
ConcurrentHashMap<>();
+ private final ConcurrentMap<MethodMetric, TimeWindowQuantile> rt = new
ConcurrentHashMap<>();
private final ApplicationModel applicationModel;
@@ -97,7 +98,7 @@ public class AggregateMetricsCollector implements
MetricsCollector, MetricsListe
private void onRTEvent(RTEvent event) {
MethodMetric metric = (MethodMetric) event.getSource();
Long responseTime = event.getRt();
- TimeWindowQuantile quantile = rt.computeIfAbsent(metric, k -> new
TimeWindowQuantile(DEFAULT_COMPRESSION, bucketNum, timeWindowSeconds));
+ TimeWindowQuantile quantile =
ConcurrentHashMapUtils.computeIfAbsent(rt, metric, k -> new
TimeWindowQuantile(DEFAULT_COMPRESSION, bucketNum, timeWindowSeconds));
quantile.add(responseTime);
}
@@ -107,30 +108,30 @@ public class AggregateMetricsCollector implements
MetricsCollector, MetricsListe
TimeWindowCounter counter = null;
switch (type) {
case TOTAL:
- counter = totalRequests.computeIfAbsent(metric, k -> new
TimeWindowCounter(bucketNum, timeWindowSeconds));
- TimeWindowCounter qpsCounter = qps.computeIfAbsent(metric, k
-> new TimeWindowCounter(bucketNum, timeWindowSeconds));
+ counter =
ConcurrentHashMapUtils.computeIfAbsent(totalRequests, metric, k -> new
TimeWindowCounter(bucketNum, timeWindowSeconds));
+ TimeWindowCounter qpsCounter =
ConcurrentHashMapUtils.computeIfAbsent(qps, metric, k -> new
TimeWindowCounter(bucketNum, timeWindowSeconds));
qpsCounter.increment();
break;
case SUCCEED:
- counter = succeedRequests.computeIfAbsent(metric, k -> new
TimeWindowCounter(bucketNum, timeWindowSeconds));
+ counter =
ConcurrentHashMapUtils.computeIfAbsent(succeedRequests, metric, k -> new
TimeWindowCounter(bucketNum, timeWindowSeconds));
break;
case FAILED:
- counter = failedRequests.computeIfAbsent(metric, k -> new
TimeWindowCounter(bucketNum, timeWindowSeconds));
+ counter =
ConcurrentHashMapUtils.computeIfAbsent(failedRequests, metric, k -> new
TimeWindowCounter(bucketNum, timeWindowSeconds));
break;
case BUSINESS_FAILED:
- counter = businessFailedRequests.computeIfAbsent(metric, k ->
new TimeWindowCounter(bucketNum, timeWindowSeconds));
+ counter =
ConcurrentHashMapUtils.computeIfAbsent(businessFailedRequests, metric, k -> new
TimeWindowCounter(bucketNum, timeWindowSeconds));
break;
case REQUEST_TIMEOUT:
- counter = timeoutRequests.computeIfAbsent(metric, k -> new
TimeWindowCounter(bucketNum, timeWindowSeconds));
+ counter =
ConcurrentHashMapUtils.computeIfAbsent(timeoutRequests, metric, k -> new
TimeWindowCounter(bucketNum, timeWindowSeconds));
break;
case REQUEST_LIMIT:
- counter = limitRequests.computeIfAbsent(metric, k -> new
TimeWindowCounter(bucketNum, timeWindowSeconds));
+ counter =
ConcurrentHashMapUtils.computeIfAbsent(limitRequests, metric, k -> new
TimeWindowCounter(bucketNum, timeWindowSeconds));
break;
case TOTAL_FAILED:
- counter = totalFailedRequests.computeIfAbsent(metric, k -> new
TimeWindowCounter(bucketNum, timeWindowSeconds));
+ counter =
ConcurrentHashMapUtils.computeIfAbsent(totalFailedRequests, metric, k -> new
TimeWindowCounter(bucketNum, timeWindowSeconds));
break;
default:
diff --git
a/dubbo-monitor/dubbo-monitor-api/src/main/java/org/apache/dubbo/monitor/support/MonitorFilter.java
b/dubbo-monitor/dubbo-monitor-api/src/main/java/org/apache/dubbo/monitor/support/MonitorFilter.java
index 61b80f3c9e..7e50cc800b 100644
---
a/dubbo-monitor/dubbo-monitor-api/src/main/java/org/apache/dubbo/monitor/support/MonitorFilter.java
+++
b/dubbo-monitor/dubbo-monitor-api/src/main/java/org/apache/dubbo/monitor/support/MonitorFilter.java
@@ -21,6 +21,7 @@ import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.url.component.ServiceConfigURL;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.NetUtils;
import org.apache.dubbo.monitor.Monitor;
import org.apache.dubbo.monitor.MonitorFactory;
@@ -109,7 +110,7 @@ public class MonitorFilter implements Filter,
Filter.Listener {
*/
private AtomicInteger getConcurrent(Invoker<?> invoker, Invocation
invocation) {
String key = invoker.getInterface().getName() + "." +
invocation.getMethodName();
- return concurrents.computeIfAbsent(key, k -> new AtomicInteger());
+ return ConcurrentHashMapUtils.computeIfAbsent(concurrents, key, k ->
new AtomicInteger());
}
@Override
diff --git
a/dubbo-monitor/dubbo-monitor-default/src/main/java/org/apache/dubbo/monitor/dubbo/DubboMonitor.java
b/dubbo-monitor/dubbo-monitor-default/src/main/java/org/apache/dubbo/monitor/dubbo/DubboMonitor.java
index e0a7d9c8b2..e66fd5f485 100644
---
a/dubbo-monitor/dubbo-monitor-default/src/main/java/org/apache/dubbo/monitor/dubbo/DubboMonitor.java
+++
b/dubbo-monitor/dubbo-monitor-default/src/main/java/org/apache/dubbo/monitor/dubbo/DubboMonitor.java
@@ -20,6 +20,7 @@ import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.threadpool.manager.FrameworkExecutorRepository;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.ExecutorUtil;
import org.apache.dubbo.monitor.Monitor;
import org.apache.dubbo.monitor.MonitorService;
@@ -152,7 +153,7 @@ public class DubboMonitor implements Monitor {
int concurrent = url.getParameter(CONCURRENT_KEY, 0);
// init atomic reference
Statistics statistics = new Statistics(url);
- AtomicReference<StatisticsItem> reference =
statisticsMap.computeIfAbsent(statistics, k -> new AtomicReference<>());
+ AtomicReference<StatisticsItem> reference =
ConcurrentHashMapUtils.computeIfAbsent(statisticsMap, statistics, k -> new
AtomicReference<>());
// use CompareAndSet to sum
StatisticsItem current;
StatisticsItem update = new StatisticsItem();
diff --git
a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/AbstractServiceDiscoveryFactory.java
b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/AbstractServiceDiscoveryFactory.java
index a2d74d6037..57b4213433 100644
---
a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/AbstractServiceDiscoveryFactory.java
+++
b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/AbstractServiceDiscoveryFactory.java
@@ -17,6 +17,7 @@
package org.apache.dubbo.registry.client;
import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.rpc.model.ApplicationModel;
import org.apache.dubbo.rpc.model.ScopeModelAware;
@@ -50,7 +51,7 @@ public abstract class AbstractServiceDiscoveryFactory
implements ServiceDiscover
@Override
public ServiceDiscovery getServiceDiscovery(URL registryURL) {
String key = registryURL.toServiceStringWithoutResolving();
- return discoveries.computeIfAbsent(key, k ->
createDiscovery(registryURL));
+ return ConcurrentHashMapUtils.computeIfAbsent(discoveries, key, k ->
createDiscovery(registryURL));
}
protected abstract ServiceDiscovery createDiscovery(URL registryURL);
diff --git
a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ReflectionBasedServiceDiscovery.java
b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ReflectionBasedServiceDiscovery.java
index a35cc893e0..61537f88c3 100644
---
a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ReflectionBasedServiceDiscovery.java
+++
b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ReflectionBasedServiceDiscovery.java
@@ -33,6 +33,7 @@ import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.JsonUtils;
import org.apache.dubbo.common.utils.NamedThreadFactory;
import org.apache.dubbo.common.utils.NetUtils;
@@ -261,7 +262,7 @@ public class ReflectionBasedServiceDiscovery extends
AbstractServiceDiscovery {
}
private synchronized MetadataService
getMetadataServiceProxy(ServiceInstance instance) {
- return metadataServiceProxies.computeIfAbsent(computeKey(instance), k
-> MetadataUtils.referProxy(instance).getProxy());
+ return ConcurrentHashMapUtils.computeIfAbsent(metadataServiceProxies,
computeKey(instance), k -> MetadataUtils.referProxy(instance).getProxy());
}
private synchronized void destroyMetadataServiceProxy(ServiceInstance
instance) {
diff --git
a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistry.java
b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistry.java
index 0347cf369c..88c04eadb9 100644
---
a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistry.java
+++
b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistry.java
@@ -20,6 +20,7 @@ import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.CollectionUtils;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.metadata.AbstractServiceNameMapping;
import org.apache.dubbo.metadata.MappingChangedEvent;
import org.apache.dubbo.metadata.MappingListener;
@@ -429,7 +430,7 @@ public class ServiceDiscoveryRegistry extends
FailbackRegistry {
}
public Lock getAppSubscription(String key) {
- return appSubscriptionLocks.computeIfAbsent(key, _k -> new
ReentrantLock());
+ return ConcurrentHashMapUtils.computeIfAbsent(appSubscriptionLocks,
key, _k -> new ReentrantLock());
}
public void removeAppSubscriptionLock(String key) {
diff --git
a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistryDirectory.java
b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistryDirectory.java
index 8cc86720cc..c39176bfc6 100644
---
a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistryDirectory.java
+++
b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistryDirectory.java
@@ -26,6 +26,7 @@ import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import java.util.stream.Collectors;
import org.apache.dubbo.common.ProtocolServiceKey;
@@ -37,6 +38,7 @@ import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.Assert;
import org.apache.dubbo.common.utils.CollectionUtils;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.NetUtils;
import org.apache.dubbo.metadata.MetadataInfo;
import org.apache.dubbo.registry.AddressListener;
@@ -86,7 +88,7 @@ public class ServiceDiscoveryRegistryDirectory<T> extends
DynamicDirectory<T> {
private final Set<String> providerFirstParams;
private final ModuleModel moduleModel;
private final ProtocolServiceKey consumerProtocolServiceKey;
- private final Map<ProtocolServiceKey, URL> customizedConsumerUrlMap = new
ConcurrentHashMap<>();
+ private final ConcurrentMap<ProtocolServiceKey, URL>
customizedConsumerUrlMap = new ConcurrentHashMap<>();
public ServiceDiscoveryRegistryDirectory(Class<T> serviceType, URL url) {
super(serviceType, url);
@@ -351,7 +353,7 @@ public class ServiceDiscoveryRegistryDirectory<T> extends
DynamicDirectory<T> {
}
if (enabled) {
if (shouldWrap) {
- URL newConsumerUrl =
customizedConsumerUrlMap.computeIfAbsent(matchedProtocolServiceKey,
+ URL newConsumerUrl =
ConcurrentHashMapUtils.computeIfAbsent(customizedConsumerUrlMap,
matchedProtocolServiceKey,
k ->
consumerUrl.setProtocol(k.getProtocol())
.addParameter(CommonConstants.GROUP_KEY, k.getGroup())
.addParameter(CommonConstants.VERSION_KEY, k.getVersion()));
diff --git
a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/migration/DefaultMigrationAddressComparator.java
b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/migration/DefaultMigrationAddressComparator.java
index 8316bdc080..b02f5209ec 100644
---
a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/migration/DefaultMigrationAddressComparator.java
+++
b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/migration/DefaultMigrationAddressComparator.java
@@ -20,6 +20,7 @@ import org.apache.dubbo.common.config.ConfigurationUtils;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.CollectionUtils;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.registry.client.migration.model.MigrationRule;
import org.apache.dubbo.rpc.Invoker;
@@ -28,6 +29,7 @@ import org.apache.dubbo.rpc.cluster.ClusterInvoker;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import static
org.apache.dubbo.common.constants.LoggerCodeConstants.COMMON_PROPERTY_TYPE_MISMATCH;
@@ -40,11 +42,11 @@ public class DefaultMigrationAddressComparator implements
MigrationAddressCompar
public static final String OLD_ADDRESS_SIZE = "OLD_ADDRESS_SIZE";
public static final String NEW_ADDRESS_SIZE = "NEW_ADDRESS_SIZE";
- private Map<String, Map<String, Integer>> serviceMigrationData = new
ConcurrentHashMap<>();
+ private ConcurrentMap<String, Map<String, Integer>> serviceMigrationData =
new ConcurrentHashMap<>();
@Override
public <T> boolean shouldMigrate(ClusterInvoker<T> newInvoker,
ClusterInvoker<T> oldInvoker, MigrationRule rule) {
- Map<String, Integer> migrationData =
serviceMigrationData.computeIfAbsent(oldInvoker.getUrl().getDisplayServiceKey(),
_k -> new ConcurrentHashMap<>());
+ Map<String, Integer> migrationData =
ConcurrentHashMapUtils.computeIfAbsent(serviceMigrationData,
oldInvoker.getUrl().getDisplayServiceKey(), _k -> new ConcurrentHashMap<>());
if (!newInvoker.hasProxyInvokers()) {
migrationData.put(OLD_ADDRESS_SIZE, getAddressSize(oldInvoker));
diff --git
a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/migration/MigrationRuleListener.java
b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/migration/MigrationRuleListener.java
index 764b0c0d33..bc5c5a4876 100644
---
a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/migration/MigrationRuleListener.java
+++
b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/migration/MigrationRuleListener.java
@@ -26,6 +26,7 @@ import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.threadpool.manager.FrameworkExecutorRepository;
import org.apache.dubbo.common.utils.CollectionUtils;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.NamedThreadFactory;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.registry.client.migration.model.MigrationRule;
@@ -40,6 +41,7 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -73,7 +75,7 @@ public class MigrationRuleListener implements
RegistryProtocolListener, Configur
private static final int MIGRATION_DEFAULT_DELAY_TIME = 60000;
private String ruleKey;
- protected final Map<MigrationInvoker, MigrationRuleHandler> handlers = new
ConcurrentHashMap<>();
+ protected final ConcurrentMap<MigrationInvoker, MigrationRuleHandler>
handlers = new ConcurrentHashMap<>();
protected final LinkedBlockingQueue<String> ruleQueue = new
LinkedBlockingQueue<>();
private final AtomicBoolean executorSubmit = new AtomicBoolean(false);
@@ -239,7 +241,7 @@ public class MigrationRuleListener implements
RegistryProtocolListener, Configur
@Override
public void onRefer(RegistryProtocol registryProtocol, ClusterInvoker<?>
invoker, URL consumerUrl, URL registryURL) {
- MigrationRuleHandler<?> migrationRuleHandler =
handlers.computeIfAbsent((MigrationInvoker<?>) invoker, _key -> {
+ MigrationRuleHandler<?> migrationRuleHandler =
ConcurrentHashMapUtils.computeIfAbsent(handlers, (MigrationInvoker<?>) invoker,
_key -> {
((MigrationInvoker<?>) invoker).setMigrationRuleListener(this);
return new MigrationRuleHandler<>((MigrationInvoker<?>) invoker,
consumerUrl);
});
diff --git
a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/support/CacheableFailbackRegistry.java
b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/support/CacheableFailbackRegistry.java
index a31b2e375d..fb77e17d62 100644
---
a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/support/CacheableFailbackRegistry.java
+++
b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/support/CacheableFailbackRegistry.java
@@ -28,6 +28,7 @@ import
org.apache.dubbo.common.url.component.ServiceAddressURL;
import org.apache.dubbo.common.url.component.URLAddress;
import org.apache.dubbo.common.url.component.URLParam;
import org.apache.dubbo.common.utils.CollectionUtils;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.common.utils.UrlUtils;
import org.apache.dubbo.registry.NotifyListener;
@@ -42,6 +43,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
@@ -83,8 +85,8 @@ public abstract class CacheableFailbackRegistry extends
FailbackRegistry {
private static String[] VARIABLE_KEYS = new
String[]{ENCODED_TIMESTAMP_KEY, ENCODED_PID_KEY};
- protected Map<String, URLAddress> stringAddress = new
ConcurrentHashMap<>();
- protected Map<String, URLParam> stringParam = new ConcurrentHashMap<>();
+ protected ConcurrentMap<String, URLAddress> stringAddress = new
ConcurrentHashMap<>();
+ protected ConcurrentMap<String, URLParam> stringParam = new
ConcurrentHashMap<>();
private ScheduledExecutorService cacheRemovalScheduler;
private int cacheRemovalTaskIntervalInMillis;
@@ -249,8 +251,8 @@ public abstract class CacheableFailbackRegistry extends
FailbackRegistry {
/**
* Create DubboServiceAddress object using provider url, consumer url, and
extra parameters.
*
- * @param rawProvider provider url string
- * @param consumerURL URL object of consumer
+ * @param rawProvider provider url string
+ * @param consumerURL URL object of consumer
* @param extraParameters extra parameters
* @return created DubboServiceAddressURL object
*/
@@ -286,10 +288,10 @@ public abstract class CacheableFailbackRegistry extends
FailbackRegistry {
boolean isEncoded = encoded;
// PathURLAddress if it's using dubbo protocol.
- URLAddress address = stringAddress.computeIfAbsent(rawAddress, k ->
URLAddress.parse(k, getDefaultURLProtocol(), isEncoded));
+ URLAddress address =
ConcurrentHashMapUtils.computeIfAbsent(stringAddress, rawAddress, k ->
URLAddress.parse(k, getDefaultURLProtocol(), isEncoded));
address.setTimestamp(System.currentTimeMillis());
- URLParam param = stringParam.computeIfAbsent(rawParams, k ->
URLParam.parse(k, isEncoded, extraParameters));
+ URLParam param = ConcurrentHashMapUtils.computeIfAbsent(stringParam,
rawParams, k -> URLParam.parse(k, isEncoded, extraParameters));
param.setTimestamp(System.currentTimeMillis());
// create service URL using cached address, param, and consumer URL.
diff --git
a/dubbo-registry/dubbo-registry-multiple/src/main/java/org/apache/dubbo/registry/multiple/MultipleServiceDiscovery.java
b/dubbo-registry/dubbo-registry-multiple/src/main/java/org/apache/dubbo/registry/multiple/MultipleServiceDiscovery.java
index 6d02ad195b..fcb6a168ea 100644
---
a/dubbo-registry/dubbo-registry-multiple/src/main/java/org/apache/dubbo/registry/multiple/MultipleServiceDiscovery.java
+++
b/dubbo-registry/dubbo-registry-multiple/src/main/java/org/apache/dubbo/registry/multiple/MultipleServiceDiscovery.java
@@ -22,10 +22,12 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import java.util.function.Function;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.constants.CommonConstants;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.metadata.MetadataInfo;
import org.apache.dubbo.registry.NotifyListener;
import org.apache.dubbo.registry.client.ServiceDiscovery;
@@ -178,7 +180,7 @@ public class MultipleServiceDiscovery implements
ServiceDiscovery {
}
protected static class MultiServiceInstancesChangedListener extends
ServiceInstancesChangedListener {
- private final Map<String, SingleServiceInstancesChangedListener>
singleListenerMap;
+ private final ConcurrentMap<String,
SingleServiceInstancesChangedListener> singleListenerMap;
public MultiServiceInstancesChangedListener(Set<String> serviceNames,
ServiceDiscovery serviceDiscovery) {
super(serviceNames, serviceDiscovery);
@@ -207,7 +209,7 @@ public class MultipleServiceDiscovery implements
ServiceDiscovery {
public SingleServiceInstancesChangedListener
getAndComputeIfAbsent(String registryKey,
Function<String, SingleServiceInstancesChangedListener> func) {
- return singleListenerMap.computeIfAbsent(registryKey, func);
+ return ConcurrentHashMapUtils.computeIfAbsent(singleListenerMap,
registryKey, func);
}
}
diff --git
a/dubbo-registry/dubbo-registry-nacos/src/main/java/org/apache/dubbo/registry/nacos/NacosNamingServiceWrapper.java
b/dubbo-registry/dubbo-registry-nacos/src/main/java/org/apache/dubbo/registry/nacos/NacosNamingServiceWrapper.java
index 32d52a6ea5..3bd6116d15 100644
---
a/dubbo-registry/dubbo-registry-nacos/src/main/java/org/apache/dubbo/registry/nacos/NacosNamingServiceWrapper.java
+++
b/dubbo-registry/dubbo-registry-nacos/src/main/java/org/apache/dubbo/registry/nacos/NacosNamingServiceWrapper.java
@@ -23,6 +23,7 @@ import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;
@@ -30,6 +31,7 @@ import java.util.stream.Collectors;
import org.apache.dubbo.common.constants.LoggerCodeConstants;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.MethodUtils;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.registry.nacos.function.NacosConsumer;
@@ -56,8 +58,8 @@ public class NacosNamingServiceWrapper {
private final boolean isSupportBatchRegister;
- private final Map<InstanceId, InstancesInfo> registerStatus = new
ConcurrentHashMap<>();
- private final Map<SubscribeInfo, NamingService> subscribeStatus = new
ConcurrentHashMap<>();
+ private final ConcurrentMap<InstanceId, InstancesInfo> registerStatus =
new ConcurrentHashMap<>();
+ private final ConcurrentMap<SubscribeInfo, NamingService> subscribeStatus
= new ConcurrentHashMap<>();
private final Lock mapLock = new ReentrantLock();
public NacosNamingServiceWrapper(NacosConnectionManager
nacosConnectionManager, int retryTimes, int sleepMsBetweenRetries) {
@@ -86,7 +88,7 @@ public class NacosNamingServiceWrapper {
public void subscribe(String serviceName, String group, EventListener
eventListener) throws NacosException {
String nacosServiceName = handleInnerSymbol(serviceName);
SubscribeInfo subscribeInfo = new SubscribeInfo(nacosServiceName,
group, eventListener);
- NamingService namingService =
subscribeStatus.computeIfAbsent(subscribeInfo, info ->
nacosConnectionManager.getNamingService());
+ NamingService namingService =
ConcurrentHashMapUtils.computeIfAbsent(subscribeStatus, subscribeInfo, info ->
nacosConnectionManager.getNamingService());
accept(() -> namingService.subscribe(nacosServiceName, group,
eventListener));
}
@@ -109,7 +111,7 @@ public class NacosNamingServiceWrapper {
InstancesInfo instancesInfo;
try {
mapLock.lock();
- instancesInfo = registerStatus.computeIfAbsent(new
InstanceId(nacosServiceName, group), id -> new InstancesInfo());
+ instancesInfo =
ConcurrentHashMapUtils.computeIfAbsent(registerStatus, new
InstanceId(nacosServiceName, group), id -> new InstancesInfo());
} finally {
mapLock.unlock();
}
@@ -177,7 +179,7 @@ public class NacosNamingServiceWrapper {
InstancesInfo instancesInfo;
try {
mapLock.lock();
- instancesInfo = registerStatus.computeIfAbsent(new
InstanceId(nacosServiceName, group), id -> new InstancesInfo());
+ instancesInfo =
ConcurrentHashMapUtils.computeIfAbsent(registerStatus, new
InstanceId(nacosServiceName, group), id -> new InstancesInfo());
} finally {
mapLock.unlock();
}
@@ -223,7 +225,7 @@ public class NacosNamingServiceWrapper {
InstancesInfo instancesInfo;
try {
mapLock.lock();
- instancesInfo = registerStatus.computeIfAbsent(new
InstanceId(nacosServiceName, group), id -> new InstancesInfo());
+ instancesInfo =
ConcurrentHashMapUtils.computeIfAbsent(registerStatus, new
InstanceId(nacosServiceName, group), id -> new InstancesInfo());
} finally {
mapLock.unlock();
}
@@ -250,7 +252,7 @@ public class NacosNamingServiceWrapper {
InstancesInfo instancesInfo;
try {
mapLock.lock();
- instancesInfo = registerStatus.computeIfAbsent(new
InstanceId(nacosServiceName, group), id -> new InstancesInfo());
+ instancesInfo =
ConcurrentHashMapUtils.computeIfAbsent(registerStatus, new
InstanceId(nacosServiceName, group), id -> new InstancesInfo());
} finally {
mapLock.unlock();
}
diff --git
a/dubbo-registry/dubbo-registry-zookeeper/src/main/java/org/apache/dubbo/registry/zookeeper/ZookeeperRegistry.java
b/dubbo-registry/dubbo-registry-zookeeper/src/main/java/org/apache/dubbo/registry/zookeeper/ZookeeperRegistry.java
index 9614772433..8087914b33 100644
---
a/dubbo-registry/dubbo-registry-zookeeper/src/main/java/org/apache/dubbo/registry/zookeeper/ZookeeperRegistry.java
+++
b/dubbo-registry/dubbo-registry-zookeeper/src/main/java/org/apache/dubbo/registry/zookeeper/ZookeeperRegistry.java
@@ -21,6 +21,7 @@ import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.CollectionUtils;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.ConcurrentHashSet;
import org.apache.dubbo.common.utils.UrlUtils;
import org.apache.dubbo.registry.NotifyListener;
@@ -180,9 +181,9 @@ public class ZookeeperRegistry extends
CacheableFailbackRegistry {
if (ANY_VALUE.equals(url.getServiceInterface())) {
String root = toRootPath();
boolean check = url.getParameter(CHECK_KEY, false);
- ConcurrentMap<NotifyListener, ChildListener> listeners =
zkListeners.computeIfAbsent(url, k -> new ConcurrentHashMap<>());
+ ConcurrentMap<NotifyListener, ChildListener> listeners =
ConcurrentHashMapUtils.computeIfAbsent(zkListeners, url, k -> new
ConcurrentHashMap<>());
- ChildListener zkListener = listeners.computeIfAbsent(listener,
k -> (parentPath, currentChildren) -> {
+ ChildListener zkListener =
ConcurrentHashMapUtils.computeIfAbsent(listeners, listener, k -> (parentPath,
currentChildren) -> {
for (String child : currentChildren) {
child = URL.decode(child);
if (!anyServices.contains(child)) {
@@ -219,8 +220,8 @@ public class ZookeeperRegistry extends
CacheableFailbackRegistry {
/dubbo/[service name]/routers
*/
for (String path : toCategoriesPath(url)) {
- ConcurrentMap<NotifyListener, ChildListener> listeners
= zkListeners.computeIfAbsent(url, k -> new ConcurrentHashMap<>());
- ChildListener zkListener =
listeners.computeIfAbsent(listener, k -> new RegistryChildListenerImpl(url, k,
latch));
+ ConcurrentMap<NotifyListener, ChildListener> listeners
= ConcurrentHashMapUtils.computeIfAbsent(zkListeners, url, k -> new
ConcurrentHashMap<>());
+ ChildListener zkListener =
ConcurrentHashMapUtils.computeIfAbsent(listeners, listener, k -> new
RegistryChildListenerImpl(url, k, latch));
if (zkListener instanceof RegistryChildListenerImpl) {
((RegistryChildListenerImpl)
zkListener).setLatch(latch);
diff --git
a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/connection/MultiplexProtocolConnectionManager.java
b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/connection/MultiplexProtocolConnectionManager.java
index a1b60eb3be..25ef6fa009 100644
---
a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/connection/MultiplexProtocolConnectionManager.java
+++
b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/connection/MultiplexProtocolConnectionManager.java
@@ -17,6 +17,7 @@
package org.apache.dubbo.remoting.api.connection;
import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.remoting.ChannelHandler;
import org.apache.dubbo.rpc.model.FrameworkModel;
@@ -37,7 +38,7 @@ public class MultiplexProtocolConnectionManager implements
ConnectionManager {
@Override
public AbstractConnectionClient connect(URL url, ChannelHandler handler) {
- final ConnectionManager manager =
protocols.computeIfAbsent(url.getProtocol(),
this::createSingleProtocolConnectionManager);
+ final ConnectionManager manager =
ConcurrentHashMapUtils.computeIfAbsent(protocols, url.getProtocol(),
this::createSingleProtocolConnectionManager);
return manager.connect(url, handler);
}
diff --git
a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/PortUnificationExchanger.java
b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/PortUnificationExchanger.java
index 3b1f9e7a82..26252adc98 100644
---
a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/PortUnificationExchanger.java
+++
b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/PortUnificationExchanger.java
@@ -19,6 +19,7 @@ package org.apache.dubbo.remoting.exchange;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.remoting.ChannelHandler;
import org.apache.dubbo.remoting.RemotingException;
import org.apache.dubbo.remoting.RemotingServer;
@@ -38,7 +39,7 @@ public class PortUnificationExchanger {
private static final ConcurrentMap<String, RemotingServer> servers = new
ConcurrentHashMap<>();
public static RemotingServer bind(URL url, ChannelHandler handler) {
- servers.computeIfAbsent(url.getAddress(), addr -> {
+ ConcurrentHashMapUtils.computeIfAbsent(servers, url.getAddress(), addr
-> {
final AbstractPortUnificationServer server;
try {
server = getTransporter(url).bind(url, handler);
@@ -85,7 +86,7 @@ public class PortUnificationExchanger {
public static PortUnificationTransporter getTransporter(URL url) {
return
url.getOrDefaultFrameworkModel().getExtensionLoader(PortUnificationTransporter.class)
- .getAdaptiveExtension();
+ .getAdaptiveExtension();
}
}
diff --git
a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/transport/CodecSupport.java
b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/transport/CodecSupport.java
index 0c7861e34a..0f0d079927 100644
---
a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/transport/CodecSupport.java
+++
b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/transport/CodecSupport.java
@@ -25,6 +25,7 @@ import org.apache.dubbo.common.serialize.ObjectInput;
import org.apache.dubbo.common.serialize.ObjectOutput;
import org.apache.dubbo.common.serialize.Serialization;
import org.apache.dubbo.common.utils.CollectionUtils;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.remoting.utils.UrlUtils;
import org.apache.dubbo.rpc.model.FrameworkModel;
import org.apache.dubbo.rpc.model.FrameworkServiceRepository;
@@ -38,6 +39,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import static org.apache.dubbo.common.BaseServiceMetadata.keyWithoutGroup;
import static
org.apache.dubbo.common.constants.LoggerCodeConstants.TRANSPORT_FAILED_SERIALIZATION;
@@ -48,7 +50,7 @@ public class CodecSupport {
private static Map<Byte, String> ID_SERIALIZATIONNAME_MAP = new
HashMap<Byte, String>();
private static Map<String, Byte> SERIALIZATIONNAME_ID_MAP = new
HashMap<String, Byte>();
// Cache null object serialize results, for heartbeat request/response
serialize use.
- private static Map<Byte, byte[]> ID_NULLBYTES_MAP = new
ConcurrentHashMap<>();
+ private static ConcurrentMap<Byte, byte[]> ID_NULLBYTES_MAP = new
ConcurrentHashMap<>();
private static final ThreadLocal<byte[]> TL_BUFFER =
ThreadLocal.withInitial(() -> new byte[1024]);
@@ -107,7 +109,7 @@ public class CodecSupport {
* @return serialize result of null object
*/
public static byte[] getNullBytesOf(Serialization s) {
- return ID_NULLBYTES_MAP.computeIfAbsent(s.getContentTypeId(), k -> {
+ return ConcurrentHashMapUtils.computeIfAbsent(ID_NULLBYTES_MAP,
s.getContentTypeId(), k -> {
//Pre-generated Null object bytes
ByteArrayOutputStream baos = new ByteArrayOutputStream();
byte[] nullBytes = new byte[0];
diff --git
a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/zookeeper/AbstractZookeeperClient.java
b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/zookeeper/AbstractZookeeperClient.java
index a4d4f64a4d..9f5d3c7bf5 100644
---
a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/zookeeper/AbstractZookeeperClient.java
+++
b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/zookeeper/AbstractZookeeperClient.java
@@ -20,6 +20,7 @@ import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.config.configcenter.ConfigItem;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.ConcurrentHashSet;
import java.util.List;
@@ -106,8 +107,8 @@ public abstract class
AbstractZookeeperClient<TargetDataListener, TargetChildLis
@Override
public List<String> addChildListener(String path, final ChildListener
listener) {
- ConcurrentMap<ChildListener, TargetChildListener> listeners =
childListeners.computeIfAbsent(path, k -> new ConcurrentHashMap<>());
- TargetChildListener targetListener =
listeners.computeIfAbsent(listener, k -> createTargetChildListener(path, k));
+ ConcurrentMap<ChildListener, TargetChildListener> listeners =
ConcurrentHashMapUtils.computeIfAbsent(childListeners, path, k -> new
ConcurrentHashMap<>());
+ TargetChildListener targetListener =
ConcurrentHashMapUtils.computeIfAbsent(listeners, listener, k ->
createTargetChildListener(path, k));
return addTargetChildListener(path, targetListener);
}
@@ -118,8 +119,8 @@ public abstract class
AbstractZookeeperClient<TargetDataListener, TargetChildLis
@Override
public void addDataListener(String path, DataListener listener, Executor
executor) {
- ConcurrentMap<DataListener, TargetDataListener> dataListenerMap =
listeners.computeIfAbsent(path, k -> new ConcurrentHashMap<>());
- TargetDataListener targetListener =
dataListenerMap.computeIfAbsent(listener, k -> createTargetDataListener(path,
k));
+ ConcurrentMap<DataListener, TargetDataListener> dataListenerMap =
ConcurrentHashMapUtils.computeIfAbsent(listeners, path, k -> new
ConcurrentHashMap<>());
+ TargetDataListener targetListener =
ConcurrentHashMapUtils.computeIfAbsent(dataListenerMap, listener, k ->
createTargetDataListener(path, k));
addTargetDataListener(path, targetListener, executor);
}
diff --git
a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AdaptiveMetrics.java
b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AdaptiveMetrics.java
index 0c6fa4cce6..9fd3052b2c 100644
---
a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AdaptiveMetrics.java
+++
b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AdaptiveMetrics.java
@@ -16,6 +16,7 @@
*/
package org.apache.dubbo.rpc;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.StringUtils;
import java.util.Map;
@@ -45,7 +46,7 @@ public class AdaptiveMetrics {
private final AtomicLong errorReq = new AtomicLong();
private double ewma = 0;
- public double getLoad(String idKey,int weight,int timeout){
+ public double getLoad(String idKey, int weight, int timeout) {
AdaptiveMetrics metrics = getStatus(idKey);
//If the time more than 2 times, mandatory selected
@@ -53,13 +54,13 @@ public class AdaptiveMetrics {
return 0;
}
- if (metrics.currentTime > 0){
+ if (metrics.currentTime > 0) {
long multiple = (System.currentTimeMillis() - metrics.currentTime)
/ timeout + 1;
if (multiple > 0) {
if (metrics.currentProviderTime == metrics.currentTime) {
//penalty value
metrics.lastLatency = timeout * 2L;
- }else {
+ } else {
metrics.lastLatency = metrics.lastLatency >> multiple;
}
metrics.ewma = metrics.beta * metrics.ewma + (1 -
metrics.beta) * metrics.lastLatency;
@@ -68,49 +69,48 @@ public class AdaptiveMetrics {
}
long inflight = metrics.consumerReq.get() -
metrics.consumerSuccess.get() - metrics.errorReq.get();
- return metrics.providerCPULoad * (Math.sqrt(metrics.ewma) + 1) *
(inflight + 1) / ((((double)metrics.consumerSuccess.get() /
(double)(metrics.consumerReq.get() + 1)) * weight) + 1);
+ return metrics.providerCPULoad * (Math.sqrt(metrics.ewma) + 1) *
(inflight + 1) / ((((double) metrics.consumerSuccess.get() / (double)
(metrics.consumerReq.get() + 1)) * weight) + 1);
}
- public AdaptiveMetrics getStatus(String idKey){
- return metricsStatistics.computeIfAbsent(idKey, k -> new
AdaptiveMetrics());
+ public AdaptiveMetrics getStatus(String idKey) {
+ return ConcurrentHashMapUtils.computeIfAbsent(metricsStatistics,
idKey, k -> new AdaptiveMetrics());
}
- public void addConsumerReq(String idKey){
+ public void addConsumerReq(String idKey) {
AdaptiveMetrics metrics = getStatus(idKey);
metrics.consumerReq.incrementAndGet();
}
- public void addConsumerSuccess(String idKey){
+ public void addConsumerSuccess(String idKey) {
AdaptiveMetrics metrics = getStatus(idKey);
metrics.consumerSuccess.incrementAndGet();
}
- public void addErrorReq(String idKey){
+ public void addErrorReq(String idKey) {
AdaptiveMetrics metrics = getStatus(idKey);
metrics.errorReq.incrementAndGet();
}
- public void setPickTime(String idKey,long time){
+ public void setPickTime(String idKey, long time) {
AdaptiveMetrics metrics = getStatus(idKey);
metrics.pickTime = time;
}
-
- public void setProviderMetrics(String idKey,Map<String,String> metricsMap){
+ public void setProviderMetrics(String idKey, Map<String, String>
metricsMap) {
AdaptiveMetrics metrics = getStatus(idKey);
- long serviceTime =
Long.parseLong(Optional.ofNullable(metricsMap.get("curTime")).filter(v ->
StringUtils.isNumeric(v,false)).orElse("0"));
+ long serviceTime =
Long.parseLong(Optional.ofNullable(metricsMap.get("curTime")).filter(v ->
StringUtils.isNumeric(v, false)).orElse("0"));
//If server time is less than the current time, discard
- if (metrics.currentProviderTime > serviceTime){
+ if (metrics.currentProviderTime > serviceTime) {
return;
}
metrics.currentProviderTime = serviceTime;
metrics.currentTime = serviceTime;
- metrics.providerCPULoad =
Double.parseDouble(Optional.ofNullable(metricsMap.get("load")).filter(v ->
StringUtils.isNumeric(v,true)).orElse("0"));
- metrics.lastLatency =
Long.parseLong((Optional.ofNullable(metricsMap.get("rt")).filter(v ->
StringUtils.isNumeric(v,false)).orElse("0")));
+ metrics.providerCPULoad =
Double.parseDouble(Optional.ofNullable(metricsMap.get("load")).filter(v ->
StringUtils.isNumeric(v, true)).orElse("0"));
+ metrics.lastLatency =
Long.parseLong((Optional.ofNullable(metricsMap.get("rt")).filter(v ->
StringUtils.isNumeric(v, false)).orElse("0")));
metrics.beta = 0.5;
//Vt = β * Vt-1 + (1 - β ) * θt
diff --git
a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcStatus.java
b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcStatus.java
index d451c342aa..6ea826a116 100644
--- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcStatus.java
+++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcStatus.java
@@ -17,6 +17,7 @@
package org.apache.dubbo.rpc;
import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@@ -33,10 +34,10 @@ import java.util.concurrent.atomic.AtomicLong;
public class RpcStatus {
private static final ConcurrentMap<String, RpcStatus> SERVICE_STATISTICS =
new ConcurrentHashMap<String,
- RpcStatus>();
+ RpcStatus>();
private static final ConcurrentMap<String, ConcurrentMap<String,
RpcStatus>> METHOD_STATISTICS =
- new ConcurrentHashMap<String, ConcurrentMap<String, RpcStatus>>();
+ new ConcurrentHashMap<String, ConcurrentMap<String, RpcStatus>>();
private final ConcurrentMap<String, Object> values = new
ConcurrentHashMap<String, Object>();
@@ -58,7 +59,7 @@ public class RpcStatus {
*/
public static RpcStatus getStatus(URL url) {
String uri = url.toIdentityString();
- return SERVICE_STATISTICS.computeIfAbsent(uri, key -> new RpcStatus());
+ return ConcurrentHashMapUtils.computeIfAbsent(SERVICE_STATISTICS, uri,
key -> new RpcStatus());
}
/**
@@ -76,8 +77,8 @@ public class RpcStatus {
*/
public static RpcStatus getStatus(URL url, String methodName) {
String uri = url.toIdentityString();
- ConcurrentMap<String, RpcStatus> map =
METHOD_STATISTICS.computeIfAbsent(uri, k -> new ConcurrentHashMap<>());
- return map.computeIfAbsent(methodName, k -> new RpcStatus());
+ ConcurrentMap<String, RpcStatus> map =
ConcurrentHashMapUtils.computeIfAbsent(METHOD_STATISTICS, uri, k -> new
ConcurrentHashMap<>());
+ return ConcurrentHashMapUtils.computeIfAbsent(map, methodName, k ->
new RpcStatus());
}
/**
diff --git
a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/AccessLogFilter.java
b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/AccessLogFilter.java
index 3433fb4dc6..f692e6d87c 100644
---
a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/AccessLogFilter.java
+++
b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/AccessLogFilter.java
@@ -20,6 +20,7 @@ import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.threadpool.manager.FrameworkExecutorRepository;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.ConfigUtils;
import org.apache.dubbo.rpc.Constants;
import org.apache.dubbo.rpc.Filter;
@@ -40,6 +41,7 @@ import java.util.Optional;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -77,7 +79,7 @@ public class AccessLogFilter implements Filter {
// It's safe to declare it as singleton since it runs on single thread only
private final DateFormat fileNameFormatter = new
SimpleDateFormat(FILE_DATE_FORMAT);
- private final Map<String, Queue<AccessLogData>> logEntries = new
ConcurrentHashMap<>();
+ private final ConcurrentMap<String, Queue<AccessLogData>> logEntries = new
ConcurrentHashMap<>();
private AtomicBoolean scheduled = new AtomicBoolean();
@@ -127,12 +129,12 @@ public class AccessLogFilter implements Filter {
}
private void log(String accessLog, AccessLogData accessLogData) {
- Queue<AccessLogData> logQueue = logEntries.computeIfAbsent(accessLog,
k -> new ConcurrentLinkedQueue<>());
+ Queue<AccessLogData> logQueue =
ConcurrentHashMapUtils.computeIfAbsent(logEntries, accessLog, k -> new
ConcurrentLinkedQueue<>());
if (logQueue.size() < LOG_MAX_BUFFER) {
logQueue.add(accessLogData);
} else {
- logger.warn(CONFIG_FILTER_VALIDATION_EXCEPTION, "", "","AccessLog
buffer is full. Do a force writing to file to clear buffer.");
+ logger.warn(CONFIG_FILTER_VALIDATION_EXCEPTION, "", "", "AccessLog
buffer is full. Do a force writing to file to clear buffer.");
//just write current logSet to file.
writeLogSetToFile(accessLog, logQueue);
//after force writing, add accessLogData to current logSet
diff --git
a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/AbstractProtocol.java
b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/AbstractProtocol.java
index 50c37d9a81..3781b2f382 100644
---
a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/AbstractProtocol.java
+++
b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/AbstractProtocol.java
@@ -41,6 +41,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import static
org.apache.dubbo.common.constants.CommonConstants.DEFAULT_SERVER_SHUTDOWN_TIMEOUT;
import static
org.apache.dubbo.common.constants.CommonConstants.SHUTDOWN_WAIT_KEY;
@@ -59,7 +60,7 @@ public abstract class AbstractProtocol implements Protocol,
ScopeModelAware {
/**
* <host:port, ProtocolServer>
*/
- protected final Map<String, ProtocolServer> serverMap = new
ConcurrentHashMap<>();
+ protected final ConcurrentMap<String, ProtocolServer> serverMap = new
ConcurrentHashMap<>();
// TODO SoftReference
protected final Set<Invoker<?>> invokers = new ConcurrentHashSet<>();
diff --git
a/dubbo-rpc/dubbo-rpc-grpc/src/main/java/org/apache/dubbo/rpc/protocol/grpc/GrpcProtocol.java
b/dubbo-rpc/dubbo-rpc-grpc/src/main/java/org/apache/dubbo/rpc/protocol/grpc/GrpcProtocol.java
index 2223311578..1f5805a298 100644
---
a/dubbo-rpc/dubbo-rpc-grpc/src/main/java/org/apache/dubbo/rpc/protocol/grpc/GrpcProtocol.java
+++
b/dubbo-rpc/dubbo-rpc-grpc/src/main/java/org/apache/dubbo/rpc/protocol/grpc/GrpcProtocol.java
@@ -19,6 +19,7 @@ package org.apache.dubbo.rpc.protocol.grpc;
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.ConcurrentHashMapUtils;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.ProtocolServer;
import org.apache.dubbo.rpc.RpcException;
@@ -55,13 +56,13 @@ public class GrpcProtocol extends AbstractProxyProtocol {
@Override
protected <T> Runnable doExport(T proxiedImpl, Class<T> type, URL url)
throws RpcException {
String key = url.getAddress();
- ProtocolServer protocolServer = serverMap.computeIfAbsent(key, k -> {
+ ProtocolServer protocolServer =
ConcurrentHashMapUtils.computeIfAbsent(serverMap, key, k -> {
DubboHandlerRegistry registry = new DubboHandlerRegistry();
NettyServerBuilder builder =
- NettyServerBuilder
+ NettyServerBuilder
.forPort(url.getPort())
- .fallbackHandlerRegistry(registry);
+ .fallbackHandlerRegistry(registry);
Server originalServer = GrpcOptionsUtils.buildServerBuilder(url,
builder).build();
GrpcRemotingServer remotingServer = new
GrpcRemotingServer(originalServer, registry);
@@ -74,7 +75,7 @@ public class GrpcProtocol extends AbstractProxyProtocol {
ProviderModel providerModel =
serviceRepository.lookupExportedService(url.getServiceKey());
if (providerModel == null) {
throw new IllegalStateException("Service " + url.getServiceKey() +
"should have already been stored in service repository, " +
- "but failed to find it.");
+ "but failed to find it.");
}
Object originalImpl = providerModel.getServiceInstance();
@@ -84,7 +85,7 @@ public class GrpcProtocol extends AbstractProxyProtocol {
method.invoke(originalImpl, proxiedImpl);
} catch (Exception e) {
throw new IllegalStateException("Failed to set dubbo proxied
service impl to stub, please make sure your stub " +
- "was generated by the dubbo-protoc-compiler.", e);
+ "was generated by the dubbo-protoc-compiler.", e);
}
grpcServer.getRegistry().addService((BindableService) originalImpl,
url.getServiceKey());
@@ -101,13 +102,13 @@ public class GrpcProtocol extends AbstractProxyProtocol {
if (enclosingClass == null) {
throw new IllegalArgumentException(type.getName() + " must be
declared inside protobuf generated classes, " +
- "should be something like ServiceNameGrpc.IServiceName.");
+ "should be something like ServiceNameGrpc.IServiceName.");
}
final Method dubboStubMethod;
try {
dubboStubMethod = enclosingClass.getDeclaredMethod("getDubboStub",
Channel.class, CallOptions.class,
- URL.class);
+ URL.class);
} catch (NoSuchMethodException e) {
throw new IllegalArgumentException("Does not find getDubboStub in
" + enclosingClass.getName() + ", please use the customized
protoc-gen-dubbo-java to update the generated classes.");
}
@@ -118,9 +119,9 @@ public class GrpcProtocol extends AbstractProxyProtocol {
// CallOptions
try {
@SuppressWarnings("unchecked") final T stub = (T)
dubboStubMethod.invoke(null,
- channel,
- GrpcOptionsUtils.buildCallOptions(url),
- url
+ channel,
+ GrpcOptionsUtils.buildCallOptions(url),
+ url
);
final Invoker<T> target = proxyFactory.getInvoker(stub, type, url);
GrpcInvoker<T> grpcInvoker = new GrpcInvoker<>(type, url, target,
channel);
diff --git
a/dubbo-rpc/dubbo-rpc-rest/src/main/java/org/apache/dubbo/rpc/protocol/rest/RestProtocol.java
b/dubbo-rpc/dubbo-rpc-rest/src/main/java/org/apache/dubbo/rpc/protocol/rest/RestProtocol.java
index f38b0139c0..1966e7f0ec 100644
---
a/dubbo-rpc/dubbo-rpc-rest/src/main/java/org/apache/dubbo/rpc/protocol/rest/RestProtocol.java
+++
b/dubbo-rpc/dubbo-rpc-rest/src/main/java/org/apache/dubbo/rpc/protocol/rest/RestProtocol.java
@@ -17,6 +17,7 @@
package org.apache.dubbo.rpc.protocol.rest;
import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.remoting.http.HttpBinder;
import org.apache.dubbo.remoting.http.servlet.BootstrapListener;
@@ -45,6 +46,7 @@ import javax.ws.rs.ProcessingException;
import javax.ws.rs.WebApplicationException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import static
org.apache.dubbo.common.constants.CommonConstants.COMMA_SPLIT_PATTERN;
@@ -72,7 +74,7 @@ public class RestProtocol extends AbstractProxyProtocol {
private final RestServerFactory serverFactory = new RestServerFactory();
- private final Map<String, ReferenceCountedClient> clients = new
ConcurrentHashMap<>();
+ private final ConcurrentMap<String, ReferenceCountedClient> clients = new
ConcurrentHashMap<>();
private volatile ConnectionMonitor connectionMonitor;
@@ -93,7 +95,7 @@ public class RestProtocol extends AbstractProxyProtocol {
protected <T> Runnable doExport(T impl, Class<T> type, URL url) throws
RpcException {
String addr = getAddr(url);
Class implClass = url.getServiceModel().getProxyObject().getClass();
- RestProtocolServer server = (RestProtocolServer)
serverMap.computeIfAbsent(addr, restServer -> {
+ RestProtocolServer server = (RestProtocolServer)
ConcurrentHashMapUtils.computeIfAbsent(serverMap, addr, restServer -> {
RestProtocolServer s =
serverFactory.createServer(url.getParameter(SERVER_KEY, DEFAULT_SERVER));
s.setAddress(url.getAddress());
s.start(url);
@@ -135,7 +137,7 @@ public class RestProtocol extends AbstractProxyProtocol {
@Override
protected <T> T doRefer(Class<T> serviceType, URL url) throws RpcException
{
- ReferenceCountedClient referenceCountedClient =
clients.computeIfAbsent(url.getAddress(), _key -> {
+ ReferenceCountedClient referenceCountedClient =
ConcurrentHashMapUtils.computeIfAbsent(clients, url.getAddress(), _key -> {
// TODO more configs to add
return createReferenceCountedClient(url);
});
diff --git
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/SingleProtobufUtils.java
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/SingleProtobufUtils.java
index 0c9aeeb62f..5f79dad3b7 100644
---
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/SingleProtobufUtils.java
+++
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/SingleProtobufUtils.java
@@ -17,6 +17,8 @@
package org.apache.dubbo.rpc.protocol.tri;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
+
import com.google.protobuf.BoolValue;
import com.google.protobuf.BytesValue;
import com.google.protobuf.DoubleValue;
@@ -110,7 +112,7 @@ public class SingleProtobufUtils {
}
private static SingleMessageMarshaller<?> getMarshaller(Class<?> clz) {
- return MARSHALLER_CACHE.computeIfAbsent(clz, k -> new
SingleMessageMarshaller(k));
+ return ConcurrentHashMapUtils.computeIfAbsent(MARSHALLER_CACHE, clz, k
-> new SingleMessageMarshaller(k));
}
public static final class SingleMessageMarshaller<T extends MessageLite> {
diff --git
a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/AbstractProtocol.java
b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/AbstractProtocol.java
index 625de92e62..1054526594 100644
---
a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/AbstractProtocol.java
+++
b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/AbstractProtocol.java
@@ -43,6 +43,7 @@ import
io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse;
import static
org.apache.dubbo.common.constants.LoggerCodeConstants.INTERNAL_INTERRUPTED;
import static
org.apache.dubbo.common.constants.LoggerCodeConstants.PROTOCOL_FAILED_REQUEST;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
public abstract class AbstractProtocol<T, S extends DeltaResource<T>>
implements XdsProtocol<T>, XdsListener {
@@ -138,7 +139,7 @@ public abstract class AbstractProtocol<T, S extends
DeltaResource<T>> implements
Consumer<Map<String, T>> futureConsumer = future::complete;
try {
writeLock.lock();
-
consumerObserveMap.computeIfAbsent(consumerObserveResourceNames, key -> new
ArrayList<>())
+
ConcurrentHashMapUtils.computeIfAbsent((ConcurrentHashMap<Set<String>,
List<Consumer<Map<String,
T>>>>)consumerObserveMap,consumerObserveResourceNames, key -> new ArrayList<>())
.add(futureConsumer);
} finally {
writeLock.unlock();
diff --git
a/dubbo-xds/src/main/java/org/apache/dubbo/rpc/cluster/router/xds/EdsEndpointManager.java
b/dubbo-xds/src/main/java/org/apache/dubbo/rpc/cluster/router/xds/EdsEndpointManager.java
index f8ff539995..6291551889 100644
---
a/dubbo-xds/src/main/java/org/apache/dubbo/rpc/cluster/router/xds/EdsEndpointManager.java
+++
b/dubbo-xds/src/main/java/org/apache/dubbo/rpc/cluster/router/xds/EdsEndpointManager.java
@@ -25,6 +25,7 @@ import java.util.stream.Collectors;
import org.apache.dubbo.common.threadpool.manager.FrameworkExecutorRepository;
import org.apache.dubbo.common.utils.CollectionUtils;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.ConcurrentHashSet;
import org.apache.dubbo.registry.xds.util.PilotExchanger;
import org.apache.dubbo.registry.xds.util.protocol.message.Endpoint;
@@ -45,7 +46,7 @@ public class EdsEndpointManager {
public synchronized void subscribeEds(String cluster, EdsEndpointListener
listener) {
- Set<EdsEndpointListener> listeners =
ENDPOINT_LISTENERS.computeIfAbsent(cluster, key ->
+ Set<EdsEndpointListener> listeners =
ConcurrentHashMapUtils.computeIfAbsent(ENDPOINT_LISTENERS, cluster, key ->
new ConcurrentHashSet<>()
);
if (CollectionUtils.isEmpty(listeners)) {
@@ -59,7 +60,7 @@ public class EdsEndpointManager {
}
private void doSubscribeEds(String cluster) {
- EDS_LISTENERS.computeIfAbsent(cluster, key -> endpoints -> {
+ ConcurrentHashMapUtils.computeIfAbsent(EDS_LISTENERS, cluster, key ->
endpoints -> {
Set<Endpoint> result = endpoints.values()
.stream()
.map(EndpointResult::getEndpoints)
diff --git
a/dubbo-xds/src/main/java/org/apache/dubbo/rpc/cluster/router/xds/RdsRouteRuleManager.java
b/dubbo-xds/src/main/java/org/apache/dubbo/rpc/cluster/router/xds/RdsRouteRuleManager.java
index 3f464f5e43..a76508a138 100644
---
a/dubbo-xds/src/main/java/org/apache/dubbo/rpc/cluster/router/xds/RdsRouteRuleManager.java
+++
b/dubbo-xds/src/main/java/org/apache/dubbo/rpc/cluster/router/xds/RdsRouteRuleManager.java
@@ -20,9 +20,11 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import java.util.function.Consumer;
import org.apache.dubbo.common.utils.CollectionUtils;
+import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.ConcurrentHashSet;
import org.apache.dubbo.registry.xds.util.PilotExchanger;
import org.apache.dubbo.registry.xds.util.protocol.message.ListenerResult;
@@ -38,7 +40,7 @@ public class RdsRouteRuleManager {
private static final ConcurrentHashMap<String, List<XdsRouteRule>>
ROUTE_DATA_CACHE = new ConcurrentHashMap<>();
- private static final Map<String, RdsVirtualHostListener> RDS_LISTENERS =
new ConcurrentHashMap<>();
+ private static final ConcurrentMap<String, RdsVirtualHostListener>
RDS_LISTENERS = new ConcurrentHashMap<>();
private static volatile Consumer<Map<String, ListenerResult>> LDS_LISTENER;
@@ -51,7 +53,7 @@ public class RdsRouteRuleManager {
public synchronized void subscribeRds(String domain, XdsRouteRuleListener
listener) {
- Set<XdsRouteRuleListener> listeners =
RULE_LISTENERS.computeIfAbsent(domain, key ->
+ Set<XdsRouteRuleListener> listeners =
ConcurrentHashMapUtils.computeIfAbsent(RULE_LISTENERS, domain, key ->
new ConcurrentHashSet<>()
);
if (CollectionUtils.isEmpty(listeners)) {
@@ -107,7 +109,7 @@ public class RdsRouteRuleManager {
}
}
}
- RDS_LISTENERS.computeIfAbsent(domain, key -> new
RdsVirtualHostListener(domain, this));
+ ConcurrentHashMapUtils.computeIfAbsent(RDS_LISTENERS, domain, key ->
new RdsVirtualHostListener(domain, this));
RDS_LISTENER.accept(RDS_RESULT);
}
