From b3988f52aad97fcbb8f308b734b4a11b1a8a80b3 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Mon, 11 May 2020 14:48:33 +0200 Subject: [PATCH] GH-1958 Improve caching of dynamic destinations Added spring.cloud.stream.dynamic-destination-cache-size property Resolves #1958 --- .../config/BindingServiceProperties.java | 15 ++++ .../cloud/stream/function/StreamBridge.java | 88 ++++++------------- 2 files changed, 42 insertions(+), 61 deletions(-) diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java index bea6c8264..fcebda21f 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java @@ -124,6 +124,13 @@ public class BindingServiceProperties */ private String[] dynamicDestinations = new String[0]; + /** + * The maximum size of Least Recently Used (LRU) cache of dynamic destinations. Once + * this size is reached, new destinations will trigger the removal of old destinations. + * Default: 10 + */ + private int dynamicDestinationCacheSize = 10; + /** * Retry interval (in seconds) used to schedule binding attempts. Default: 30 sec. */ @@ -310,6 +317,14 @@ public class BindingServiceProperties } } + public int getDynamicDestinationCacheSize() { + return dynamicDestinationCacheSize; + } + + public void setDynamicDestinationCacheSize(int dynamicDestinationCacheSize) { + this.dynamicDestinationCacheSize = dynamicDestinationCacheSize; + } + /* * The "necessary" implies the scenario where only defaults are defined. */ diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java index a954e5019..9e0989c50 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java @@ -16,7 +16,6 @@ package org.springframework.cloud.stream.function; -import java.util.HashMap; import java.util.LinkedHashMap; import java.util.Map; import java.util.Map.Entry; @@ -34,16 +33,12 @@ import org.springframework.cloud.function.context.FunctionType; import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper; import org.springframework.cloud.stream.binder.Binding; import org.springframework.cloud.stream.binder.ProducerProperties; -import org.springframework.cloud.stream.binding.BinderAwareChannelResolver; import org.springframework.cloud.stream.binding.BindingService; -import org.springframework.cloud.stream.config.BindingProperties; import org.springframework.cloud.stream.config.BindingServiceProperties; import org.springframework.cloud.stream.messaging.DirectWithAttributesChannel; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.messaging.Message; -import org.springframework.messaging.MessageChannel; import org.springframework.messaging.SubscribableChannel; -import org.springframework.messaging.support.GenericMessage; import org.springframework.util.MimeType; import org.springframework.util.MimeTypeUtils; @@ -67,7 +62,7 @@ public final class StreamBridge implements SmartInitializingSingleton { protected final Log logger = LogFactory.getLog(getClass()); - private final Map outputChannelsOnly = new HashMap<>(); + private final Map channelCache; private final FunctionCatalog functionCatalog; @@ -79,9 +74,6 @@ public final class StreamBridge implements SmartInitializingSingleton { private boolean initialized; - @Autowired - private BinderAwareChannelResolver dynamicDestinationResolver; - @Autowired private BindingService bindingService; @@ -92,12 +84,23 @@ public final class StreamBridge implements SmartInitializingSingleton { * @param bindingServiceProperties instance of {@link BindingServiceProperties} * @param applicationContext instance of {@link ConfigurableApplicationContext} */ + @SuppressWarnings("serial") StreamBridge(FunctionCatalog functionCatalog, FunctionRegistry functionRegistry, BindingServiceProperties bindingServiceProperties, ConfigurableApplicationContext applicationContext) { this.functionCatalog = functionCatalog; this.functionRegistry = functionRegistry; this.applicationContext = applicationContext; this.bindingServiceProperties = bindingServiceProperties; + this.channelCache = new LinkedHashMap() { + @Override + protected boolean removeEldestEntry(Map.Entry eldest) { + boolean remove = size() > bindingServiceProperties.getDynamicDestinationCacheSize(); + if (remove && logger.isDebugEnabled()) { + logger.debug("Removing message channel from cache " + eldest.getKey()); + } + return remove; + } + }; } /** @@ -111,16 +114,12 @@ public final class StreamBridge implements SmartInitializingSingleton { return this.send(bindingName, data, MimeTypeUtils.APPLICATION_JSON); } - public boolean sendLight(String bindingName, Object data) { - return this.sendLight(bindingName, data, MimeTypeUtils.APPLICATION_JSON); - } - /** * Sends 'data' to an output binding specified by 'bindingName' argument while * using the content type specified by the 'outputContentType' argument to deal * with output type conversion (if necessary). * For typical cases `bindingName` is configured using 'spring.cloud.stream.source' property. - * However, this operation also supports sending to dynamic destinations. This means if the name + * However, this operation also supports sending to truly dynamic destinations. This means if the name * provided via 'bindingName' does not have a corresponding binding such name will be * treated as dynamic destination. * @@ -129,59 +128,26 @@ public final class StreamBridge implements SmartInitializingSingleton { * @param outputContentType content type to be used to deal with output type conversion * @return true if data was sent successfully, otherwise false or throws an exception. */ - @SuppressWarnings("unchecked") + @SuppressWarnings({ "unchecked", "unused" }) public boolean send(String bindingName, Object data, MimeType outputContentType) { - if (!this.outputChannelsOnly.containsKey(bindingName)) { - logger.info("Binding name '" + bindingName + "' does not exist. This means that value '" - + bindingName + "' will be treated as dynamic destination. If this is not your intention please " - + "provide 'spring.cloud.stream.source' property"); - this.outputChannelsOnly.put(bindingName, dynamicDestinationResolver.resolveDestination(bindingName)); + SubscribableChannel messageChannel = this.channelCache.get(bindingName); + if (messageChannel == null) { FunctionRegistration> fr = new FunctionRegistration<>(v -> v, bindingName); this.functionRegistry.register(fr.type(FunctionType.from(Object.class).to(Object.class).message())); - } - - FunctionInvocationWrapper functionWrapper = this.functionCatalog.lookup(bindingName, outputContentType.toString()); - - BindingProperties bindingProperties = this.bindingServiceProperties.getBindings().get(bindingName); - ProducerProperties producerProperties = bindingProperties.getProducer(); - Function functionToInvoke = functionWrapper; - if (producerProperties != null && producerProperties.isPartitioned()) { - functionToInvoke = new PartitionAwareFunctionWrapper(functionWrapper, this.applicationContext, producerProperties); - } - - Message resultMessage = (Message) functionToInvoke.apply(data); - this.outputChannelsOnly.get(bindingName).send(resultMessage); - return true; - } - - private Map channelCache = new LinkedHashMap() { - private static final int MAX_ENTRIES = 2; - - @Override - protected boolean removeEldestEntry(Map.Entry eldest) { - boolean remove = size() > MAX_ENTRIES; - if (remove) { - System.err.println("==> REMOVING ENTRY: " + eldest); - } - return remove; - } - }; - - public boolean sendLight(String bindingName, Object data, MimeType outputContentType) { - SubscribableChannel messageChannel = null; - if (channelCache.containsKey(bindingName)) { - messageChannel = channelCache.get(bindingName); - } - else { - ProducerProperties producerProperties = this.bindingServiceProperties.getProducerProperties(bindingName); - producerProperties.setRequiredGroups(bindingName); messageChannel = new DirectWithAttributesChannel(); Binding binding = this.bindingService.bindProducer(messageChannel, bindingName, false); -// binding.start(); // looks like this method is never called anyway. - channelCache.put(bindingName, messageChannel); + this.channelCache.put(bindingName, messageChannel); } - return messageChannel.send(new GenericMessage<>(data)); + ProducerProperties producerProperties = this.bindingServiceProperties.getProducerProperties(bindingName); + //producerProperties.setRequiredGroups(bindingName); + Function functionToInvoke = this.functionCatalog.lookup(bindingName, outputContentType.toString()); + if (producerProperties != null && producerProperties.isPartitioned()) { + functionToInvoke = new PartitionAwareFunctionWrapper((FunctionInvocationWrapper) functionToInvoke, this.applicationContext, producerProperties); + } + // this function is a pass through and is only required to force output conversion if necessary on SCF side. + Message resultMessage = (Message) functionToInvoke.apply(data); + return messageChannel.send(resultMessage); } @Override @@ -192,7 +158,7 @@ public final class StreamBridge implements SmartInitializingSingleton { Map channels = applicationContext.getBeansOfType(DirectWithAttributesChannel.class); for (Entry channelEntry : channels.entrySet()) { if (channelEntry.getValue().getAttribute("type").equals("output")) { - outputChannelsOnly.put(channelEntry.getKey(), channelEntry.getValue()); + this.channelCache.put(channelEntry.getKey(), channelEntry.getValue()); // we're registering a dummy pass-through function to ensure that it goes through the // same process (type conversion, etc) as other function invocation. FunctionRegistration> fr = new FunctionRegistration<>(v -> v, channelEntry.getKey());