From 94142aa80fccf24c009caf0cca80ff340cecb4f9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EA=B9=80=EB=8F=99=EA=B7=BC?= Date: Wed, 2 Oct 2024 14:01:23 +0900 Subject: [PATCH] change synchronized to reentrant lock in streambridge --- .../cloud/stream/function/StreamBridge.java | 18 ++++++++++++++---- 1 file changed, 14 insertions(+), 4 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 e734cec5d..60ddb59df 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 @@ -188,7 +188,14 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi ProducerProperties producerProperties = this.bindingServiceProperties.getProducerProperties(bindingName); MessageChannel messageChannel = this.resolveDestination(bindingName, producerProperties, binderName); - Function functionToInvoke = this.getStreamBridgeFunction(outputContentType.toString(), producerProperties); + Function functionToInvoke; + lock.lock(); + try { + functionToInvoke = this.getStreamBridgeFunction(outputContentType.toString(), producerProperties); + } finally { + lock.unlock(); + } + if (producerProperties != null && producerProperties.isPartitioned()) { functionToInvoke = new PartitionAwareFunctionWrapper(functionToInvoke, this.applicationContext, producerProperties); @@ -202,8 +209,11 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi : new GenericMessage<>(data, Collections.singletonMap(MessageUtils.TARGET_PROTOCOL, targetType)); Message resultMessage; - synchronized (this) { - resultMessage = (Message) functionToInvoke.apply(messageToSend); + lock.lock(); + try { + resultMessage = (Message)functionToInvoke.apply(messageToSend); + } finally { + lock.unlock(); } if (resultMessage == null @@ -225,7 +235,7 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi return hash; } - private synchronized FunctionInvocationWrapper getStreamBridgeFunction(String outputContentType, ProducerProperties producerProperties) { + private FunctionInvocationWrapper getStreamBridgeFunction(String outputContentType, ProducerProperties producerProperties) { int streamBridgeFunctionKey = this.hashProducerProperties(producerProperties, outputContentType); if (this.streamBridgeFunctionCache.containsKey(streamBridgeFunctionKey)) { return this.streamBridgeFunctionCache.get(streamBridgeFunctionKey);