Replaced ReentrantLock with ConcurrentHashMap.computeIfAbsent()

2tsumo-hitori · olegz · commit bb3324a5b599 · 2025-03-24T23:41:07.000+01:00
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 &lt;audghks1996@naver.com&gt;
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
@@ -18,7 +18,7 @@
 
 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 @@ protected boolean removeEldestEntry(Map.Entry<String, MessageChannel> eldest) {
 			}
 		};
 		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 boolean send(String bindingName, @Nullable String binderName, Object data
 		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 @@ private int hashProducerProperties(ProducerProperties producerProperties, String
 
 	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
