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

[ISSUE #11294] Optimize ConcurrentHashMap#computeIfAbsent have performance problem in jdk1.8 (#11326)

This commit is contained in:
mxsm 2023-01-19 23:07:12 +08:00 committed by GitHub
parent 4836bf7919
commit d2eb0ef458
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
56 changed files with 401 additions and 225 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/CacheableRouterFactory.java
View File

@ -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);

9
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/loadbalance/RoundRobinLoadBalance.java
View File

@ -17,6 +17,7 @@
package org.apache.dubbo.rpc.cluster.loadbalance;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
@ -35,6 +36,8 @@ public class RoundRobinLoadBalance extends AbstractLoadBalance {
private static final int RECYCLE_PERIOD = 60000;
private ConcurrentMap<String, ConcurrentMap<String, WeightedRoundRobin>> methodWeightMap = new ConcurrentHashMap<>();
protected static class WeightedRoundRobin {
private int weight;
private AtomicLong current = new AtomicLong(0);
@ -66,8 +69,6 @@ public class RoundRobinLoadBalance extends AbstractLoadBalance {
}
}
private ConcurrentMap<String, ConcurrentMap<String, WeightedRoundRobin>> methodWeightMap = new ConcurrentHashMap<String, ConcurrentMap<String, WeightedRoundRobin>>();
/**
* get invoker addr list cached for specified invocation
* <p>
@ -89,7 +90,7 @@ public class RoundRobinLoadBalance extends AbstractLoadBalance {
@Override
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();
ConcurrentMap<String, WeightedRoundRobin> map = methodWeightMap.computeIfAbsent(key, k -> new ConcurrentHashMap<>());
ConcurrentMap<String, WeightedRoundRobin> map = ConcurrentHashMapUtils.computeIfAbsent(methodWeightMap, key, k -> new ConcurrentHashMap<>());
int totalWeight = 0;
long maxCurrent = Long.MIN_VALUE;
long now = System.currentTimeMillis();
@ -98,7 +99,7 @@ public class RoundRobinLoadBalance extends AbstractLoadBalance {
for (Invoker<T> invoker : invokers) {
String identifyString = invoker.getUrl().toIdentityString();
int weight = getWeight(invoker, invocation);
WeightedRoundRobin weightedRoundRobin = map.computeIfAbsent(identifyString, k -> {
WeightedRoundRobin weightedRoundRobin = ConcurrentHashMapUtils.computeIfAbsent(map, identifyString, k -> {
WeightedRoundRobin wrr = new WeightedRoundRobin();
wrr.setWeight(weight);
return wrr;

3
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/loadbalance/ShortestResponseLoadBalance.java
View File

@ -18,6 +18,7 @@ package org.apache.dubbo.rpc.cluster.loadbalance;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.threadpool.manager.FrameworkExecutorRepository;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.RpcStatus;
@ -116,7 +117,7 @@ public class ShortestResponseLoadBalance extends AbstractLoadBalance implements
for (int i = 0; i < length; i++) {
Invoker<T> invoker = invokers.get(i);
RpcStatus rpcStatus = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName());
SlideWindowData slideWindowData = methodMap.computeIfAbsent(rpcStatus, SlideWindowData::new);
SlideWindowData slideWindowData = ConcurrentHashMapUtils.computeIfAbsent(methodMap, rpcStatus, SlideWindowData::new);
// Calculate the estimated response time from the product of active connections and succeeded average elapsed time.
long estimateResponse = slideWindowData.getEstimateResponse();

8
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mesh/util/MeshRuleDispatcher.java
View File

@ -20,12 +20,14 @@ package org.apache.dubbo.rpc.cluster.router.mesh.util;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.ConcurrentHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.CLUSTER_NO_RULE_LISTENER;
@ -34,7 +36,7 @@ public class MeshRuleDispatcher {
private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(MeshRuleDispatcher.class);
private final String appName;
private final Map<String, Set<MeshRuleListener>> listenerMap = new ConcurrentHashMap<>();
private final ConcurrentMap<String, Set<MeshRuleListener>> listenerMap = new ConcurrentHashMap<>();
public MeshRuleDispatcher(String appName) {
this.appName = appName;
@ -57,7 +59,7 @@ public class MeshRuleDispatcher {
listener.onRuleChange(appName, entry.getValue());
}
} else {
logger.warn(CLUSTER_NO_RULE_LISTENER,"Receive mesh rule but none of listener has been registered","","Receive rule but none of listener has been registered. Maybe type not matched. Rule Type: " + ruleType);
logger.warn(CLUSTER_NO_RULE_LISTENER, "Receive mesh rule but none of listener has been registered", "", "Receive rule but none of listener has been registered. Maybe type not matched. Rule Type: " + ruleType);
}
}
// clear rule listener not being notified in this time
@ -75,7 +77,7 @@ public class MeshRuleDispatcher {
if (listener == null) {
return;
}
listenerMap.computeIfAbsent(listener.ruleSuffix(), (k) -> new ConcurrentHashSet<>())
ConcurrentHashMapUtils.computeIfAbsent(listenerMap, listener.ruleSuffix(), (k) -> new ConcurrentHashSet<>())
.add(listener);
}

15
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/script/ScriptStateRouter.java
View File

@ -19,6 +19,7 @@ package org.apache.dubbo.rpc.cluster.router.script;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.Holder;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.rpc.Invocation;
@ -45,8 +46,8 @@ import java.security.cert.Certificate;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.stream.Collectors;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.CLUSTER_SCRIPT_EXCEPTION;
@ -64,7 +65,7 @@ public class ScriptStateRouter<T> extends AbstractStateRouter<T> {
private static final int SCRIPT_ROUTER_DEFAULT_PRIORITY = 0;
private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(ScriptStateRouter.class);
private static final Map<String, ScriptEngine> ENGINES = new ConcurrentHashMap<>();
private static final ConcurrentMap<String, ScriptEngine> ENGINES = new ConcurrentHashMap<>();
private final ScriptEngine engine;
@ -93,8 +94,8 @@ public class ScriptStateRouter<T> extends AbstractStateRouter<T> {
Compilable compilable = (Compilable) engine;
function = compilable.compile(rule);
} catch (ScriptException e) {
logger.error(CLUSTER_SCRIPT_EXCEPTION,"script route rule invalid","","script route error, rule has been ignored. rule: " + rule +
", url: " + RpcContext.getServiceContext().getUrl(),e);
logger.error(CLUSTER_SCRIPT_EXCEPTION, "script route rule invalid", "", "script route error, rule has been ignored. rule: " + rule +
", url: " + RpcContext.getServiceContext().getUrl(), e);
}
}
@ -115,7 +116,7 @@ public class ScriptStateRouter<T> extends AbstractStateRouter<T> {
private ScriptEngine getEngine(URL url) {
String type = url.getParameter(TYPE_KEY, DEFAULT_SCRIPT_TYPE_KEY);
return ENGINES.computeIfAbsent(type, t -> {
return ConcurrentHashMapUtils.computeIfAbsent(ENGINES, type, t -> {
ScriptEngine scriptEngine = new ScriptEngineManager().getEngineByName(type);
if (scriptEngine == null) {
throw new IllegalStateException("unsupported route engine type: " + type);
@ -137,8 +138,8 @@ public class ScriptStateRouter<T> extends AbstractStateRouter<T> {
try {
return function.eval(bindings);
} catch (ScriptException e) {
logger.error(CLUSTER_SCRIPT_EXCEPTION,"Scriptrouter exec script error","","Script route error, rule has been ignored. rule: " + rule + ", method:" +
invocation.getMethodName() + ", url: " + RpcContext.getContext().getUrl(),e);
logger.error(CLUSTER_SCRIPT_EXCEPTION, "Scriptrouter exec script error", "", "Script route error, rule has been ignored. rule: " + rule + ", method:" +
invocation.getMethodName() + ", url: " + RpcContext.getContext().getUrl(), e);
return invokers;
}
}, accessControlContext));

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

@ -17,6 +17,7 @@
package org.apache.dubbo.rpc.cluster.router.state;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@ -31,7 +32,7 @@ public abstract class CacheableStateRouterFactory implements StateRouterFactory
@Override
public <T> StateRouter<T> getRouter(Class<T> interfaceClass, URL url) {
return routerMap.computeIfAbsent(url.getServiceKey(), k -> createRouter(interfaceClass, url));
return ConcurrentHashMapUtils.computeIfAbsent(routerMap, url.getServiceKey(), k -> createRouter(interfaceClass, url));
}
protected abstract <T> StateRouter<T> createRouter(Class<T> interfaceClass, URL url);

6
dubbo-common/src/main/java/org/apache/dubbo/common/beans/factory/ScopeBeanFactory.java
View File

@ -24,12 +24,12 @@ import org.apache.dubbo.common.extension.ExtensionPostProcessor;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.resource.Disposable;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.rpc.model.ScopeModelAccessor;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicBoolean;
@ -49,7 +49,7 @@ public class ScopeBeanFactory {
private final ScopeBeanFactory parent;
private ExtensionAccessor extensionAccessor;
private List<ExtensionPostProcessor> extensionPostProcessors;
private Map<Class, AtomicInteger> beanNameIdCounterMap = new ConcurrentHashMap<>();
private ConcurrentHashMap<Class, AtomicInteger> beanNameIdCounterMap = new ConcurrentHashMap<>();
private List<BeanInfo> registeredBeanInfos = new CopyOnWriteArrayList<>();
private InstantiationStrategy instantiationStrategy;
private AtomicBoolean destroyed = new AtomicBoolean();
@ -183,7 +183,7 @@ public class ScopeBeanFactory {
}
private int getNextId(Class<?> beanClass) {
return beanNameIdCounterMap.computeIfAbsent(beanClass, key -> new AtomicInteger()).incrementAndGet();
return ConcurrentHashMapUtils.computeIfAbsent(beanNameIdCounterMap, beanClass, key -> new AtomicInteger()).incrementAndGet();
}
public <T> T getBean(Class<T> type) {

16
dubbo-common/src/main/java/org/apache/dubbo/common/bytecode/Wrapper.java
View File

@ -17,6 +17,7 @@
package org.apache.dubbo.common.bytecode;
import org.apache.dubbo.common.utils.ClassUtils;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.ReflectUtils;
import javassist.ClassPool;
@ -34,6 +35,7 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.regex.Matcher;
import java.util.stream.Collectors;
@ -42,7 +44,7 @@ import java.util.stream.Collectors;
* Wrapper.
*/
public abstract class Wrapper {
private static final Map<Class<?>, Wrapper> WRAPPER_MAP = new ConcurrentHashMap<Class<?>, Wrapper>(); //class wrapper map
private static final ConcurrentMap<Class<?>, Wrapper> WRAPPER_MAP = new ConcurrentHashMap<Class<?>, Wrapper>(); //class wrapper map
private static final String[] EMPTY_STRING_ARRAY = new String[0];
private static final String[] OBJECT_METHODS = new String[]{"getClass", "hashCode", "toString", "equals"};
private static final Wrapper OBJECT_WRAPPER = new Wrapper() {
@ -119,7 +121,7 @@ public abstract class Wrapper {
return OBJECT_WRAPPER;
}
return WRAPPER_MAP.computeIfAbsent(c, Wrapper::makeWrapper);
return ConcurrentHashMapUtils.computeIfAbsent(WRAPPER_MAP, c, Wrapper::makeWrapper);
}
private static Wrapper makeWrapper(Class<?> c) {
@ -171,16 +173,16 @@ public abstract class Wrapper {
}
Method[] methods = Arrays.stream(c.getMethods())
.filter(method -> allMethod.contains(ReflectUtils.getDesc(method)))
.collect(Collectors.toList())
.toArray(new Method[] {});
.filter(method -> allMethod.contains(ReflectUtils.getDesc(method)))
.collect(Collectors.toList())
.toArray(new Method[]{});
// get all public method.
boolean hasMethod = ClassUtils.hasMethods(methods);
if (hasMethod) {
Map<String, Integer> sameNameMethodCount = new HashMap<>((int) (methods.length / 0.75f) + 1);
for (Method m : methods) {
sameNameMethodCount.compute(m.getName(),
(key, oldValue) -> oldValue == null ? 1 : oldValue + 1);
(key, oldValue) -> oldValue == null ? 1 : oldValue + 1);
}
c3.append(" try{");
@ -200,7 +202,7 @@ public abstract class Wrapper {
if (len > 0) {
for (int l = 0; l < len; l++) {
c3.append(" && ").append(" $3[").append(l).append("].getName().equals(\"")
.append(m.getParameterTypes()[l].getName()).append("\")");
.append(m.getParameterTypes()[l].getName()).append("\")");
}
}
}

8
dubbo-common/src/main/java/org/apache/dubbo/common/cache/FileCacheStoreFactory.java vendored
View File

@ -19,6 +19,7 @@ package org.apache.dubbo.common.cache;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import java.io.File;
import java.io.IOException;
@ -33,6 +34,7 @@ import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.COMMON_CACHE_PATH_INACCESSIBLE;
@ -50,11 +52,11 @@ public final class FileCacheStoreFactory {
}
private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(FileCacheStoreFactory.class);
private static final Map<String, FileCacheStore> cacheMap = new ConcurrentHashMap<>();
private static final ConcurrentMap<String, FileCacheStore> cacheMap = new ConcurrentHashMap<>();
private static final String SUFFIX = ".dubbo.cache";
private static final char ESCAPE_MARK = '%';
private static final Set<Character> LEGAL_CHARACTERS = Collections.unmodifiableSet(new HashSet<Character>(){{
private static final Set<Character> LEGAL_CHARACTERS = Collections.unmodifiableSet(new HashSet<Character>() {{
// - $ . _ 0-9 a-z A-Z
add('-');
add('$');
@ -108,7 +110,7 @@ public final class FileCacheStoreFactory {
String cacheFilePath = basePath + File.separator + cacheName;
return cacheMap.computeIfAbsent(cacheFilePath, k -> getFile(k, enableFileCache));
return ConcurrentHashMapUtils.computeIfAbsent(cacheMap, cacheFilePath, k -> getFile(k, enableFileCache));
}
/**

6
dubbo-common/src/main/java/org/apache/dubbo/common/config/configcenter/AbstractDynamicConfigurationFactory.java
View File

@ -17,8 +17,8 @@
package org.apache.dubbo.common.config.configcenter;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_KEY;
@ -31,12 +31,12 @@ import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_KEY;
*/
public abstract class AbstractDynamicConfigurationFactory implements DynamicConfigurationFactory {
private volatile Map<String, DynamicConfiguration> dynamicConfigurations = new ConcurrentHashMap<>();
private volatile ConcurrentHashMap<String, DynamicConfiguration> dynamicConfigurations = new ConcurrentHashMap<>();
@Override
public final DynamicConfiguration getDynamicConfiguration(URL url) {
String key = url == null ? DEFAULT_KEY : url.toServiceString();
return dynamicConfigurations.computeIfAbsent(key, k -> createDynamicConfiguration(url));
return ConcurrentHashMapUtils.computeIfAbsent(dynamicConfigurations, key, k -> createDynamicConfiguration(url));
}
protected abstract DynamicConfiguration createDynamicConfiguration(URL url);

9
dubbo-common/src/main/java/org/apache/dubbo/common/convert/ConverterUtil.java
View File

@ -17,16 +17,17 @@
package org.apache.dubbo.common.convert;
import org.apache.dubbo.common.extension.ExtensionLoader;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.rpc.model.FrameworkModel;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.stream.Collectors;
public class ConverterUtil {
private final FrameworkModel frameworkModel;
private final Map<Class<?>, Map<Class<?>, List<Converter>>> converterCache = new ConcurrentHashMap<>();
private final ConcurrentMap<Class<?>, ConcurrentMap<Class<?>, List<Converter>>> converterCache = new ConcurrentHashMap<>();
public ConverterUtil(FrameworkModel frameworkModel) {
this.frameworkModel = frameworkModel;
@ -41,8 +42,8 @@ public class ConverterUtil {
* @see ExtensionLoader#getSupportedExtensionInstances()
*/
public Converter<?, ?> getConverter(Class<?> sourceType, Class<?> targetType) {
Map<Class<?>, List<Converter>> toTargetMap = converterCache.computeIfAbsent(sourceType, (k) -> new ConcurrentHashMap<>());
List<Converter> converters = toTargetMap.computeIfAbsent(targetType, (k) -> frameworkModel.getExtensionLoader(Converter.class)
ConcurrentMap<Class<?>, List<Converter>> toTargetMap = ConcurrentHashMapUtils.computeIfAbsent(converterCache, sourceType, (k) -> new ConcurrentHashMap<>());
List<Converter> converters = ConcurrentHashMapUtils.computeIfAbsent(toTargetMap, targetType, (k) -> frameworkModel.getExtensionLoader(Converter.class)
.getSupportedExtensionInstances()
.stream()
.filter(converter -> converter.accept(sourceType, targetType))

9
dubbo-common/src/main/java/org/apache/dubbo/common/logger/LoggerFactory.java
View File

@ -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)));
}
/**

49
dubbo-common/src/main/java/org/apache/dubbo/common/metrics/collector/stat/MetricsStatComposite.java
View File

@ -20,6 +20,7 @@ package org.apache.dubbo.common.metrics.collector.stat;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.LongAccumulator;
@ -29,20 +30,22 @@ import org.apache.dubbo.common.metrics.event.RTEvent;
import org.apache.dubbo.common.metrics.event.RequestEvent;
import org.apache.dubbo.common.metrics.listener.MetricsListener;
import org.apache.dubbo.common.metrics.model.MethodMetric;
public class MetricsStatComposite{
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
public class MetricsStatComposite {
public Map<RequestEvent.Type, MetricsStatHandler> stats = new ConcurrentHashMap<>();
private final Map<MethodMetric, AtomicLong> lastRT = new ConcurrentHashMap<>();
private final Map<MethodMetric, LongAccumulator> minRT = new ConcurrentHashMap<>();
private final Map<MethodMetric, LongAccumulator> maxRT = new ConcurrentHashMap<>();
private final Map<MethodMetric, AtomicLong> avgRT = new ConcurrentHashMap<>();
private final Map<MethodMetric, AtomicLong> totalRT = new ConcurrentHashMap<>();
private final Map<MethodMetric, AtomicLong> rtCount = new ConcurrentHashMap<>();
private final ConcurrentMap<MethodMetric, AtomicLong> lastRT = new ConcurrentHashMap<>();
private final ConcurrentMap<MethodMetric, LongAccumulator> minRT = new ConcurrentHashMap<>();
private final ConcurrentMap<MethodMetric, LongAccumulator> maxRT = new ConcurrentHashMap<>();
private final ConcurrentMap<MethodMetric, AtomicLong> avgRT = new ConcurrentHashMap<>();
private final ConcurrentMap<MethodMetric, AtomicLong> totalRT = new ConcurrentHashMap<>();
private final ConcurrentMap<MethodMetric, AtomicLong> rtCount = new ConcurrentHashMap<>();
private final String applicationName;
private final List<MetricsListener> listeners;
private DefaultMetricsCollector collector;
public MetricsStatComposite(String applicationName, DefaultMetricsCollector collector){
public MetricsStatComposite(String applicationName, DefaultMetricsCollector collector) {
this.applicationName = applicationName;
this.listeners = collector.getListener();
this.collector = collector;
@ -53,23 +56,27 @@ public class MetricsStatComposite{
return stats.get(statType);
}
public Map<MethodMetric, AtomicLong> getLastRT(){
public Map<MethodMetric, AtomicLong> getLastRT() {
return this.lastRT;
}
public Map<MethodMetric, LongAccumulator> getMinRT(){
public Map<MethodMetric, LongAccumulator> getMinRT() {
return this.minRT;
}
public Map<MethodMetric, LongAccumulator> getMaxRT(){
public Map<MethodMetric, LongAccumulator> getMaxRT() {
return this.maxRT;
}
public Map<MethodMetric, AtomicLong> getAvgRT(){
public Map<MethodMetric, AtomicLong> getAvgRT() {
return this.avgRT;
}
public Map<MethodMetric, AtomicLong> getTotalRT(){
public Map<MethodMetric, AtomicLong> getTotalRT() {
return this.totalRT;
}
public Map<MethodMetric, AtomicLong> getRtCount(){
public Map<MethodMetric, AtomicLong> getRtCount() {
return this.rtCount;
}
@ -77,29 +84,29 @@ public class MetricsStatComposite{
if (collector.isCollectEnabled()) {
MethodMetric metric = new MethodMetric(applicationName, interfaceName, methodName, group, version);
AtomicLong last = lastRT.computeIfAbsent(metric, k -> new AtomicLong());
AtomicLong last = ConcurrentHashMapUtils.computeIfAbsent(lastRT, metric, k -> new AtomicLong());
last.set(responseTime);
LongAccumulator min = minRT.computeIfAbsent(metric, k -> new LongAccumulator(Long::min, Long.MAX_VALUE));
LongAccumulator min = ConcurrentHashMapUtils.computeIfAbsent(minRT, metric, k -> new LongAccumulator(Long::min, Long.MAX_VALUE));
min.accumulate(responseTime);
LongAccumulator max = maxRT.computeIfAbsent(metric, k -> new LongAccumulator(Long::max, Long.MIN_VALUE));
LongAccumulator max = ConcurrentHashMapUtils.computeIfAbsent(maxRT, metric, k -> new LongAccumulator(Long::max, Long.MIN_VALUE));
max.accumulate(responseTime);
AtomicLong total = totalRT.computeIfAbsent(metric, k -> new AtomicLong());
AtomicLong total = ConcurrentHashMapUtils.computeIfAbsent(totalRT, metric, k -> new AtomicLong());
total.addAndGet(responseTime);
AtomicLong count = rtCount.computeIfAbsent(metric, k -> new AtomicLong());
AtomicLong count = ConcurrentHashMapUtils.computeIfAbsent(rtCount, metric, k -> new AtomicLong());
count.incrementAndGet();
avgRT.computeIfAbsent(metric, k -> new AtomicLong());
ConcurrentHashMapUtils.computeIfAbsent(avgRT, metric, k -> new AtomicLong());
publishEvent(new RTEvent(metric, responseTime));
}
}
private void init() {
stats.put(RequestEvent.Type.TOTAL, new DefaultMetricsStatHandler(applicationName){
stats.put(RequestEvent.Type.TOTAL, new DefaultMetricsStatHandler(applicationName) {
@Override
public void doNotify(MethodMetric metric) {
publishEvent(new RequestEvent(metric, RequestEvent.Type.TOTAL));

5
dubbo-common/src/main/java/org/apache/dubbo/common/store/support/SimpleDataStore.java
View File

@ -18,6 +18,7 @@
package org.apache.dubbo.common.store.support;
import org.apache.dubbo.common.store.DataStore;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import java.util.HashMap;
import java.util.Map;
@ -28,7 +29,7 @@ public class SimpleDataStore implements DataStore {
// <component name or id, <data-name, data-value>>
private ConcurrentMap<String, ConcurrentMap<String, Object>> data =
new ConcurrentHashMap<String, ConcurrentMap<String, Object>>();
new ConcurrentHashMap<String, ConcurrentMap<String, Object>>();
@Override
public Map<String, Object> get(String componentName) {
@ -50,7 +51,7 @@ public class SimpleDataStore implements DataStore {
@Override
public void put(String componentName, String key, Object value) {
Map<String, Object> componentData = data.computeIfAbsent(componentName, k -> new ConcurrentHashMap<>());
Map<String, Object> componentData = ConcurrentHashMapUtils.computeIfAbsent(data, componentName, k -> new ConcurrentHashMap<>());
componentData.put(key, value);
}

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

@ -23,6 +23,7 @@ import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.store.DataStore;
import org.apache.dubbo.common.threadpool.ThreadPool;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.ExecutorUtil;
import org.apache.dubbo.common.utils.NamedThreadFactory;
import org.apache.dubbo.common.utils.StringUtils;
@ -93,14 +94,14 @@ public class DefaultExecutorRepository implements ExecutorRepository, ExtensionA
@Override
public synchronized ExecutorService createExecutorIfAbsent(URL url) {
String executorKey = getExecutorKey(url);
Map<String, ExecutorService> executors = data.computeIfAbsent(executorKey, k -> new ConcurrentHashMap<>());
ConcurrentMap<String, ExecutorService> executors = ConcurrentHashMapUtils.computeIfAbsent(data, executorKey, k -> new ConcurrentHashMap<>());
String executorCacheKey = getExecutorSecondKey(url);
url = setThreadNameIfAbsent(url, executorCacheKey);
URL finalUrl = url;
ExecutorService executor = executors.computeIfAbsent(executorCacheKey, k -> createExecutor(finalUrl));
ExecutorService executor = ConcurrentHashMapUtils.computeIfAbsent(executors, executorCacheKey, k -> createExecutor(finalUrl));
// If executor has been shut down, create a new one
if (executor.isShutdown() || executor.isTerminated()) {
executors.remove(executorCacheKey);

24
dubbo-common/src/main/java/org/apache/dubbo/common/url/component/ServiceConfigURL.java
View File

@ -18,18 +18,20 @@ package org.apache.dubbo.common.url.component;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.StringUtils;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
public class ServiceConfigURL extends URL {
private volatile transient Map<String, URL> urls;
private volatile transient Map<String, Number> numbers;
private volatile transient Map<String, Map<String, Number>> methodNumbers;
private volatile transient ConcurrentMap<String, URL> urls;
private volatile transient ConcurrentMap<String, Number> numbers;
private volatile transient ConcurrentMap<String, Map<String, Number>> methodNumbers;
private volatile transient String full;
private volatile transient String string;
private volatile transient String identity;
@ -77,12 +79,12 @@ public class ServiceConfigURL extends URL {
}
public ServiceConfigURL(String protocol,
String username,
String password,
String host,
int port,
String path,
Map<String, String> parameters) {
String username,
String password,
String host,
int port,
String path,
Map<String, String> parameters) {
this(new PathURLAddress(protocol, username, password, path, host, port), URLParam.parse(parameters), null);
}
@ -551,11 +553,11 @@ public class ServiceConfigURL extends URL {
}
private void updateCachedNumber(String method, String key, Number n) {
Map<String, Number> keyNumber = getMethodNumbers().computeIfAbsent(method, m -> new HashMap<>());
Map<String, Number> keyNumber = ConcurrentHashMapUtils.computeIfAbsent(getMethodNumbers(), method, m -> new HashMap<>());
keyNumber.put(key, n);
}
protected Map<String, Map<String, Number>> getMethodNumbers() {
protected ConcurrentMap<String, Map<String, Number>> getMethodNumbers() {
if (methodNumbers == null) { // concurrent initialization is tolerant
methodNumbers = new ConcurrentHashMap<>();
}

57
dubbo-common/src/main/java/org/apache/dubbo/common/utils/ConcurrentHashMapUtils.java Normal file
View File

@ -0,0 +1,57 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dubbo.common.utils;
import java.util.concurrent.ConcurrentMap;
import java.util.function.Function;
/**
* ConcurrentHashMap util
*/
public class ConcurrentHashMapUtils {
private static boolean IS_JAVA8;
static {
try {
IS_JAVA8 = System.getProperty("java.version").startsWith("1.8.");
} catch (Exception ignore) {
// exception is ignored
IS_JAVA8 = true;
}
}
/**
* A temporary workaround for Java 8 ConcurrentHashMap#computeIfAbsent specific performance issue: JDK-8161372.</br>
* @see <a href="https://bugs.openjdk.java.net/browse/JDK-8161372">https://bugs.openjdk.java.net/browse/JDK-8161372</a>
*
*/
public static <K, V> V computeIfAbsent(ConcurrentMap<K, V> map, K key, Function<? super K, ? extends V> func) {
if (IS_JAVA8) {
V v = map.get(key);
if (null == v) {
v = map.computeIfAbsent(key, func);
}
return v;
} else {
return map.computeIfAbsent(key, func);
}
}
}

10
dubbo-common/src/main/java/org/apache/dubbo/config/AbstractConfig.java
View File

@ -36,6 +36,7 @@ import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.dubbo.common.URL;
@ -48,6 +49,7 @@ import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.ClassUtils;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.FieldUtils;
import org.apache.dubbo.common.utils.MethodUtils;
import org.apache.dubbo.common.utils.ReflectUtils;
@ -81,12 +83,12 @@ public abstract class AbstractConfig implements Serializable {
/**
* tag name cache, speed up get tag name frequently
*/
private static final Map<Class, String> tagNameCache = new ConcurrentHashMap<>();
private static final ConcurrentMap<Class, String> tagNameCache = new ConcurrentHashMap<>();
/**
* attributed getter method cache for equals(), hashCode() and toString()
*/
private static final Map<Class, List<Method>> attributedMethodCache = new ConcurrentHashMap<>();
private static final ConcurrentMap<Class, List<Method>> attributedMethodCache = new ConcurrentHashMap<>();
/**
* The suffix container
@ -122,7 +124,7 @@ public abstract class AbstractConfig implements Serializable {
}
public static String getTagName(Class<?> cls) {
return tagNameCache.computeIfAbsent(cls, (key) -> {
return ConcurrentHashMapUtils.computeIfAbsent(tagNameCache, cls, (key) -> {
String tag = cls.getSimpleName();
for (String suffix : SUFFIXES) {
if (tag.endsWith(suffix)) {
@ -1031,7 +1033,7 @@ public abstract class AbstractConfig implements Serializable {
private List<Method> getAttributedMethods() {
Class<? extends AbstractConfig> cls = this.getClass();
return attributedMethodCache.computeIfAbsent(cls, (key) -> computeAttributedMethods());
return ConcurrentHashMapUtils.computeIfAbsent(attributedMethodCache, cls, (key) -> computeAttributedMethods());
}
/**

5
dubbo-common/src/main/java/org/apache/dubbo/rpc/model/FrameworkServiceRepository.java
View File

@ -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) {

9
dubbo-common/src/main/java/org/apache/dubbo/rpc/model/ModuleServiceRepository.java
View File

@ -17,6 +17,7 @@
package org.apache.dubbo.rpc.model;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.config.ReferenceConfigBase;
import org.apache.dubbo.config.ServiceConfigBase;
@ -81,7 +82,7 @@ public class ModuleServiceRepository {
}
public void registerConsumer(ConsumerModel consumerModel) {
consumers.computeIfAbsent(consumerModel.getServiceKey(), (serviceKey) -> new CopyOnWriteArrayList<>()).add(consumerModel);
ConcurrentHashMapUtils.computeIfAbsent(consumers, consumerModel.getServiceKey(), (serviceKey) -> new CopyOnWriteArrayList<>()).add(consumerModel);
}
/**
@ -119,7 +120,7 @@ public class ModuleServiceRepository {
}
public ServiceDescriptor registerService(Class<?> interfaceClazz, ServiceDescriptor serviceDescriptor) {
List<ServiceDescriptor> serviceDescriptors = services.computeIfAbsent(interfaceClazz.getName(),
List<ServiceDescriptor> serviceDescriptors = ConcurrentHashMapUtils.computeIfAbsent(services, interfaceClazz.getName(),
k -> new CopyOnWriteArrayList<>());
synchronized (serviceDescriptors) {
Optional<ServiceDescriptor> previous = serviceDescriptors.stream()
@ -149,7 +150,7 @@ public class ModuleServiceRepository {
ServiceDescriptor serviceDescriptor = registerService(interfaceClass);
// if path is different with interface name, add extra path mapping
if (!interfaceClass.getName().equals(path)) {
List<ServiceDescriptor> serviceDescriptors = services.computeIfAbsent(path,
List<ServiceDescriptor> serviceDescriptors = ConcurrentHashMapUtils.computeIfAbsent(services, path,
_k -> new CopyOnWriteArrayList<>());
synchronized (serviceDescriptors) {
Optional<ServiceDescriptor> previous = serviceDescriptors.stream()
@ -179,7 +180,7 @@ public class ModuleServiceRepository {
public void reRegisterConsumer(String newServiceKey, String serviceKey) {
List<ConsumerModel> consumerModel = this.consumers.get(serviceKey);
consumerModel.forEach(c -> c.setServiceKey(newServiceKey));
this.consumers.computeIfAbsent(newServiceKey, (k) -> new CopyOnWriteArrayList<>()).addAll(consumerModel);
ConcurrentHashMapUtils.computeIfAbsent(this.consumers, newServiceKey, (k) -> new CopyOnWriteArrayList<>()).addAll(consumerModel);
this.consumers.remove(serviceKey);
}

40
dubbo-common/src/test/java/org/apache/dubbo/common/utils/ConcurrentHashMapUtilsTest.java Normal file
View File

@ -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<String, String> map = new ConcurrentHashMap<>();
String ifAbsent = ConcurrentHashMapUtils.computeIfAbsent(map, "mxsm", k -> "mxsm");
assertEquals("mxsm", ifAbsent);
ifAbsent = ConcurrentHashMapUtils.computeIfAbsent(map, "mxsm", k -> "mxsm1");
assertEquals("mxsm", ifAbsent);
map.remove("mxsm");
ifAbsent = ConcurrentHashMapUtils.computeIfAbsent(map, "mxsm", k -> "mxsm1");
assertEquals("mxsm1", ifAbsent);
}
}

12
dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/bootstrap/DubboBootstrap.java
View File

@ -25,6 +25,7 @@ import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.threadpool.manager.ExecutorRepository;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.config.ApplicationConfig;
import org.apache.dubbo.config.ConfigCenterConfig;
import org.apache.dubbo.config.ConsumerConfig;
@ -53,8 +54,8 @@ import org.apache.dubbo.rpc.model.FrameworkModel;
import org.apache.dubbo.rpc.model.ModuleModel;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Condition;
@ -80,7 +81,7 @@ public final class DubboBootstrap {
private static final Logger logger = LoggerFactory.getLogger(DubboBootstrap.class);
private static volatile Map<ApplicationModel, DubboBootstrap> instanceMap = new ConcurrentHashMap<>();
private static volatile ConcurrentMap<ApplicationModel, DubboBootstrap> instanceMap = new ConcurrentHashMap<>();
private static volatile DubboBootstrap instance;
private final AtomicBoolean awaited = new AtomicBoolean(false);
@ -116,7 +117,7 @@ public final class DubboBootstrap {
}
public static DubboBootstrap getInstance(ApplicationModel applicationModel) {
return instanceMap.computeIfAbsent(applicationModel, _k -> new DubboBootstrap(applicationModel));
return ConcurrentHashMapUtils.computeIfAbsent(instanceMap, applicationModel, _k -> new DubboBootstrap(applicationModel));
}
public static DubboBootstrap newInstance() {
@ -217,6 +218,7 @@ public final class DubboBootstrap {
/**
* Start dubbo application
*
* @param wait If true, wait for startup to complete, or else no waiting.
* @return
*/
@ -234,6 +236,7 @@ public final class DubboBootstrap {
/**
* Start dubbo application but no wait for finish.
*
* @return the future object
*/
public Future asyncStart() {
@ -242,6 +245,7 @@ public final class DubboBootstrap {
/**
* Stop dubbo application
*
* @return
* @throws IllegalStateException
*/
@ -369,7 +373,7 @@ public final class DubboBootstrap {
return metadataReport(null, consumerBuilder);
}
public DubboBootstrap metadataReport(String id,Consumer<MetadataReportBuilder> consumerBuilder) {
public DubboBootstrap metadataReport(String id, Consumer<MetadataReportBuilder> consumerBuilder) {
MetadataReportBuilder metadataReportBuilder = createMetadataReportBuilder(id);
consumerBuilder.accept(metadataReportBuilder);
return this;

13
dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/utils/SimpleReferenceCache.java
View File

@ -21,6 +21,7 @@ import org.apache.dubbo.common.config.ReferenceCache;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.config.ReferenceConfigBase;
import org.apache.dubbo.rpc.service.Destroyable;
@ -70,8 +71,8 @@ public class SimpleReferenceCache implements ReferenceCache {
private final String name;
private final KeyGenerator generator;
private final Map<String, List<ReferenceConfigBase<?>>> referenceKeyMap = new ConcurrentHashMap<>();
private final Map<Class<?>, List<ReferenceConfigBase<?>>> referenceTypeMap = new ConcurrentHashMap<>();
private final ConcurrentMap<String, List<ReferenceConfigBase<?>>> referenceKeyMap = new ConcurrentHashMap<>();
private final ConcurrentMap<Class<?>, List<ReferenceConfigBase<?>>> referenceTypeMap = new ConcurrentHashMap<>();
private final Map<ReferenceConfigBase<?>, Object> references = new ConcurrentHashMap<>();
protected SimpleReferenceCache(String name, KeyGenerator generator) {
@ -104,7 +105,7 @@ public class SimpleReferenceCache implements ReferenceCache {
* Create cache if not existed yet.
*/
public static SimpleReferenceCache getCache(String name, KeyGenerator keyGenerator) {
return CACHE_HOLDER.computeIfAbsent(name, k -> new SimpleReferenceCache(k, keyGenerator));
return ConcurrentHashMapUtils.computeIfAbsent(CACHE_HOLDER, name, k -> new SimpleReferenceCache(k, keyGenerator));
}
@Override
@ -124,9 +125,9 @@ public class SimpleReferenceCache implements ReferenceCache {
}
if (proxy == null) {
List<ReferenceConfigBase<?>> referencesOfType = referenceTypeMap.computeIfAbsent(type, _t -> Collections.synchronizedList(new ArrayList<>()));
List<ReferenceConfigBase<?>> referencesOfType = ConcurrentHashMapUtils.computeIfAbsent(referenceTypeMap, type, _t -> Collections.synchronizedList(new ArrayList<>()));
referencesOfType.add(rc);
List<ReferenceConfigBase<?>> referenceConfigList = referenceKeyMap.computeIfAbsent(key, _k -> Collections.synchronizedList(new ArrayList<>()));
List<ReferenceConfigBase<?>> referenceConfigList = ConcurrentHashMapUtils.computeIfAbsent(referenceKeyMap, key, _k -> Collections.synchronizedList(new ArrayList<>()));
referenceConfigList.add(rc);
proxy = rc.get();
}
@ -262,7 +263,7 @@ public class SimpleReferenceCache implements ReferenceCache {
private void destroyReference(ReferenceConfigBase<?> rc) {
Destroyable proxy = (Destroyable) rc.get();
if (proxy != null){
if (proxy != null) {
proxy.$destroy();
}
rc.destroy();

12
dubbo-config/dubbo-config-spring/src/main/java/org/apache/dubbo/config/spring/reference/ReferenceBeanManager.java
View File

@ -19,6 +19,7 @@ package org.apache.dubbo.config.spring.reference;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.Assert;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.config.ReferenceConfig;
import org.apache.dubbo.config.spring.ReferenceBean;
@ -37,6 +38,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.CONFIG_DUBBO_BEAN_INITIALIZER;
@ -46,16 +48,16 @@ public class ReferenceBeanManager implements ApplicationContextAware {
private final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(getClass());
//reference key -> reference bean names
private Map<String, List<String>> referenceKeyMap = new ConcurrentHashMap<>();
private ConcurrentMap<String, List<String>> referenceKeyMap = new ConcurrentHashMap<>();
// reference alias -> reference bean name
private Map<String, String> referenceAliasMap = new ConcurrentHashMap<>();
private ConcurrentMap<String, String> referenceAliasMap = new ConcurrentHashMap<>();
//reference bean name -> ReferenceBean
private Map<String, ReferenceBean> referenceBeanMap = new ConcurrentHashMap<>();
private ConcurrentMap<String, ReferenceBean> referenceBeanMap = new ConcurrentHashMap<>();
//reference key -> ReferenceConfig instance
private Map<String, ReferenceConfig> referenceConfigMap = new ConcurrentHashMap<>();
private ConcurrentMap<String, ReferenceConfig> referenceConfigMap = new ConcurrentHashMap<>();
private ApplicationContext applicationContext;
private volatile boolean initialized = false;
@ -105,7 +107,7 @@ public class ReferenceBeanManager implements ApplicationContextAware {
}
public void registerReferenceKeyAndBeanName(String referenceKey, String referenceBeanNameOrAlias) {
List<String> list = referenceKeyMap.computeIfAbsent(referenceKey, (key) -> new ArrayList<>());
List<String> list = ConcurrentHashMapUtils.computeIfAbsent(referenceKeyMap, referenceKey, (key) -> new ArrayList<>());
if (!list.contains(referenceBeanNameOrAlias)) {
list.add(referenceBeanNameOrAlias);
// register bean name as alias

7
dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/AbstractRegistryService.java
View File

@ -20,6 +20,7 @@ import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.registry.NotifyListener;
import org.apache.dubbo.registry.RegistryService;
@ -102,7 +103,7 @@ public abstract class AbstractRegistryService implements RegistryService {
if (url == null) {
throw new IllegalArgumentException("url == null");
}
List<URL> urls = registered.computeIfAbsent(service, k -> new CopyOnWriteArrayList<>());
List<URL> urls = ConcurrentHashMapUtils.computeIfAbsent(registered, service, k -> new CopyOnWriteArrayList<>());
if (!urls.contains(url)) {
urls.add(url);
}
@ -162,7 +163,7 @@ public abstract class AbstractRegistryService implements RegistryService {
if (listener == null) {
return;
}
List<NotifyListener> listeners = notifyListeners.computeIfAbsent(service, k -> new CopyOnWriteArrayList<>());
List<NotifyListener> listeners = ConcurrentHashMapUtils.computeIfAbsent(notifyListeners, service, k -> new CopyOnWriteArrayList<>());
if (!listeners.contains(listener)) {
listeners.add(listener);
}
@ -202,7 +203,7 @@ public abstract class AbstractRegistryService implements RegistryService {
protected final void notify(String service, List<URL> urls) {
if (StringUtils.isEmpty(service)
|| CollectionUtils.isEmpty(urls)) {
|| CollectionUtils.isEmpty(urls)) {
return;
}
doNotify(service, urls);

21
dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/SimpleRegistryService.java
View File

@ -21,6 +21,7 @@ import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.url.component.ServiceConfigURL;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.NetUtils;
import org.apache.dubbo.common.utils.UrlUtils;
import org.apache.dubbo.registry.NotifyListener;
@ -46,7 +47,7 @@ public class SimpleRegistryService extends AbstractRegistryService {
public void register(String service, URL url) {
super.register(service, url);
String client = RpcContext.getServiceContext().getRemoteAddressString();
Map<String, URL> urls = remoteRegistered.computeIfAbsent(client, k -> new ConcurrentHashMap<>());
Map<String, URL> urls = ConcurrentHashMapUtils.computeIfAbsent(remoteRegistered, client, k -> new ConcurrentHashMap<>());
urls.put(service, url);
notify(service, getRegistered().get(service));
}
@ -70,12 +71,12 @@ public class SimpleRegistryService extends AbstractRegistryService {
}
List<URL> urls = getRegistered().get(service);
if ((RegistryService.class.getName() + ":0.0.0").equals(service)
&& CollectionUtils.isEmpty(urls)) {
&& CollectionUtils.isEmpty(urls)) {
register(service, new ServiceConfigURL("dubbo",
NetUtils.getLocalHost(),
RpcContext.getServiceContext().getLocalPort(),
RegistryService.class.getName(),
url.getParameters()));
NetUtils.getLocalHost(),
RpcContext.getServiceContext().getLocalPort(),
RegistryService.class.getName(),
url.getParameters()));
List<String> rs = registries;
if (rs != null && rs.size() > 0) {
for (String registry : rs) {
@ -85,7 +86,7 @@ public class SimpleRegistryService extends AbstractRegistryService {
}
super.subscribe(service, url, listener);
Map<String, NotifyListener> listeners = remoteListeners.computeIfAbsent(client, k -> new ConcurrentHashMap<>());
Map<String, NotifyListener> listeners = ConcurrentHashMapUtils.computeIfAbsent(remoteListeners, client, k -> new ConcurrentHashMap<>());
listeners.put(service, listener);
urls = getRegistered().get(service);
if (urls != null && urls.size() > 0) {
@ -125,9 +126,9 @@ public class SimpleRegistryService extends AbstractRegistryService {
for (Map.Entry<String, NotifyListener> entry : listeners.entrySet()) {
String service = entry.getKey();
super.unsubscribe(service, new ServiceConfigURL("subscribe",
RpcContext.getServiceContext().getRemoteHost(),
RpcContext.getServiceContext().getRemotePort(),
RegistryService.class.getName(), getSubscribed(service)), entry.getValue());
RpcContext.getServiceContext().getRemoteHost(),
RpcContext.getServiceContext().getRemotePort(),
RegistryService.class.getName(), getSubscribed(service)), entry.getValue());
}
}
}

7
dubbo-configcenter/dubbo-configcenter-nacos/src/main/java/org/apache/dubbo/configcenter/support/nacos/NacosDynamicConfiguration.java
View File

@ -21,6 +21,7 @@ import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.Executor;
@ -33,6 +34,7 @@ import org.apache.dubbo.common.config.configcenter.DynamicConfiguration;
import org.apache.dubbo.common.constants.LoggerCodeConstants;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.MD5Utils;
import org.apache.dubbo.common.utils.StringUtils;
@ -81,7 +83,7 @@ public class NacosDynamicConfiguration implements DynamicConfiguration {
/**
* The map store the key to {@link NacosConfigListener} mapping
*/
private final Map<String, NacosConfigListener> watchListenerMap;
private final ConcurrentMap<String, NacosConfigListener> watchListenerMap;
private MD5Utils md5Utils = new MD5Utils();
@ -209,8 +211,7 @@ public class NacosDynamicConfiguration implements DynamicConfiguration {
@Override
public void addListener(String key, String group, ConfigurationListener listener) {
String listenerKey = buildListenerKey(key, group);
NacosConfigListener nacosConfigListener =
watchListenerMap.computeIfAbsent(listenerKey, k -> createTargetListener(key, group));
NacosConfigListener nacosConfigListener = ConcurrentHashMapUtils.computeIfAbsent(watchListenerMap, listenerKey, k -> createTargetListener(key, group));
nacosConfigListener.addListener(listener);
try {
configService.addListener(key, group, nacosConfigListener);

6
dubbo-configcenter/dubbo-configcenter-zookeeper/src/main/java/org/apache/dubbo/configcenter/support/zookeeper/CacheListener.java
View File

@ -18,22 +18,24 @@ package org.apache.dubbo.configcenter.support.zookeeper;
import org.apache.dubbo.common.config.configcenter.ConfigurationListener;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
/**
* one path has one zookeeperDataListener
*/
public class CacheListener {
private Map<String, ZookeeperDataListener> pathKeyListeners = new ConcurrentHashMap<>();
private ConcurrentMap<String, ZookeeperDataListener> pathKeyListeners = new ConcurrentHashMap<>();
public CacheListener() {
}
public ZookeeperDataListener addListener(String pathKey, ConfigurationListener configurationListener, String key, String group) {
ZookeeperDataListener zookeeperDataListener = pathKeyListeners.computeIfAbsent(pathKey,
ZookeeperDataListener zookeeperDataListener = ConcurrentHashMapUtils.computeIfAbsent(pathKeyListeners, pathKey,
_pathKey -> new ZookeeperDataListener(_pathKey, key, group));
zookeeperDataListener.addListener(configurationListener);
return zookeeperDataListener;

6
dubbo-metadata/dubbo-metadata-report-zookeeper/src/main/java/org/apache/dubbo/metadata/store/zookeeper/ZookeeperMetadataReport.java
View File

@ -20,6 +20,7 @@ import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.config.configcenter.ConfigItem;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.JsonUtils;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.metadata.MappingChangedEvent;
@ -45,6 +46,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import static org.apache.dubbo.common.constants.CommonConstants.PATH_SEPARATOR;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_ZOOKEEPER_EXCEPTION;
@ -62,7 +64,7 @@ public class ZookeeperMetadataReport extends AbstractMetadataReport {
ZookeeperClient zkClient;
private Map<String, MappingDataListener> casListenerMap = new ConcurrentHashMap<>();
private ConcurrentMap<String, MappingDataListener> casListenerMap = new ConcurrentHashMap<>();
public ZookeeperMetadataReport(URL url, ZookeeperTransporter zookeeperTransporter) {
@ -162,7 +164,7 @@ public class ZookeeperMetadataReport extends AbstractMetadataReport {
@Override
public Set<String> getServiceAppMapping(String serviceKey, MappingListener listener, URL url) {
String path = buildPathKey(DEFAULT_MAPPING_GROUP, serviceKey);
MappingDataListener mappingDataListener = casListenerMap.computeIfAbsent(path, _k -> {
MappingDataListener mappingDataListener = ConcurrentHashMapUtils.computeIfAbsent(casListenerMap, path, _k -> {
MappingDataListener newMappingListener = new MappingDataListener(serviceKey, path);
zkClient.addDataListener(path, newMappingListener);
return newMappingListener;

39
dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/collector/AggregateMetricsCollector.java
View File

@ -27,6 +27,7 @@ import org.apache.dubbo.common.metrics.model.MethodMetric;
import org.apache.dubbo.common.metrics.model.MetricsKey;
import org.apache.dubbo.common.metrics.model.sample.GaugeMetricSample;
import org.apache.dubbo.common.metrics.model.sample.MetricSample;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.config.MetricsConfig;
import org.apache.dubbo.config.context.ConfigManager;
import org.apache.dubbo.config.nested.AggregationConfig;
@ -36,8 +37,8 @@ import org.apache.dubbo.rpc.model.ApplicationModel;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import static org.apache.dubbo.common.metrics.model.MetricsCategory.REQUESTS;
import static org.apache.dubbo.common.metrics.model.MetricsCategory.QPS;
@ -51,15 +52,15 @@ public class AggregateMetricsCollector implements MetricsCollector, MetricsListe
private int bucketNum;
private int timeWindowSeconds;
private final Map<MethodMetric, TimeWindowCounter> totalRequests = new ConcurrentHashMap<>();
private final Map<MethodMetric, TimeWindowCounter> succeedRequests = new ConcurrentHashMap<>();
private final Map<MethodMetric, TimeWindowCounter> failedRequests = new ConcurrentHashMap<>();
private final Map<MethodMetric, TimeWindowCounter> businessFailedRequests = new ConcurrentHashMap<>();
private final Map<MethodMetric, TimeWindowCounter> timeoutRequests = new ConcurrentHashMap<>();
private final Map<MethodMetric, TimeWindowCounter> limitRequests = new ConcurrentHashMap<>();
private final Map<MethodMetric, TimeWindowCounter> totalFailedRequests = new ConcurrentHashMap<>();
private final Map<MethodMetric, TimeWindowCounter> qps = new ConcurrentHashMap<>();
private final Map<MethodMetric, TimeWindowQuantile> rt = new ConcurrentHashMap<>();
private final ConcurrentMap<MethodMetric, TimeWindowCounter> totalRequests = new ConcurrentHashMap<>();
private final ConcurrentMap<MethodMetric, TimeWindowCounter> succeedRequests = new ConcurrentHashMap<>();
private final ConcurrentMap<MethodMetric, TimeWindowCounter> failedRequests = new ConcurrentHashMap<>();
private final ConcurrentMap<MethodMetric, TimeWindowCounter> businessFailedRequests = new ConcurrentHashMap<>();
private final ConcurrentMap<MethodMetric, TimeWindowCounter> timeoutRequests = new ConcurrentHashMap<>();
private final ConcurrentMap<MethodMetric, TimeWindowCounter> limitRequests = new ConcurrentHashMap<>();
private final ConcurrentMap<MethodMetric, TimeWindowCounter> totalFailedRequests = new ConcurrentHashMap<>();
private final ConcurrentMap<MethodMetric, TimeWindowCounter> qps = new ConcurrentHashMap<>();
private final ConcurrentMap<MethodMetric, TimeWindowQuantile> rt = new ConcurrentHashMap<>();
private final ApplicationModel applicationModel;
@ -97,7 +98,7 @@ public class AggregateMetricsCollector implements MetricsCollector, MetricsListe
private void onRTEvent(RTEvent event) {
MethodMetric metric = (MethodMetric) event.getSource();
Long responseTime = event.getRt();
TimeWindowQuantile quantile = rt.computeIfAbsent(metric, k -> new TimeWindowQuantile(DEFAULT_COMPRESSION, bucketNum, timeWindowSeconds));
TimeWindowQuantile quantile = ConcurrentHashMapUtils.computeIfAbsent(rt, metric, k -> new TimeWindowQuantile(DEFAULT_COMPRESSION, bucketNum, timeWindowSeconds));
quantile.add(responseTime);
}
@ -107,30 +108,30 @@ public class AggregateMetricsCollector implements MetricsCollector, MetricsListe
TimeWindowCounter counter = null;
switch (type) {
case TOTAL:
counter = totalRequests.computeIfAbsent(metric, k -> new TimeWindowCounter(bucketNum, timeWindowSeconds));
TimeWindowCounter qpsCounter = qps.computeIfAbsent(metric, k -> new TimeWindowCounter(bucketNum, timeWindowSeconds));
counter = ConcurrentHashMapUtils.computeIfAbsent(totalRequests, metric, k -> new TimeWindowCounter(bucketNum, timeWindowSeconds));
TimeWindowCounter qpsCounter = ConcurrentHashMapUtils.computeIfAbsent(qps, metric, k -> new TimeWindowCounter(bucketNum, timeWindowSeconds));
qpsCounter.increment();
break;
case SUCCEED:
counter = succeedRequests.computeIfAbsent(metric, k -> new TimeWindowCounter(bucketNum, timeWindowSeconds));
counter = ConcurrentHashMapUtils.computeIfAbsent(succeedRequests, metric, k -> new TimeWindowCounter(bucketNum, timeWindowSeconds));
break;
case FAILED:
counter = failedRequests.computeIfAbsent(metric, k -> new TimeWindowCounter(bucketNum, timeWindowSeconds));
counter = ConcurrentHashMapUtils.computeIfAbsent(failedRequests, metric, k -> new TimeWindowCounter(bucketNum, timeWindowSeconds));
break;
case BUSINESS_FAILED:
counter = businessFailedRequests.computeIfAbsent(metric, k -> new TimeWindowCounter(bucketNum, timeWindowSeconds));
counter = ConcurrentHashMapUtils.computeIfAbsent(businessFailedRequests, metric, k -> new TimeWindowCounter(bucketNum, timeWindowSeconds));
break;
case REQUEST_TIMEOUT:
counter = timeoutRequests.computeIfAbsent(metric, k -> new TimeWindowCounter(bucketNum, timeWindowSeconds));
counter = ConcurrentHashMapUtils.computeIfAbsent(timeoutRequests, metric, k -> new TimeWindowCounter(bucketNum, timeWindowSeconds));
break;
case REQUEST_LIMIT:
counter = limitRequests.computeIfAbsent(metric, k -> new TimeWindowCounter(bucketNum, timeWindowSeconds));
counter = ConcurrentHashMapUtils.computeIfAbsent(limitRequests, metric, k -> new TimeWindowCounter(bucketNum, timeWindowSeconds));
break;
case TOTAL_FAILED:
counter = totalFailedRequests.computeIfAbsent(metric, k -> new TimeWindowCounter(bucketNum, timeWindowSeconds));
counter = ConcurrentHashMapUtils.computeIfAbsent(totalFailedRequests, metric, k -> new TimeWindowCounter(bucketNum, timeWindowSeconds));
break;
default:

3
dubbo-monitor/dubbo-monitor-api/src/main/java/org/apache/dubbo/monitor/support/MonitorFilter.java
View File

@ -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

3
dubbo-monitor/dubbo-monitor-default/src/main/java/org/apache/dubbo/monitor/dubbo/DubboMonitor.java
View File

@ -20,6 +20,7 @@ import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.threadpool.manager.FrameworkExecutorRepository;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.ExecutorUtil;
import org.apache.dubbo.monitor.Monitor;
import org.apache.dubbo.monitor.MonitorService;
@ -152,7 +153,7 @@ public class DubboMonitor implements Monitor {
int concurrent = url.getParameter(CONCURRENT_KEY, 0);
// init atomic reference
Statistics statistics = new Statistics(url);
AtomicReference<StatisticsItem> reference = statisticsMap.computeIfAbsent(statistics, k -> new AtomicReference<>());
AtomicReference<StatisticsItem> reference = ConcurrentHashMapUtils.computeIfAbsent(statisticsMap, statistics, k -> new AtomicReference<>());
// use CompareAndSet to sum
StatisticsItem current;
StatisticsItem update = new StatisticsItem();

3
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/AbstractServiceDiscoveryFactory.java
View File

@ -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);

3
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ReflectionBasedServiceDiscovery.java
View File

@ -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) {

3
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistry.java
View File

@ -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) {

6
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistryDirectory.java
View File

@ -26,6 +26,7 @@ import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.stream.Collectors;
import org.apache.dubbo.common.ProtocolServiceKey;
@ -37,6 +38,7 @@ import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.Assert;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.NetUtils;
import org.apache.dubbo.metadata.MetadataInfo;
import org.apache.dubbo.registry.AddressListener;
@ -86,7 +88,7 @@ public class ServiceDiscoveryRegistryDirectory<T> extends DynamicDirectory<T> {
private final Set<String> providerFirstParams;
private final ModuleModel moduleModel;
private final ProtocolServiceKey consumerProtocolServiceKey;
private final Map<ProtocolServiceKey, URL> customizedConsumerUrlMap = new ConcurrentHashMap<>();
private final ConcurrentMap<ProtocolServiceKey, URL> customizedConsumerUrlMap = new ConcurrentHashMap<>();
public ServiceDiscoveryRegistryDirectory(Class<T> serviceType, URL url) {
super(serviceType, url);
@ -351,7 +353,7 @@ public class ServiceDiscoveryRegistryDirectory<T> extends DynamicDirectory<T> {
}
if (enabled) {
if (shouldWrap) {
URL newConsumerUrl = customizedConsumerUrlMap.computeIfAbsent(matchedProtocolServiceKey,
URL newConsumerUrl = ConcurrentHashMapUtils.computeIfAbsent(customizedConsumerUrlMap, matchedProtocolServiceKey,
k -> consumerUrl.setProtocol(k.getProtocol())
.addParameter(CommonConstants.GROUP_KEY, k.getGroup())
.addParameter(CommonConstants.VERSION_KEY, k.getVersion()));

6
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/migration/DefaultMigrationAddressComparator.java
View File

@ -20,6 +20,7 @@ import org.apache.dubbo.common.config.ConfigurationUtils;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.registry.client.migration.model.MigrationRule;
import org.apache.dubbo.rpc.Invoker;
@ -28,6 +29,7 @@ import org.apache.dubbo.rpc.cluster.ClusterInvoker;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.COMMON_PROPERTY_TYPE_MISMATCH;
@ -40,11 +42,11 @@ public class DefaultMigrationAddressComparator implements MigrationAddressCompar
public static final String OLD_ADDRESS_SIZE = "OLD_ADDRESS_SIZE";
public static final String NEW_ADDRESS_SIZE = "NEW_ADDRESS_SIZE";
private Map<String, Map<String, Integer>> serviceMigrationData = new ConcurrentHashMap<>();
private ConcurrentMap<String, Map<String, Integer>> serviceMigrationData = new ConcurrentHashMap<>();
@Override
public <T> boolean shouldMigrate(ClusterInvoker<T> newInvoker, ClusterInvoker<T> oldInvoker, MigrationRule rule) {
Map<String, Integer> migrationData = serviceMigrationData.computeIfAbsent(oldInvoker.getUrl().getDisplayServiceKey(), _k -> new ConcurrentHashMap<>());
Map<String, Integer> migrationData = ConcurrentHashMapUtils.computeIfAbsent(serviceMigrationData, oldInvoker.getUrl().getDisplayServiceKey(), _k -> new ConcurrentHashMap<>());
if (!newInvoker.hasProxyInvokers()) {
migrationData.put(OLD_ADDRESS_SIZE, getAddressSize(oldInvoker));

6
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/migration/MigrationRuleListener.java
View File

@ -26,6 +26,7 @@ import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.threadpool.manager.FrameworkExecutorRepository;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.NamedThreadFactory;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.registry.client.migration.model.MigrationRule;
@ -40,6 +41,7 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@ -73,7 +75,7 @@ public class MigrationRuleListener implements RegistryProtocolListener, Configur
private static final int MIGRATION_DEFAULT_DELAY_TIME = 60000;
private String ruleKey;
protected final Map<MigrationInvoker, MigrationRuleHandler> handlers = new ConcurrentHashMap<>();
protected final ConcurrentMap<MigrationInvoker, MigrationRuleHandler> handlers = new ConcurrentHashMap<>();
protected final LinkedBlockingQueue<String> ruleQueue = new LinkedBlockingQueue<>();
private final AtomicBoolean executorSubmit = new AtomicBoolean(false);
@ -239,7 +241,7 @@ public class MigrationRuleListener implements RegistryProtocolListener, Configur
@Override
public void onRefer(RegistryProtocol registryProtocol, ClusterInvoker<?> invoker, URL consumerUrl, URL registryURL) {
MigrationRuleHandler<?> migrationRuleHandler = handlers.computeIfAbsent((MigrationInvoker<?>) invoker, _key -> {
MigrationRuleHandler<?> migrationRuleHandler = ConcurrentHashMapUtils.computeIfAbsent(handlers, (MigrationInvoker<?>) invoker, _key -> {
((MigrationInvoker<?>) invoker).setMigrationRuleListener(this);
return new MigrationRuleHandler<>((MigrationInvoker<?>) invoker, consumerUrl);
});

14
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/support/CacheableFailbackRegistry.java
View File

@ -28,6 +28,7 @@ import org.apache.dubbo.common.url.component.ServiceAddressURL;
import org.apache.dubbo.common.url.component.URLAddress;
import org.apache.dubbo.common.url.component.URLParam;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.common.utils.UrlUtils;
import org.apache.dubbo.registry.NotifyListener;
@ -42,6 +43,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
@ -83,8 +85,8 @@ public abstract class CacheableFailbackRegistry extends FailbackRegistry {
private static String[] VARIABLE_KEYS = new String[]{ENCODED_TIMESTAMP_KEY, ENCODED_PID_KEY};
protected Map<String, URLAddress> stringAddress = new ConcurrentHashMap<>();
protected Map<String, URLParam> stringParam = new ConcurrentHashMap<>();
protected ConcurrentMap<String, URLAddress> stringAddress = new ConcurrentHashMap<>();
protected ConcurrentMap<String, URLParam> stringParam = new ConcurrentHashMap<>();
private ScheduledExecutorService cacheRemovalScheduler;
private int cacheRemovalTaskIntervalInMillis;
@ -249,8 +251,8 @@ public abstract class CacheableFailbackRegistry extends FailbackRegistry {
/**
* Create DubboServiceAddress object using provider url, consumer url, and extra parameters.
*
* @param rawProvider provider url string
* @param consumerURL URL object of consumer
* @param rawProvider provider url string
* @param consumerURL URL object of consumer
* @param extraParameters extra parameters
* @return created DubboServiceAddressURL object
*/
@ -286,10 +288,10 @@ public abstract class CacheableFailbackRegistry extends FailbackRegistry {
boolean isEncoded = encoded;
// PathURLAddress if it's using dubbo protocol.
URLAddress address = stringAddress.computeIfAbsent(rawAddress, k -> URLAddress.parse(k, getDefaultURLProtocol(), isEncoded));
URLAddress address = ConcurrentHashMapUtils.computeIfAbsent(stringAddress, rawAddress, k -> URLAddress.parse(k, getDefaultURLProtocol(), isEncoded));
address.setTimestamp(System.currentTimeMillis());
URLParam param = stringParam.computeIfAbsent(rawParams, k -> URLParam.parse(k, isEncoded, extraParameters));
URLParam param = ConcurrentHashMapUtils.computeIfAbsent(stringParam, rawParams, k -> URLParam.parse(k, isEncoded, extraParameters));
param.setTimestamp(System.currentTimeMillis());
// create service URL using cached address, param, and consumer URL.

6
dubbo-registry/dubbo-registry-multiple/src/main/java/org/apache/dubbo/registry/multiple/MultipleServiceDiscovery.java
View File

@ -22,10 +22,12 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.function.Function;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.metadata.MetadataInfo;
import org.apache.dubbo.registry.NotifyListener;
import org.apache.dubbo.registry.client.ServiceDiscovery;
@ -178,7 +180,7 @@ public class MultipleServiceDiscovery implements ServiceDiscovery {
}
protected static class MultiServiceInstancesChangedListener extends ServiceInstancesChangedListener {
private final Map<String, SingleServiceInstancesChangedListener> singleListenerMap;
private final ConcurrentMap<String, SingleServiceInstancesChangedListener> singleListenerMap;
public MultiServiceInstancesChangedListener(Set<String> serviceNames, ServiceDiscovery serviceDiscovery) {
super(serviceNames, serviceDiscovery);
@ -207,7 +209,7 @@ public class MultipleServiceDiscovery implements ServiceDiscovery {
public SingleServiceInstancesChangedListener getAndComputeIfAbsent(String registryKey,
Function<String, SingleServiceInstancesChangedListener> func) {
return singleListenerMap.computeIfAbsent(registryKey, func);
return ConcurrentHashMapUtils.computeIfAbsent(singleListenerMap, registryKey, func);
}
}

16
dubbo-registry/dubbo-registry-nacos/src/main/java/org/apache/dubbo/registry/nacos/NacosNamingServiceWrapper.java
View File

@ -23,6 +23,7 @@ import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;
@ -30,6 +31,7 @@ import java.util.stream.Collectors;
import org.apache.dubbo.common.constants.LoggerCodeConstants;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.MethodUtils;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.registry.nacos.function.NacosConsumer;
@ -56,8 +58,8 @@ public class NacosNamingServiceWrapper {
private final boolean isSupportBatchRegister;
private final Map<InstanceId, InstancesInfo> registerStatus = new ConcurrentHashMap<>();
private final Map<SubscribeInfo, NamingService> subscribeStatus = new ConcurrentHashMap<>();
private final ConcurrentMap<InstanceId, InstancesInfo> registerStatus = new ConcurrentHashMap<>();
private final ConcurrentMap<SubscribeInfo, NamingService> subscribeStatus = new ConcurrentHashMap<>();
private final Lock mapLock = new ReentrantLock();
public NacosNamingServiceWrapper(NacosConnectionManager nacosConnectionManager, int retryTimes, int sleepMsBetweenRetries) {
@ -86,7 +88,7 @@ public class NacosNamingServiceWrapper {
public void subscribe(String serviceName, String group, EventListener eventListener) throws NacosException {
String nacosServiceName = handleInnerSymbol(serviceName);
SubscribeInfo subscribeInfo = new SubscribeInfo(nacosServiceName, group, eventListener);
NamingService namingService = subscribeStatus.computeIfAbsent(subscribeInfo, info -> nacosConnectionManager.getNamingService());
NamingService namingService = ConcurrentHashMapUtils.computeIfAbsent(subscribeStatus, subscribeInfo, info -> nacosConnectionManager.getNamingService());
accept(() -> namingService.subscribe(nacosServiceName, group, eventListener));
}
@ -109,7 +111,7 @@ public class NacosNamingServiceWrapper {
InstancesInfo instancesInfo;
try {
mapLock.lock();
instancesInfo = registerStatus.computeIfAbsent(new InstanceId(nacosServiceName, group), id -> new InstancesInfo());
instancesInfo = ConcurrentHashMapUtils.computeIfAbsent(registerStatus, new InstanceId(nacosServiceName, group), id -> new InstancesInfo());
} finally {
mapLock.unlock();
}
@ -177,7 +179,7 @@ public class NacosNamingServiceWrapper {
InstancesInfo instancesInfo;
try {
mapLock.lock();
instancesInfo = registerStatus.computeIfAbsent(new InstanceId(nacosServiceName, group), id -> new InstancesInfo());
instancesInfo = ConcurrentHashMapUtils.computeIfAbsent(registerStatus, new InstanceId(nacosServiceName, group), id -> new InstancesInfo());
} finally {
mapLock.unlock();
}
@ -223,7 +225,7 @@ public class NacosNamingServiceWrapper {
InstancesInfo instancesInfo;
try {
mapLock.lock();
instancesInfo = registerStatus.computeIfAbsent(new InstanceId(nacosServiceName, group), id -> new InstancesInfo());
instancesInfo = ConcurrentHashMapUtils.computeIfAbsent(registerStatus, new InstanceId(nacosServiceName, group), id -> new InstancesInfo());
} finally {
mapLock.unlock();
}
@ -250,7 +252,7 @@ public class NacosNamingServiceWrapper {
InstancesInfo instancesInfo;
try {
mapLock.lock();
instancesInfo = registerStatus.computeIfAbsent(new InstanceId(nacosServiceName, group), id -> new InstancesInfo());
instancesInfo = ConcurrentHashMapUtils.computeIfAbsent(registerStatus, new InstanceId(nacosServiceName, group), id -> new InstancesInfo());
} finally {
mapLock.unlock();
}

9
dubbo-registry/dubbo-registry-zookeeper/src/main/java/org/apache/dubbo/registry/zookeeper/ZookeeperRegistry.java
View File

@ -21,6 +21,7 @@ import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.ConcurrentHashSet;
import org.apache.dubbo.common.utils.UrlUtils;
import org.apache.dubbo.registry.NotifyListener;
@ -180,9 +181,9 @@ public class ZookeeperRegistry extends CacheableFailbackRegistry {
if (ANY_VALUE.equals(url.getServiceInterface())) {
String root = toRootPath();
boolean check = url.getParameter(CHECK_KEY, false);
ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.computeIfAbsent(url, k -> new ConcurrentHashMap<>());
ConcurrentMap<NotifyListener, ChildListener> listeners = ConcurrentHashMapUtils.computeIfAbsent(zkListeners, url, k -> new ConcurrentHashMap<>());
ChildListener zkListener = listeners.computeIfAbsent(listener, k -> (parentPath, currentChildren) -> {
ChildListener zkListener = ConcurrentHashMapUtils.computeIfAbsent(listeners, listener, k -> (parentPath, currentChildren) -> {
for (String child : currentChildren) {
child = URL.decode(child);
if (!anyServices.contains(child)) {
@ -219,8 +220,8 @@ public class ZookeeperRegistry extends CacheableFailbackRegistry {
/dubbo/[service name]/routers
*/
for (String path : toCategoriesPath(url)) {
ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.computeIfAbsent(url, k -> new ConcurrentHashMap<>());
ChildListener zkListener = listeners.computeIfAbsent(listener, k -> new RegistryChildListenerImpl(url, k, latch));
ConcurrentMap<NotifyListener, ChildListener> listeners = ConcurrentHashMapUtils.computeIfAbsent(zkListeners, url, k -> new ConcurrentHashMap<>());
ChildListener zkListener = ConcurrentHashMapUtils.computeIfAbsent(listeners, listener, k -> new RegistryChildListenerImpl(url, k, latch));
if (zkListener instanceof RegistryChildListenerImpl) {
((RegistryChildListenerImpl) zkListener).setLatch(latch);

3
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/connection/MultiplexProtocolConnectionManager.java
View File

@ -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);
}

5
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/PortUnificationExchanger.java
View File

@ -19,6 +19,7 @@ package org.apache.dubbo.remoting.exchange;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.remoting.ChannelHandler;
import org.apache.dubbo.remoting.RemotingException;
import org.apache.dubbo.remoting.RemotingServer;
@ -38,7 +39,7 @@ public class PortUnificationExchanger {
private static final ConcurrentMap<String, RemotingServer> servers = new ConcurrentHashMap<>();
public static RemotingServer bind(URL url, ChannelHandler handler) {
servers.computeIfAbsent(url.getAddress(), addr -> {
ConcurrentHashMapUtils.computeIfAbsent(servers, url.getAddress(), addr -> {
final AbstractPortUnificationServer server;
try {
server = getTransporter(url).bind(url, handler);
@ -85,7 +86,7 @@ public class PortUnificationExchanger {
public static PortUnificationTransporter getTransporter(URL url) {
return url.getOrDefaultFrameworkModel().getExtensionLoader(PortUnificationTransporter.class)
.getAdaptiveExtension();
.getAdaptiveExtension();
}
}

6
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/transport/CodecSupport.java
View File

@ -25,6 +25,7 @@ import org.apache.dubbo.common.serialize.ObjectInput;
import org.apache.dubbo.common.serialize.ObjectOutput;
import org.apache.dubbo.common.serialize.Serialization;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.remoting.utils.UrlUtils;
import org.apache.dubbo.rpc.model.FrameworkModel;
import org.apache.dubbo.rpc.model.FrameworkServiceRepository;
@ -38,6 +39,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import static org.apache.dubbo.common.BaseServiceMetadata.keyWithoutGroup;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.TRANSPORT_FAILED_SERIALIZATION;
@ -48,7 +50,7 @@ public class CodecSupport {
private static Map<Byte, String> ID_SERIALIZATIONNAME_MAP = new HashMap<Byte, String>();
private static Map<String, Byte> SERIALIZATIONNAME_ID_MAP = new HashMap<String, Byte>();
// Cache null object serialize results, for heartbeat request/response serialize use.
private static Map<Byte, byte[]> ID_NULLBYTES_MAP = new ConcurrentHashMap<>();
private static ConcurrentMap<Byte, byte[]> ID_NULLBYTES_MAP = new ConcurrentHashMap<>();
private static final ThreadLocal<byte[]> TL_BUFFER = ThreadLocal.withInitial(() -> new byte[1024]);
@ -107,7 +109,7 @@ public class CodecSupport {
* @return serialize result of null object
*/
public static byte[] getNullBytesOf(Serialization s) {
return ID_NULLBYTES_MAP.computeIfAbsent(s.getContentTypeId(), k -> {
return ConcurrentHashMapUtils.computeIfAbsent(ID_NULLBYTES_MAP, s.getContentTypeId(), k -> {
//Pre-generated Null object bytes
ByteArrayOutputStream baos = new ByteArrayOutputStream();
byte[] nullBytes = new byte[0];

9
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/zookeeper/AbstractZookeeperClient.java
View File

@ -20,6 +20,7 @@ import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.config.configcenter.ConfigItem;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.ConcurrentHashSet;
import java.util.List;
@ -106,8 +107,8 @@ public abstract class AbstractZookeeperClient<TargetDataListener, TargetChildLis
@Override
public List<String> addChildListener(String path, final ChildListener listener) {
ConcurrentMap<ChildListener, TargetChildListener> listeners = childListeners.computeIfAbsent(path, k -> new ConcurrentHashMap<>());
TargetChildListener targetListener = listeners.computeIfAbsent(listener, k -> createTargetChildListener(path, k));
ConcurrentMap<ChildListener, TargetChildListener> listeners = ConcurrentHashMapUtils.computeIfAbsent(childListeners, path, k -> new ConcurrentHashMap<>());
TargetChildListener targetListener = ConcurrentHashMapUtils.computeIfAbsent(listeners, listener, k -> createTargetChildListener(path, k));
return addTargetChildListener(path, targetListener);
}
@ -118,8 +119,8 @@ public abstract class AbstractZookeeperClient<TargetDataListener, TargetChildLis
@Override
public void addDataListener(String path, DataListener listener, Executor executor) {
ConcurrentMap<DataListener, TargetDataListener> dataListenerMap = listeners.computeIfAbsent(path, k -> new ConcurrentHashMap<>());
TargetDataListener targetListener = dataListenerMap.computeIfAbsent(listener, k -> createTargetDataListener(path, k));
ConcurrentMap<DataListener, TargetDataListener> dataListenerMap = ConcurrentHashMapUtils.computeIfAbsent(listeners, path, k -> new ConcurrentHashMap<>());
TargetDataListener targetListener = ConcurrentHashMapUtils.computeIfAbsent(dataListenerMap, listener, k -> createTargetDataListener(path, k));
addTargetDataListener(path, targetListener, executor);
}

32
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AdaptiveMetrics.java
View File

@ -16,6 +16,7 @@
*/
package org.apache.dubbo.rpc;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.StringUtils;
import java.util.Map;
@ -45,7 +46,7 @@ public class AdaptiveMetrics {
private final AtomicLong errorReq = new AtomicLong();
private double ewma = 0;
public double getLoad(String idKey,int weight,int timeout){
public double getLoad(String idKey, int weight, int timeout) {
AdaptiveMetrics metrics = getStatus(idKey);
//If the time more than 2 times, mandatory selected
@ -53,13 +54,13 @@ public class AdaptiveMetrics {
return 0;
}
if (metrics.currentTime > 0){
if (metrics.currentTime > 0) {
long multiple = (System.currentTimeMillis() - metrics.currentTime) / timeout + 1;
if (multiple > 0) {
if (metrics.currentProviderTime == metrics.currentTime) {
//penalty value
metrics.lastLatency = timeout * 2L;
}else {
} else {
metrics.lastLatency = metrics.lastLatency >> multiple;
}
metrics.ewma = metrics.beta * metrics.ewma + (1 - metrics.beta) * metrics.lastLatency;
@ -68,49 +69,48 @@ public class AdaptiveMetrics {
}
long inflight = metrics.consumerReq.get() - metrics.consumerSuccess.get() - metrics.errorReq.get();
return metrics.providerCPULoad * (Math.sqrt(metrics.ewma) + 1) * (inflight + 1) / ((((double)metrics.consumerSuccess.get() / (double)(metrics.consumerReq.get() + 1)) * weight) + 1);
return metrics.providerCPULoad * (Math.sqrt(metrics.ewma) + 1) * (inflight + 1) / ((((double) metrics.consumerSuccess.get() / (double) (metrics.consumerReq.get() + 1)) * weight) + 1);
}
public AdaptiveMetrics getStatus(String idKey){
return metricsStatistics.computeIfAbsent(idKey, k -> new AdaptiveMetrics());
public AdaptiveMetrics getStatus(String idKey) {
return ConcurrentHashMapUtils.computeIfAbsent(metricsStatistics, idKey, k -> new AdaptiveMetrics());
}
public void addConsumerReq(String idKey){
public void addConsumerReq(String idKey) {
AdaptiveMetrics metrics = getStatus(idKey);
metrics.consumerReq.incrementAndGet();
}
public void addConsumerSuccess(String idKey){
public void addConsumerSuccess(String idKey) {
AdaptiveMetrics metrics = getStatus(idKey);
metrics.consumerSuccess.incrementAndGet();
}
public void addErrorReq(String idKey){
public void addErrorReq(String idKey) {
AdaptiveMetrics metrics = getStatus(idKey);
metrics.errorReq.incrementAndGet();
}
public void setPickTime(String idKey,long time){
public void setPickTime(String idKey, long time) {
AdaptiveMetrics metrics = getStatus(idKey);
metrics.pickTime = time;
}
public void setProviderMetrics(String idKey,Map<String,String> metricsMap){
public void setProviderMetrics(String idKey, Map<String, String> metricsMap) {
AdaptiveMetrics metrics = getStatus(idKey);
long serviceTime = Long.parseLong(Optional.ofNullable(metricsMap.get("curTime")).filter(v -> StringUtils.isNumeric(v,false)).orElse("0"));
long serviceTime = Long.parseLong(Optional.ofNullable(metricsMap.get("curTime")).filter(v -> StringUtils.isNumeric(v, false)).orElse("0"));
//If server time is less than the current time, discard
if (metrics.currentProviderTime > serviceTime){
if (metrics.currentProviderTime > serviceTime) {
return;
}
metrics.currentProviderTime = serviceTime;
metrics.currentTime = serviceTime;
metrics.providerCPULoad = Double.parseDouble(Optional.ofNullable(metricsMap.get("load")).filter(v -> StringUtils.isNumeric(v,true)).orElse("0"));
metrics.lastLatency = Long.parseLong((Optional.ofNullable(metricsMap.get("rt")).filter(v -> StringUtils.isNumeric(v,false)).orElse("0")));
metrics.providerCPULoad = Double.parseDouble(Optional.ofNullable(metricsMap.get("load")).filter(v -> StringUtils.isNumeric(v, true)).orElse("0"));
metrics.lastLatency = Long.parseLong((Optional.ofNullable(metricsMap.get("rt")).filter(v -> StringUtils.isNumeric(v, false)).orElse("0")));
metrics.beta = 0.5;
//Vt = β * Vt-1 + (1 - β ) * θt

11
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcStatus.java
View File

@ -17,6 +17,7 @@
package org.apache.dubbo.rpc;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@ -33,10 +34,10 @@ import java.util.concurrent.atomic.AtomicLong;
public class RpcStatus {
private static final ConcurrentMap<String, RpcStatus> SERVICE_STATISTICS = new ConcurrentHashMap<String,
RpcStatus>();
RpcStatus>();
private static final ConcurrentMap<String, ConcurrentMap<String, RpcStatus>> METHOD_STATISTICS =
new ConcurrentHashMap<String, ConcurrentMap<String, RpcStatus>>();
new ConcurrentHashMap<String, ConcurrentMap<String, RpcStatus>>();
private final ConcurrentMap<String, Object> values = new ConcurrentHashMap<String, Object>();
@ -58,7 +59,7 @@ public class RpcStatus {
*/
public static RpcStatus getStatus(URL url) {
String uri = url.toIdentityString();
return SERVICE_STATISTICS.computeIfAbsent(uri, key -> new RpcStatus());
return ConcurrentHashMapUtils.computeIfAbsent(SERVICE_STATISTICS, uri, key -> new RpcStatus());
}
/**
@ -76,8 +77,8 @@ public class RpcStatus {
*/
public static RpcStatus getStatus(URL url, String methodName) {
String uri = url.toIdentityString();
ConcurrentMap<String, RpcStatus> map = METHOD_STATISTICS.computeIfAbsent(uri, k -> new ConcurrentHashMap<>());
return map.computeIfAbsent(methodName, k -> new RpcStatus());
ConcurrentMap<String, RpcStatus> map = ConcurrentHashMapUtils.computeIfAbsent(METHOD_STATISTICS, uri, k -> new ConcurrentHashMap<>());
return ConcurrentHashMapUtils.computeIfAbsent(map, methodName, k -> new RpcStatus());
}
/**

8
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/AccessLogFilter.java
View File

@ -20,6 +20,7 @@ import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.threadpool.manager.FrameworkExecutorRepository;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.ConfigUtils;
import org.apache.dubbo.rpc.Constants;
import org.apache.dubbo.rpc.Filter;
@ -40,6 +41,7 @@ import java.util.Optional;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
@ -77,7 +79,7 @@ public class AccessLogFilter implements Filter {
// It's safe to declare it as singleton since it runs on single thread only
private final DateFormat fileNameFormatter = new SimpleDateFormat(FILE_DATE_FORMAT);
private final Map<String, Queue<AccessLogData>> logEntries = new ConcurrentHashMap<>();
private final ConcurrentMap<String, Queue<AccessLogData>> logEntries = new ConcurrentHashMap<>();
private AtomicBoolean scheduled = new AtomicBoolean();
@ -127,12 +129,12 @@ public class AccessLogFilter implements Filter {
}
private void log(String accessLog, AccessLogData accessLogData) {
Queue<AccessLogData> logQueue = logEntries.computeIfAbsent(accessLog, k -> new ConcurrentLinkedQueue<>());
Queue<AccessLogData> logQueue = ConcurrentHashMapUtils.computeIfAbsent(logEntries, accessLog, k -> new ConcurrentLinkedQueue<>());
if (logQueue.size() < LOG_MAX_BUFFER) {
logQueue.add(accessLogData);
} else {
logger.warn(CONFIG_FILTER_VALIDATION_EXCEPTION, "", "","AccessLog buffer is full. Do a force writing to file to clear buffer.");
logger.warn(CONFIG_FILTER_VALIDATION_EXCEPTION, "", "", "AccessLog buffer is full. Do a force writing to file to clear buffer.");
//just write current logSet to file.
writeLogSetToFile(accessLog, logQueue);
//after force writing, add accessLogData to current logSet

3
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/AbstractProtocol.java
View File

@ -41,6 +41,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_SERVER_SHUTDOWN_TIMEOUT;
import static org.apache.dubbo.common.constants.CommonConstants.SHUTDOWN_WAIT_KEY;
@ -59,7 +60,7 @@ public abstract class AbstractProtocol implements Protocol, ScopeModelAware {
/**
* <host:port, ProtocolServer>
*/
protected final Map<String, ProtocolServer> serverMap = new ConcurrentHashMap<>();
protected final ConcurrentMap<String, ProtocolServer> serverMap = new ConcurrentHashMap<>();
// TODO SoftReference
protected final Set<Invoker<?>> invokers = new ConcurrentHashSet<>();

21
dubbo-rpc/dubbo-rpc-grpc/src/main/java/org/apache/dubbo/rpc/protocol/grpc/GrpcProtocol.java
View File

@ -19,6 +19,7 @@ package org.apache.dubbo.rpc.protocol.grpc;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.ProtocolServer;
import org.apache.dubbo.rpc.RpcException;
@ -55,13 +56,13 @@ public class GrpcProtocol extends AbstractProxyProtocol {
@Override
protected <T> Runnable doExport(T proxiedImpl, Class<T> type, URL url) throws RpcException {
String key = url.getAddress();
ProtocolServer protocolServer = serverMap.computeIfAbsent(key, k -> {
ProtocolServer protocolServer = ConcurrentHashMapUtils.computeIfAbsent(serverMap, key, k -> {
DubboHandlerRegistry registry = new DubboHandlerRegistry();
NettyServerBuilder builder =
NettyServerBuilder
NettyServerBuilder
.forPort(url.getPort())
.fallbackHandlerRegistry(registry);
.fallbackHandlerRegistry(registry);
Server originalServer = GrpcOptionsUtils.buildServerBuilder(url, builder).build();
GrpcRemotingServer remotingServer = new GrpcRemotingServer(originalServer, registry);
@ -74,7 +75,7 @@ public class GrpcProtocol extends AbstractProxyProtocol {
ProviderModel providerModel = serviceRepository.lookupExportedService(url.getServiceKey());
if (providerModel == null) {
throw new IllegalStateException("Service " + url.getServiceKey() + "should have already been stored in service repository, " +
"but failed to find it.");
"but failed to find it.");
}
Object originalImpl = providerModel.getServiceInstance();
@ -84,7 +85,7 @@ public class GrpcProtocol extends AbstractProxyProtocol {
method.invoke(originalImpl, proxiedImpl);
} catch (Exception e) {
throw new IllegalStateException("Failed to set dubbo proxied service impl to stub, please make sure your stub " +
"was generated by the dubbo-protoc-compiler.", e);
"was generated by the dubbo-protoc-compiler.", e);
}
grpcServer.getRegistry().addService((BindableService) originalImpl, url.getServiceKey());
@ -101,13 +102,13 @@ public class GrpcProtocol extends AbstractProxyProtocol {
if (enclosingClass == null) {
throw new IllegalArgumentException(type.getName() + " must be declared inside protobuf generated classes, " +
"should be something like ServiceNameGrpc.IServiceName.");
"should be something like ServiceNameGrpc.IServiceName.");
}
final Method dubboStubMethod;
try {
dubboStubMethod = enclosingClass.getDeclaredMethod("getDubboStub", Channel.class, CallOptions.class,
URL.class);
URL.class);
} catch (NoSuchMethodException e) {
throw new IllegalArgumentException("Does not find getDubboStub in " + enclosingClass.getName() + ", please use the customized protoc-gen-dubbo-java to update the generated classes.");
}
@ -118,9 +119,9 @@ public class GrpcProtocol extends AbstractProxyProtocol {
// CallOptions
try {
@SuppressWarnings("unchecked") final T stub = (T) dubboStubMethod.invoke(null,
channel,
GrpcOptionsUtils.buildCallOptions(url),
url
channel,
GrpcOptionsUtils.buildCallOptions(url),
url
);
final Invoker<T> target = proxyFactory.getInvoker(stub, type, url);
GrpcInvoker<T> grpcInvoker = new GrpcInvoker<>(type, url, target, channel);

8
dubbo-rpc/dubbo-rpc-rest/src/main/java/org/apache/dubbo/rpc/protocol/rest/RestProtocol.java
View File

@ -17,6 +17,7 @@
package org.apache.dubbo.rpc.protocol.rest;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.remoting.http.HttpBinder;
import org.apache.dubbo.remoting.http.servlet.BootstrapListener;
@ -45,6 +46,7 @@ import javax.ws.rs.ProcessingException;
import javax.ws.rs.WebApplicationException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import static org.apache.dubbo.common.constants.CommonConstants.COMMA_SPLIT_PATTERN;
@ -72,7 +74,7 @@ public class RestProtocol extends AbstractProxyProtocol {
private final RestServerFactory serverFactory = new RestServerFactory();
private final Map<String, ReferenceCountedClient> clients = new ConcurrentHashMap<>();
private final ConcurrentMap<String, ReferenceCountedClient> clients = new ConcurrentHashMap<>();
private volatile ConnectionMonitor connectionMonitor;
@ -93,7 +95,7 @@ public class RestProtocol extends AbstractProxyProtocol {
protected <T> Runnable doExport(T impl, Class<T> type, URL url) throws RpcException {
String addr = getAddr(url);
Class implClass = url.getServiceModel().getProxyObject().getClass();
RestProtocolServer server = (RestProtocolServer) serverMap.computeIfAbsent(addr, restServer -> {
RestProtocolServer server = (RestProtocolServer) ConcurrentHashMapUtils.computeIfAbsent(serverMap, addr, restServer -> {
RestProtocolServer s = serverFactory.createServer(url.getParameter(SERVER_KEY, DEFAULT_SERVER));
s.setAddress(url.getAddress());
s.start(url);
@ -135,7 +137,7 @@ public class RestProtocol extends AbstractProxyProtocol {
@Override
protected <T> T doRefer(Class<T> serviceType, URL url) throws RpcException {
ReferenceCountedClient referenceCountedClient = clients.computeIfAbsent(url.getAddress(), _key -> {
ReferenceCountedClient referenceCountedClient = ConcurrentHashMapUtils.computeIfAbsent(clients, url.getAddress(), _key -> {
// TODO more configs to add
return createReferenceCountedClient(url);
});

4
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/SingleProtobufUtils.java
View File

@ -17,6 +17,8 @@
package org.apache.dubbo.rpc.protocol.tri;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import com.google.protobuf.BoolValue;
import com.google.protobuf.BytesValue;
import com.google.protobuf.DoubleValue;
@ -110,7 +112,7 @@ public class SingleProtobufUtils {
}
private static SingleMessageMarshaller<?> getMarshaller(Class<?> clz) {
return MARSHALLER_CACHE.computeIfAbsent(clz, k -> new SingleMessageMarshaller(k));
return ConcurrentHashMapUtils.computeIfAbsent(MARSHALLER_CACHE, clz, k -> new SingleMessageMarshaller(k));
}
public static final class SingleMessageMarshaller<T extends MessageLite> {

3
dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/AbstractProtocol.java
View File

@ -43,6 +43,7 @@ import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.INTERNAL_INTERRUPTED;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.PROTOCOL_FAILED_REQUEST;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
public abstract class AbstractProtocol<T, S extends DeltaResource<T>> implements XdsProtocol<T>, XdsListener {
@ -138,7 +139,7 @@ public abstract class AbstractProtocol<T, S extends DeltaResource<T>> implements
Consumer<Map<String, T>> futureConsumer = future::complete;
try {
writeLock.lock();
consumerObserveMap.computeIfAbsent(consumerObserveResourceNames, key -> new ArrayList<>())
ConcurrentHashMapUtils.computeIfAbsent((ConcurrentHashMap<Set<String>, List<Consumer<Map<String, T>>>>)consumerObserveMap,consumerObserveResourceNames, key -> new ArrayList<>())
.add(futureConsumer);
} finally {
writeLock.unlock();

5
dubbo-xds/src/main/java/org/apache/dubbo/rpc/cluster/router/xds/EdsEndpointManager.java
View File

@ -25,6 +25,7 @@ import java.util.stream.Collectors;
import org.apache.dubbo.common.threadpool.manager.FrameworkExecutorRepository;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.ConcurrentHashSet;
import org.apache.dubbo.registry.xds.util.PilotExchanger;
import org.apache.dubbo.registry.xds.util.protocol.message.Endpoint;
@ -45,7 +46,7 @@ public class EdsEndpointManager {
public synchronized void subscribeEds(String cluster, EdsEndpointListener listener) {
Set<EdsEndpointListener> listeners = ENDPOINT_LISTENERS.computeIfAbsent(cluster, key ->
Set<EdsEndpointListener> listeners = ConcurrentHashMapUtils.computeIfAbsent(ENDPOINT_LISTENERS, cluster, key ->
new ConcurrentHashSet<>()
);
if (CollectionUtils.isEmpty(listeners)) {
@ -59,7 +60,7 @@ public class EdsEndpointManager {
}
private void doSubscribeEds(String cluster) {
EDS_LISTENERS.computeIfAbsent(cluster, key -> endpoints -> {
ConcurrentHashMapUtils.computeIfAbsent(EDS_LISTENERS, cluster, key -> endpoints -> {
Set<Endpoint> result = endpoints.values()
.stream()
.map(EndpointResult::getEndpoints)

8
dubbo-xds/src/main/java/org/apache/dubbo/rpc/cluster/router/xds/RdsRouteRuleManager.java
View File

@ -20,9 +20,11 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.function.Consumer;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.ConcurrentHashSet;
import org.apache.dubbo.registry.xds.util.PilotExchanger;
import org.apache.dubbo.registry.xds.util.protocol.message.ListenerResult;
@ -38,7 +40,7 @@ public class RdsRouteRuleManager {
private static final ConcurrentHashMap<String, List<XdsRouteRule>> ROUTE_DATA_CACHE = new ConcurrentHashMap<>();
private static final Map<String, RdsVirtualHostListener> RDS_LISTENERS = new ConcurrentHashMap<>();
private static final ConcurrentMap<String, RdsVirtualHostListener> RDS_LISTENERS = new ConcurrentHashMap<>();
private static volatile Consumer<Map<String, ListenerResult>> LDS_LISTENER;
@ -51,7 +53,7 @@ public class RdsRouteRuleManager {
public synchronized void subscribeRds(String domain, XdsRouteRuleListener listener) {
Set<XdsRouteRuleListener> listeners = RULE_LISTENERS.computeIfAbsent(domain, key ->
Set<XdsRouteRuleListener> listeners = ConcurrentHashMapUtils.computeIfAbsent(RULE_LISTENERS, domain, key ->
new ConcurrentHashSet<>()
);
if (CollectionUtils.isEmpty(listeners)) {
@ -107,7 +109,7 @@ public class RdsRouteRuleManager {
}
}
}
RDS_LISTENERS.computeIfAbsent(domain, key -> new RdsVirtualHostListener(domain, this));
ConcurrentHashMapUtils.computeIfAbsent(RDS_LISTENERS, domain, key -> new RdsVirtualHostListener(domain, this));
RDS_LISTENER.accept(RDS_RESULT);
}