From d2eb0ef4589b956d1648d6809cffd0a9966f5eb6 Mon Sep 17 00:00:00 2001 From: mxsm Date: Thu, 19 Jan 2023 23:07:12 +0800 Subject: [PATCH] [ISSUE #11294] Optimize ConcurrentHashMap#computeIfAbsent have performance problem in jdk1.8 (#11326) --- .../rpc/cluster/CacheableRouterFactory.java | 3 +- .../loadbalance/RoundRobinLoadBalance.java | 9 +-- .../ShortestResponseLoadBalance.java | 3 +- .../router/mesh/util/MeshRuleDispatcher.java | 8 ++- .../router/script/ScriptStateRouter.java | 15 ++--- .../state/CacheableStateRouterFactory.java | 3 +- .../beans/factory/ScopeBeanFactory.java | 6 +- .../apache/dubbo/common/bytecode/Wrapper.java | 16 +++--- .../common/cache/FileCacheStoreFactory.java | 8 ++- .../AbstractDynamicConfigurationFactory.java | 6 +- .../dubbo/common/convert/ConverterUtil.java | 9 +-- .../dubbo/common/logger/LoggerFactory.java | 9 +-- .../collector/stat/MetricsStatComposite.java | 49 +++++++++------- .../common/store/support/SimpleDataStore.java | 5 +- .../manager/DefaultExecutorRepository.java | 5 +- .../url/component/ServiceConfigURL.java | 24 ++++---- .../common/utils/ConcurrentHashMapUtils.java | 57 +++++++++++++++++++ .../apache/dubbo/config/AbstractConfig.java | 10 ++-- .../rpc/model/FrameworkServiceRepository.java | 5 +- .../rpc/model/ModuleServiceRepository.java | 9 +-- .../utils/ConcurrentHashMapUtilsTest.java | 40 +++++++++++++ .../config/bootstrap/DubboBootstrap.java | 12 ++-- .../config/utils/SimpleReferenceCache.java | 13 +++-- .../reference/ReferenceBeanManager.java | 12 ++-- .../spring/AbstractRegistryService.java | 7 ++- .../config/spring/SimpleRegistryService.java | 21 +++---- .../nacos/NacosDynamicConfiguration.java | 7 ++- .../support/zookeeper/CacheListener.java | 6 +- .../zookeeper/ZookeeperMetadataReport.java | 6 +- .../collector/AggregateMetricsCollector.java | 39 ++++++------- .../dubbo/monitor/support/MonitorFilter.java | 3 +- .../dubbo/monitor/dubbo/DubboMonitor.java | 3 +- .../AbstractServiceDiscoveryFactory.java | 3 +- .../ReflectionBasedServiceDiscovery.java | 3 +- .../client/ServiceDiscoveryRegistry.java | 3 +- .../ServiceDiscoveryRegistryDirectory.java | 6 +- .../DefaultMigrationAddressComparator.java | 6 +- .../migration/MigrationRuleListener.java | 6 +- .../support/CacheableFailbackRegistry.java | 14 +++-- .../multiple/MultipleServiceDiscovery.java | 6 +- .../nacos/NacosNamingServiceWrapper.java | 16 +++--- .../registry/zookeeper/ZookeeperRegistry.java | 9 +-- .../MultiplexProtocolConnectionManager.java | 3 +- .../exchange/PortUnificationExchanger.java | 5 +- .../remoting/transport/CodecSupport.java | 6 +- .../zookeeper/AbstractZookeeperClient.java | 9 +-- .../org/apache/dubbo/rpc/AdaptiveMetrics.java | 32 +++++------ .../java/org/apache/dubbo/rpc/RpcStatus.java | 11 ++-- .../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 +- .../router/xds/EdsEndpointManager.java | 5 +- .../router/xds/RdsRouteRuleManager.java | 8 ++- 56 files changed, 401 insertions(+), 225 deletions(-) create mode 100644 dubbo-common/src/main/java/org/apache/dubbo/common/utils/ConcurrentHashMapUtils.java create mode 100644 dubbo-common/src/test/java/org/apache/dubbo/common/utils/ConcurrentHashMapUtilsTest.java 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> 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> methodWeightMap = new ConcurrentHashMap>(); - /** * get invoker addr list cached for specified invocation *

@@ -89,7 +90,7 @@ public class RoundRobinLoadBalance extends AbstractLoadBalance { @Override protected Invoker doSelect(List> invokers, URL url, Invocation invocation) { String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName(); - ConcurrentMap map = methodWeightMap.computeIfAbsent(key, k -> new ConcurrentHashMap<>()); + ConcurrentMap 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 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 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> listenerMap = new ConcurrentHashMap<>(); + private final ConcurrentMap> 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 extends AbstractStateRouter { private static final int SCRIPT_ROUTER_DEFAULT_PRIORITY = 0; private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(ScriptStateRouter.class); - private static final Map ENGINES = new ConcurrentHashMap<>(); + private static final ConcurrentMap ENGINES = new ConcurrentHashMap<>(); private final ScriptEngine engine; @@ -93,8 +94,8 @@ public class ScriptStateRouter extends AbstractStateRouter { 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 extends AbstractStateRouter { 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 extends AbstractStateRouter { 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 StateRouter getRouter(Class interfaceClass, URL url) { - return routerMap.computeIfAbsent(url.getServiceKey(), k -> createRouter(interfaceClass, url)); + return ConcurrentHashMapUtils.computeIfAbsent(routerMap, url.getServiceKey(), k -> createRouter(interfaceClass, url)); } protected abstract StateRouter createRouter(Class 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 extensionPostProcessors; - private Map beanNameIdCounterMap = new ConcurrentHashMap<>(); + private ConcurrentHashMap beanNameIdCounterMap = new ConcurrentHashMap<>(); private List 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 getBean(Class 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, Wrapper> WRAPPER_MAP = new ConcurrentHashMap, Wrapper>(); //class wrapper map + private static final ConcurrentMap, Wrapper> WRAPPER_MAP = new ConcurrentHashMap, 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 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 cacheMap = new ConcurrentHashMap<>(); + private static final ConcurrentMap cacheMap = new ConcurrentHashMap<>(); private static final String SUFFIX = ".dubbo.cache"; private static final char ESCAPE_MARK = '%'; - private static final Set LEGAL_CHARACTERS = Collections.unmodifiableSet(new HashSet(){{ + private static final Set LEGAL_CHARACTERS = Collections.unmodifiableSet(new HashSet() {{ // - $ . _ 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 dynamicConfigurations = new ConcurrentHashMap<>(); + private volatile ConcurrentHashMap 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, Map, List>> converterCache = new ConcurrentHashMap<>(); + private final ConcurrentMap, ConcurrentMap, List>> 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, List> toTargetMap = converterCache.computeIfAbsent(sourceType, (k) -> new ConcurrentHashMap<>()); - List converters = toTargetMap.computeIfAbsent(targetType, (k) -> frameworkModel.getExtensionLoader(Converter.class) + ConcurrentMap, List> toTargetMap = ConcurrentHashMapUtils.computeIfAbsent(converterCache, sourceType, (k) -> new ConcurrentHashMap<>()); + List 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 stats = new ConcurrentHashMap<>(); - private final Map lastRT = new ConcurrentHashMap<>(); - private final Map minRT = new ConcurrentHashMap<>(); - private final Map maxRT = new ConcurrentHashMap<>(); - private final Map avgRT = new ConcurrentHashMap<>(); - private final Map totalRT = new ConcurrentHashMap<>(); - private final Map rtCount = new ConcurrentHashMap<>(); + private final ConcurrentMap lastRT = new ConcurrentHashMap<>(); + private final ConcurrentMap minRT = new ConcurrentHashMap<>(); + private final ConcurrentMap maxRT = new ConcurrentHashMap<>(); + private final ConcurrentMap avgRT = new ConcurrentHashMap<>(); + private final ConcurrentMap totalRT = new ConcurrentHashMap<>(); + private final ConcurrentMap rtCount = new ConcurrentHashMap<>(); private final String applicationName; private final List 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 getLastRT(){ + public Map getLastRT() { return this.lastRT; } - public Map getMinRT(){ + + public Map getMinRT() { return this.minRT; } - public Map getMaxRT(){ + public Map getMaxRT() { return this.maxRT; } - public Map getAvgRT(){ + + public Map getAvgRT() { return this.avgRT; } - public Map getTotalRT(){ + + public Map getTotalRT() { return this.totalRT; } - public Map getRtCount(){ + + public Map 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 { // > private ConcurrentMap> data = - new ConcurrentHashMap>(); + new ConcurrentHashMap>(); @Override public Map get(String componentName) { @@ -50,7 +51,7 @@ public class SimpleDataStore implements DataStore { @Override public void put(String componentName, String key, Object value) { - Map componentData = data.computeIfAbsent(componentName, k -> new ConcurrentHashMap<>()); + Map 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 executors = data.computeIfAbsent(executorKey, k -> new ConcurrentHashMap<>()); + ConcurrentMap 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 urls; - private volatile transient Map numbers; - private volatile transient Map> methodNumbers; + private volatile transient ConcurrentMap urls; + private volatile transient ConcurrentMap numbers; + private volatile transient ConcurrentMap> 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 parameters) { + String username, + String password, + String host, + int port, + String path, + Map 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 keyNumber = getMethodNumbers().computeIfAbsent(method, m -> new HashMap<>()); + Map keyNumber = ConcurrentHashMapUtils.computeIfAbsent(getMethodNumbers(), method, m -> new HashMap<>()); keyNumber.put(key, n); } - protected Map> getMethodNumbers() { + protected ConcurrentMap> 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.
+ * @see https://bugs.openjdk.java.net/browse/JDK-8161372 + * + */ + public static V computeIfAbsent(ConcurrentMap map, K key, Function 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 tagNameCache = new ConcurrentHashMap<>(); + private static final ConcurrentMap tagNameCache = new ConcurrentHashMap<>(); /** * attributed getter method cache for equals(), hashCode() and toString() */ - private static final Map> attributedMethodCache = new ConcurrentHashMap<>(); + private static final ConcurrentMap> 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 getAttributedMethods() { Class 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 serviceDescriptors = services.computeIfAbsent(interfaceClazz.getName(), + List serviceDescriptors = ConcurrentHashMapUtils.computeIfAbsent(services, interfaceClazz.getName(), k -> new CopyOnWriteArrayList<>()); synchronized (serviceDescriptors) { Optional 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 serviceDescriptors = services.computeIfAbsent(path, + List serviceDescriptors = ConcurrentHashMapUtils.computeIfAbsent(services, path, _k -> new CopyOnWriteArrayList<>()); synchronized (serviceDescriptors) { Optional previous = serviceDescriptors.stream() @@ -179,7 +180,7 @@ public class ModuleServiceRepository { public void reRegisterConsumer(String newServiceKey, String serviceKey) { List 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-common/src/test/java/org/apache/dubbo/common/utils/ConcurrentHashMapUtilsTest.java b/dubbo-common/src/test/java/org/apache/dubbo/common/utils/ConcurrentHashMapUtilsTest.java new file mode 100644 index 0000000000..3fc895497f --- /dev/null +++ b/dubbo-common/src/test/java/org/apache/dubbo/common/utils/ConcurrentHashMapUtilsTest.java @@ -0,0 +1,40 @@ +/* + * 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 org.junit.jupiter.api.Test; + +import java.util.concurrent.ConcurrentHashMap; + +import static org.junit.jupiter.api.Assertions.assertEquals; + + +class ConcurrentHashMapUtilsTest { + + @Test + public void testComputeIfAbsent() { + ConcurrentHashMap 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 instanceMap = new ConcurrentHashMap<>(); + private static volatile ConcurrentMap 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 consumerBuilder) { + public DubboBootstrap metadataReport(String id, Consumer 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>> referenceKeyMap = new ConcurrentHashMap<>(); - private final Map, List>> referenceTypeMap = new ConcurrentHashMap<>(); + private final ConcurrentMap>> referenceKeyMap = new ConcurrentHashMap<>(); + private final ConcurrentMap, List>> referenceTypeMap = new ConcurrentHashMap<>(); private final Map, 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> referencesOfType = referenceTypeMap.computeIfAbsent(type, _t -> Collections.synchronizedList(new ArrayList<>())); + List> referencesOfType = ConcurrentHashMapUtils.computeIfAbsent(referenceTypeMap, type, _t -> Collections.synchronizedList(new ArrayList<>())); referencesOfType.add(rc); - List> referenceConfigList = referenceKeyMap.computeIfAbsent(key, _k -> Collections.synchronizedList(new ArrayList<>())); + List> 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> referenceKeyMap = new ConcurrentHashMap<>(); + private ConcurrentMap> referenceKeyMap = new ConcurrentHashMap<>(); // reference alias -> reference bean name - private Map referenceAliasMap = new ConcurrentHashMap<>(); + private ConcurrentMap referenceAliasMap = new ConcurrentHashMap<>(); //reference bean name -> ReferenceBean - private Map referenceBeanMap = new ConcurrentHashMap<>(); + private ConcurrentMap referenceBeanMap = new ConcurrentHashMap<>(); //reference key -> ReferenceConfig instance - private Map referenceConfigMap = new ConcurrentHashMap<>(); + private ConcurrentMap 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 list = referenceKeyMap.computeIfAbsent(referenceKey, (key) -> new ArrayList<>()); + List 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 urls = registered.computeIfAbsent(service, k -> new CopyOnWriteArrayList<>()); + List 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 listeners = notifyListeners.computeIfAbsent(service, k -> new CopyOnWriteArrayList<>()); + List 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 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 urls = remoteRegistered.computeIfAbsent(client, k -> new ConcurrentHashMap<>()); + Map 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 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 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 listeners = remoteListeners.computeIfAbsent(client, k -> new ConcurrentHashMap<>()); + Map 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 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 watchListenerMap; + private final ConcurrentMap 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 pathKeyListeners = new ConcurrentHashMap<>(); + private ConcurrentMap 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 casListenerMap = new ConcurrentHashMap<>(); + private ConcurrentMap casListenerMap = new ConcurrentHashMap<>(); public ZookeeperMetadataReport(URL url, ZookeeperTransporter zookeeperTransporter) { @@ -162,7 +164,7 @@ public class ZookeeperMetadataReport extends AbstractMetadataReport { @Override public Set 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 totalRequests = new ConcurrentHashMap<>(); - private final Map succeedRequests = new ConcurrentHashMap<>(); - private final Map failedRequests = new ConcurrentHashMap<>(); - private final Map businessFailedRequests = new ConcurrentHashMap<>(); - private final Map timeoutRequests = new ConcurrentHashMap<>(); - private final Map limitRequests = new ConcurrentHashMap<>(); - private final Map totalFailedRequests = new ConcurrentHashMap<>(); - private final Map qps = new ConcurrentHashMap<>(); - private final Map rt = new ConcurrentHashMap<>(); + private final ConcurrentMap totalRequests = new ConcurrentHashMap<>(); + private final ConcurrentMap succeedRequests = new ConcurrentHashMap<>(); + private final ConcurrentMap failedRequests = new ConcurrentHashMap<>(); + private final ConcurrentMap businessFailedRequests = new ConcurrentHashMap<>(); + private final ConcurrentMap timeoutRequests = new ConcurrentHashMap<>(); + private final ConcurrentMap limitRequests = new ConcurrentHashMap<>(); + private final ConcurrentMap totalFailedRequests = new ConcurrentHashMap<>(); + private final ConcurrentMap qps = new ConcurrentHashMap<>(); + private final ConcurrentMap 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 reference = statisticsMap.computeIfAbsent(statistics, k -> new AtomicReference<>()); + AtomicReference 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 extends DynamicDirectory { private final Set providerFirstParams; private final ModuleModel moduleModel; private final ProtocolServiceKey consumerProtocolServiceKey; - private final Map customizedConsumerUrlMap = new ConcurrentHashMap<>(); + private final ConcurrentMap customizedConsumerUrlMap = new ConcurrentHashMap<>(); public ServiceDiscoveryRegistryDirectory(Class serviceType, URL url) { super(serviceType, url); @@ -351,7 +353,7 @@ public class ServiceDiscoveryRegistryDirectory extends DynamicDirectory { } 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> serviceMigrationData = new ConcurrentHashMap<>(); + private ConcurrentMap> serviceMigrationData = new ConcurrentHashMap<>(); @Override public boolean shouldMigrate(ClusterInvoker newInvoker, ClusterInvoker oldInvoker, MigrationRule rule) { - Map migrationData = serviceMigrationData.computeIfAbsent(oldInvoker.getUrl().getDisplayServiceKey(), _k -> new ConcurrentHashMap<>()); + Map 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 handlers = new ConcurrentHashMap<>(); + protected final ConcurrentMap handlers = new ConcurrentHashMap<>(); protected final LinkedBlockingQueue 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 stringAddress = new ConcurrentHashMap<>(); - protected Map stringParam = new ConcurrentHashMap<>(); + protected ConcurrentMap stringAddress = new ConcurrentHashMap<>(); + protected ConcurrentMap 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 singleListenerMap; + private final ConcurrentMap singleListenerMap; public MultiServiceInstancesChangedListener(Set serviceNames, ServiceDiscovery serviceDiscovery) { super(serviceNames, serviceDiscovery); @@ -207,7 +209,7 @@ public class MultipleServiceDiscovery implements ServiceDiscovery { public SingleServiceInstancesChangedListener getAndComputeIfAbsent(String registryKey, Function 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 registerStatus = new ConcurrentHashMap<>(); - private final Map subscribeStatus = new ConcurrentHashMap<>(); + private final ConcurrentMap registerStatus = new ConcurrentHashMap<>(); + private final ConcurrentMap 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 listeners = zkListeners.computeIfAbsent(url, k -> new ConcurrentHashMap<>()); + ConcurrentMap 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 listeners = zkListeners.computeIfAbsent(url, k -> new ConcurrentHashMap<>()); - ChildListener zkListener = listeners.computeIfAbsent(listener, k -> new RegistryChildListenerImpl(url, k, latch)); + ConcurrentMap 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 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 ID_SERIALIZATIONNAME_MAP = new HashMap(); private static Map SERIALIZATIONNAME_ID_MAP = new HashMap(); // Cache null object serialize results, for heartbeat request/response serialize use. - private static Map ID_NULLBYTES_MAP = new ConcurrentHashMap<>(); + private static ConcurrentMap ID_NULLBYTES_MAP = new ConcurrentHashMap<>(); private static final ThreadLocal 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 addChildListener(String path, final ChildListener listener) { - ConcurrentMap listeners = childListeners.computeIfAbsent(path, k -> new ConcurrentHashMap<>()); - TargetChildListener targetListener = listeners.computeIfAbsent(listener, k -> createTargetChildListener(path, k)); + ConcurrentMap 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 dataListenerMap = listeners.computeIfAbsent(path, k -> new ConcurrentHashMap<>()); - TargetDataListener targetListener = dataListenerMap.computeIfAbsent(listener, k -> createTargetDataListener(path, k)); + ConcurrentMap 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 metricsMap){ + public void setProviderMetrics(String idKey, Map 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 SERVICE_STATISTICS = new ConcurrentHashMap(); + RpcStatus>(); private static final ConcurrentMap> METHOD_STATISTICS = - new ConcurrentHashMap>(); + new ConcurrentHashMap>(); private final ConcurrentMap values = new ConcurrentHashMap(); @@ -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 map = METHOD_STATISTICS.computeIfAbsent(uri, k -> new ConcurrentHashMap<>()); - return map.computeIfAbsent(methodName, k -> new RpcStatus()); + ConcurrentMap 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> logEntries = new ConcurrentHashMap<>(); + private final ConcurrentMap> 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 logQueue = logEntries.computeIfAbsent(accessLog, k -> new ConcurrentLinkedQueue<>()); + Queue 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 { /** * */ - protected final Map serverMap = new ConcurrentHashMap<>(); + protected final ConcurrentMap serverMap = new ConcurrentHashMap<>(); // TODO SoftReference protected final Set> 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 Runnable doExport(T proxiedImpl, Class 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 target = proxyFactory.getInvoker(stub, type, url); GrpcInvoker 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 clients = new ConcurrentHashMap<>(); + private final ConcurrentMap clients = new ConcurrentHashMap<>(); private volatile ConnectionMonitor connectionMonitor; @@ -93,7 +95,7 @@ public class RestProtocol extends AbstractProxyProtocol { protected Runnable doExport(T impl, Class 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 doRefer(Class 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 { 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> implements XdsProtocol, XdsListener { @@ -138,7 +139,7 @@ public abstract class AbstractProtocol> implements Consumer> futureConsumer = future::complete; try { writeLock.lock(); - consumerObserveMap.computeIfAbsent(consumerObserveResourceNames, key -> new ArrayList<>()) + ConcurrentHashMapUtils.computeIfAbsent((ConcurrentHashMap, List>>>)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 listeners = ENDPOINT_LISTENERS.computeIfAbsent(cluster, key -> + Set 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 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> ROUTE_DATA_CACHE = new ConcurrentHashMap<>(); - private static final Map RDS_LISTENERS = new ConcurrentHashMap<>(); + private static final ConcurrentMap RDS_LISTENERS = new ConcurrentHashMap<>(); private static volatile Consumer> LDS_LISTENER; @@ -51,7 +53,7 @@ public class RdsRouteRuleManager { public synchronized void subscribeRds(String domain, XdsRouteRuleListener listener) { - Set listeners = RULE_LISTENERS.computeIfAbsent(domain, key -> + Set 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); }