From bb3324a5b599befe82bed4b883326d17b2884e3d Mon Sep 17 00:00:00 2001 From: 2tsumo-hitori Date: Sun, 16 Mar 2025 21:30:43 +0900 Subject: [PATCH] Replaced ReentrantLock with ConcurrentHashMap.computeIfAbsent() 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 --- .../cloud/stream/function/StreamBridge.java | 23 +++++-------------- 1 file changed, 6 insertions(+), 17 deletions(-) diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java index 688fe4901..9e2e1229e 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java @@ -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