SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Replaced ReentrantLock with ConcurrentHashMap.computeIfAbsent()

Browse Source 
to improve performance in StreamBridge.send() method.

- Avoids unnecessary locking overhead
- Improves concurrency and reduces contention
- Enhances throughput for high-load scenarios

Signed-off-by: 2tsumo-hitori <audghks1996@naver.com>
This commit is contained in:
2tsumo-hitori
2025-03-16 21:30:43 +09:00
committed by Oleg Zhurakousky
parent c733627d8c
commit bb3324a5b5
1 changed files with 6 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java
View File

@@ -18,7 +18,7 @@ package org.springframework.cloud.stream.function;
import java.lang.reflect.Type;
import java.util.Collections;
import java.util.HashMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
@@ -163,7 +163,7 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi
}
};
this.functionInvocationHelper = applicationContext.getBean(FunctionInvocationHelper.class);
this.streamBridgeFunctionCache = new HashMap<>();
this.streamBridgeFunctionCache = new ConcurrentHashMap<>();
observationRegistries.ifAvailable(registry -> this.observationRegistry = registry);
}
@@ -198,15 +198,7 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi
ProducerProperties producerProperties = this.bindingServiceProperties.getProducerProperties(bindingName);
MessageChannel messageChannel = this.resolveDestination(bindingName, producerProperties, binderName);
Function functionToInvoke;
lock.lock();
try {
functionToInvoke = this.getStreamBridgeFunction(outputContentType.toString(), producerProperties);
}
finally {
lock.unlock();
}
Function functionToInvoke = this.getStreamBridgeFunction(outputContentType.toString(), producerProperties);
if (producerProperties != null && producerProperties.isPartitioned()) {
functionToInvoke = new PartitionAwareFunctionWrapper(functionToInvoke, this.applicationContext, producerProperties);
@@ -253,15 +245,12 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi
private FunctionInvocationWrapper getStreamBridgeFunction(String outputContentType, ProducerProperties producerProperties) {
int streamBridgeFunctionKey = this.hashProducerProperties(producerProperties, outputContentType);
if (this.streamBridgeFunctionCache.containsKey(streamBridgeFunctionKey)) {
return this.streamBridgeFunctionCache.get(streamBridgeFunctionKey);
}
else {
return this.streamBridgeFunctionCache.computeIfAbsent(streamBridgeFunctionKey, key -> {
FunctionInvocationWrapper functionToInvoke = this.functionCatalog.lookup(STREAM_BRIDGE_FUNC_NAME, outputContentType.toString());
this.streamBridgeFunctionCache.put(streamBridgeFunctionKey, functionToInvoke);
functionToInvoke.setSkipOutputConversion(producerProperties.isUseNativeEncoding());
return functionToInvoke;
}
});
}
@Override