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

change synchronized to reentrant lock in streambridge

Browse Source
This commit is contained in:
김동근
2024-10-02 14:01:23 +09:00
committed by Oleg Zhurakousky
parent 56d86c9807
commit 94142aa80f
1 changed files with 14 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<byte[]>) functionToInvoke.apply(messageToSend);
lock.lock();
try {
resultMessage = (Message<byte[]>)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);