Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-4137: Improve Aggregator Performance

Browse Source 
JIRA: https://jira.spring.io/browse/INT-4137

Aggregators with `SequenceSizeReleaseStrategy` do not perform well with large groups
because of an O(n) linear search to reject (discard) duplicate sequences.

Change the aggregator to use a `SimpleSequenceSizeReleaseStrategy` by default, unless
`setReleasePartialSequences(true)`.

This avoids the linear search, which is not needed for most splitter -> ... -> aggregator
flows.

Log a warning if SSRS is used.

Polishing - PR Comments

* Remove `this.logger.isWarnEnabled()` since it is redundant when our log message is just string constant
This commit is contained in:
Gary Russell
2016-10-10 16:14:31 -04:00
committed by Artem Bilan
parent 8d94bd5e3e
commit 890ad63584
10 changed files with 129 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files

109
spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java
View File

@@ -46,7 +46,6 @@ import org.springframework.integration.handler.AbstractMessageProducingHandler;
import org.springframework.integration.handler.DiscardingMessageHandler;
import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.store.MessageGroupStore;
import org.springframework.integration.store.MessageGroupStore.MessageGroupCallback;
import org.springframework.integration.store.MessageStore;
import org.springframework.integration.store.SimpleMessageGroup;
import org.springframework.integration.store.SimpleMessageStore;
@@ -88,7 +87,7 @@ import org.springframework.util.CollectionUtils;
public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageProducingHandler
implements DiscardingMessageHandler, DisposableBean, ApplicationEventPublisherAware {
private static final Log logger = LogFactory.getLog(AbstractCorrelatingMessageHandler.class);
protected final Log logger = LogFactory.getLog(getClass());
private final Comparator<Message<?>> sequenceNumberComparator = new SequenceNumberComparator();
@@ -102,6 +101,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
private volatile ReleaseStrategy releaseStrategy;
private volatile boolean releaseStrategySet;
private volatile MessageChannel discardChannel;
private volatile String discardChannelName;
@@ -140,7 +141,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
this.correlationStrategy = (correlationStrategy == null
? new HeaderAttributeCorrelationStrategy(IntegrationMessageHeaderAccessor.CORRELATION_ID)
: correlationStrategy);
this.releaseStrategy = releaseStrategy == null ? new SequenceSizeReleaseStrategy() : releaseStrategy;
this.releaseStrategy = releaseStrategy == null ? new SimpleSequenceSizeReleaseStrategy() : releaseStrategy;
this.sequenceAware = this.releaseStrategy instanceof SequenceSizeReleaseStrategy;
}
@@ -161,12 +162,9 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
public final void setMessageStore(MessageGroupStore store) {
this.messageStore = store;
store.registerMessageGroupExpiryCallback(new MessageGroupCallback() {
@Override
public void execute(MessageGroupStore messageGroupStore, MessageGroup group) {
AbstractCorrelatingMessageHandler.this.forceReleaseProcessor.processMessageGroup(group);
}
});
store.registerMessageGroupExpiryCallback(
(messageGroupStore, group) ->
AbstractCorrelatingMessageHandler.this.forceReleaseProcessor.processMessageGroup(group));
}
public void setCorrelationStrategy(CorrelationStrategy correlationStrategy) {
@@ -178,6 +176,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
Assert.notNull(releaseStrategy);
this.releaseStrategy = releaseStrategy;
this.sequenceAware = this.releaseStrategy instanceof SequenceSizeReleaseStrategy;
this.releaseStrategySet = true;
}
public void setGroupTimeoutExpression(Expression groupTimeoutExpression) {
@@ -228,14 +227,20 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
if (this.releasePartialSequences) {
Assert.isInstanceOf(SequenceSizeReleaseStrategy.class, this.releaseStrategy,
"Release strategy of type [" + this.releaseStrategy.getClass().getSimpleName() +
"] cannot release partial sequences. Use the default SequenceSizeReleaseStrategy instead.");
((SequenceSizeReleaseStrategy) this.releaseStrategy).setReleasePartialSequences(this.releasePartialSequences);
"] cannot release partial sequences. Use a SequenceSizeReleaseStrategy instead.");
((SequenceSizeReleaseStrategy) this.releaseStrategy)
.setReleasePartialSequences(this.releasePartialSequences);
}
if (this.evaluationContext == null) {
this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory());
}
if (this.sequenceAware) {
this.logger.warn("Using a SequenceSizeReleaseStrategy with large groups may not perform well, consider "
+ "using a SimpleSequenceSizeReleaseStrategy");
}
/*
* Disallow any further changes to the lock registry
* (checked in the setter).
@@ -287,11 +292,14 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
}
/**
* Set {@code releasePartialSequences} on an underlying
* {@link SequenceSizeReleaseStrategy}.
* Set {@code releasePartialSequences} on an underlying default
* {@link SequenceSizeReleaseStrategy}. Ignored for other release strategies.
* @param releasePartialSequences true to allow release.
*/
public void setReleasePartialSequences(boolean releasePartialSequences) {
if (!this.releaseStrategySet && releasePartialSequences) {
setReleaseStrategy(new SequenceSizeReleaseStrategy());
}
this.releasePartialSequences = releasePartialSequences;
}
@@ -384,8 +392,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
Object correlationKey = this.correlationStrategy.getCorrelationKey(message);
Assert.state(correlationKey != null, "Null correlation not allowed. Maybe the CorrelationStrategy is failing?");
if (logger.isDebugEnabled()) {
logger.debug("Handling message with correlationKey [" + correlationKey + "]: " + message);
if (this.logger.isDebugEnabled()) {
this.logger.debug("Handling message with correlationKey [" + correlationKey + "]: " + message);
}
UUID groupIdUuid = UUIDConverter.getUUID(correlationKey);
@@ -396,8 +404,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
ScheduledFuture<?> scheduledFuture = this.expireGroupScheduledFutures.remove(groupIdUuid);
if (scheduledFuture != null) {
boolean canceled = scheduledFuture.cancel(true);
if (canceled && logger.isDebugEnabled()) {
logger.debug("Cancel 'forceComplete' scheduling for MessageGroup with Correlation Key [ "
if (canceled && this.logger.isDebugEnabled()) {
this.logger.debug("Cancel 'forceComplete' scheduling for MessageGroup with Correlation Key [ "
+ correlationKey + "].");
}
}
@@ -407,8 +415,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
}
if (!messageGroup.isComplete() && messageGroup.canAdd(message)) {
if (logger.isTraceEnabled()) {
logger.trace("Adding message to group [ " + messageGroup + "]");
if (this.logger.isTraceEnabled()) {
this.logger.trace("Adding message to group [ " + messageGroup + "]");
}
messageGroup = this.store(correlationKey, message);
@@ -446,26 +454,21 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
if (groupTimeout > 0) {
final Object groupId = messageGroup.getGroupId();
ScheduledFuture<?> scheduledFuture = getTaskScheduler()
.schedule(new Runnable() {
@Override
public void run() {
try {
processForceRelease(groupId);
}
catch (MessageDeliveryException e) {
if (logger.isDebugEnabled()) {
logger.debug("The MessageGroup [ " + groupId +
"] is rescheduled by the reason: " + e.getMessage());
}
scheduleGroupToForceComplete(groupId);
}
.schedule(() -> {
try {
processForceRelease(groupId);
}
catch (MessageDeliveryException e) {
if (AbstractCorrelatingMessageHandler.this.logger.isDebugEnabled()) {
AbstractCorrelatingMessageHandler.this.logger.debug("The MessageGroup [ "
+ groupId + "] is rescheduled by the reason: " + e.getMessage());
}
scheduleGroupToForceComplete(groupId);
}
}, new Date(System.currentTimeMillis() + groupTimeout));
if (logger.isDebugEnabled()) {
logger.debug("Schedule MessageGroup [ " + messageGroup + "] to 'forceComplete'.");
if (this.logger.isDebugEnabled()) {
this.logger.debug("Schedule MessageGroup [ " + messageGroup + "] to 'forceComplete'.");
}
this.expireGroupScheduledFutures.put(UUIDConverter.getUUID(groupId), scheduledFuture);
}
@@ -520,8 +523,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
this.expireGroupScheduledFutures.remove(UUIDConverter.getUUID(correlationKey));
if (scheduledFuture != null) {
boolean canceled = scheduledFuture.cancel(false);
if (canceled && logger.isDebugEnabled()) {
logger.debug("Cancel 'forceComplete' scheduling for MessageGroup [ " + group + "].");
if (canceled && this.logger.isDebugEnabled()) {
this.logger.debug("Cancel 'forceComplete' scheduling for MessageGroup [ " + group + "].");
}
}
MessageGroup groupNow = group;
@@ -568,23 +571,23 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
* setting minimumTimeoutForEmptyGroups.
*/
removeGroup = lastModifiedNow <= (System.currentTimeMillis() - this.minimumTimeoutForEmptyGroups);
if (removeGroup && logger.isDebugEnabled()) {
logger.debug("Removing empty group: " + correlationKey);
if (removeGroup && this.logger.isDebugEnabled()) {
this.logger.debug("Removing empty group: " + correlationKey);
}
}
}
else {
removeGroup = false;
if (logger.isDebugEnabled()) {
logger.debug("Group expiry candidate (" + correlationKey +
if (this.logger.isDebugEnabled()) {
this.logger.debug("Group expiry candidate (" + correlationKey +
") has changed - it may be reconsidered for a future expiration");
}
}
}
catch (MessageDeliveryException e) {
removeGroup = false;
if (logger.isDebugEnabled()) {
logger.debug("Group expiry candidate (" + correlationKey +
if (this.logger.isDebugEnabled()) {
this.logger.debug("Group expiry candidate (" + correlationKey +
") has been affected by MessageDeliveryException - " +
"it may be reconsidered for a future expiration one more time");
}
@@ -603,7 +606,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
}
catch (InterruptedException ie) {
Thread.currentThread().interrupt();
logger.debug("Thread was interrupted while trying to obtain lock");
this.logger.debug("Thread was interrupted while trying to obtain lock");
}
}
@@ -622,19 +625,19 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
}
protected void expireGroup(Object correlationKey, MessageGroup group) {
if (logger.isInfoEnabled()) {
logger.info("Expiring MessageGroup with correlationKey[" + correlationKey + "]");
if (this.logger.isInfoEnabled()) {
this.logger.info("Expiring MessageGroup with correlationKey[" + correlationKey + "]");
}
if (this.sendPartialResultOnExpiry) {
if (logger.isDebugEnabled()) {
logger.debug("Prematurely releasing partially complete group with key ["
if (this.logger.isDebugEnabled()) {
this.logger.debug("Prematurely releasing partially complete group with key ["
+ correlationKey + "] to: " + getOutputChannel());
}
completeGroup(correlationKey, group);
}
else {
if (logger.isDebugEnabled()) {
logger.debug("Discarding messages of partially complete group with key ["
if (this.logger.isDebugEnabled()) {
this.logger.debug("Discarding messages of partially complete group with key ["
+ correlationKey + "] to: "
+ (this.discardChannelName != null ? this.discardChannelName : this.discardChannel));
}
@@ -658,8 +661,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
@SuppressWarnings("unchecked")
protected Collection<Message<?>> completeGroup(Message<?> message, Object correlationKey, MessageGroup group) {
if (logger.isDebugEnabled()) {
logger.debug("Completing group with correlationKey [" + correlationKey + "]");
if (this.logger.isDebugEnabled()) {
this.logger.debug("Completing group with correlationKey [" + correlationKey + "]");
}
Object result = this.outputProcessor.processMessageGroup(group);

15
spring-integration-core/src/main/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategy.java
View File

@@ -28,8 +28,9 @@ import org.springframework.integration.store.MessageGroup;
import org.springframework.messaging.Message;
/**
* An implementation of {@link ReleaseStrategy} that simply compares the current size of the message list to the
* expected 'sequenceSize'.
* An implementation of {@link ReleaseStrategy} that simply compares the current size of
* the message list to the expected 'sequenceSize'. Supports release of partial sequences.
* Correlating message handlers prevent the addition of duplicate sequences to the group.
*
* @author Mark Fisher
* @author Marius Bogoevici
@@ -47,10 +48,19 @@ public class SequenceSizeReleaseStrategy implements ReleaseStrategy {
private volatile boolean releasePartialSequences;
/**
* Construct an instance that does not support releasing partial sequences.
*/
public SequenceSizeReleaseStrategy() {
this(false);
}
/**
* Construct an instance that supports releasing partial sequences if
* releasePartialSequences is true. This can be an expensive operation on large
* groups.
* @param releasePartialSequences true to allow the release of partial sequences.
*/
public SequenceSizeReleaseStrategy(boolean releasePartialSequences) {
this.releasePartialSequences = releasePartialSequences;
}
@@ -59,6 +69,7 @@ public class SequenceSizeReleaseStrategy implements ReleaseStrategy {
* Flag that determines if partial sequences are allowed. If true then as soon as
* enough messages arrive that can be ordered they will be released, provided they
* all have sequence numbers greater than those already released.
* This can be an expensive operation for large groups.
* @param releasePartialSequences true when partial sequences should be released.
*/
public void setReleasePartialSequences(boolean releasePartialSequences) {

37
spring-integration-core/src/main/java/org/springframework/integration/aggregator/SimpleSequenceSizeReleaseStrategy.java Normal file
View File

@@ -0,0 +1,37 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.aggregator;
import org.springframework.integration.store.MessageGroup;
/**
* An implementation of {@link ReleaseStrategy} that simply compares the current size of
* the message list to the expected 'sequenceSize'. It does not support releasing partial
* sequences. Correlating message handlers using this strategy do not check for duplicate
* sequence numbers.
* @author Gary Russell
* @since 4.3.4
*
*/
public class SimpleSequenceSizeReleaseStrategy implements ReleaseStrategy {
@Override
public boolean canRelease(MessageGroup group) {
return group.getSequenceSize() == group.size();
}
}

1
spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java
View File

@@ -145,6 +145,7 @@ public interface MessageGroupStore extends BasicMessageGroupStore {
/**
* Invoked when a MessageGroupStore expires a group.
*/
@FunctionalInterface
interface MessageGroupCallback {
void execute(MessageGroupStore messageGroupStore, MessageGroup group);

1
spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java
View File

@@ -415,6 +415,7 @@ public class AggregatorTests {
Message<?> message2 = createMessage(5, "ABC", 3, 2, replyChannel, null);
Message<?> message3 = createMessage(7, "ABC", 3, 3, replyChannel, null);
Message<?> message4 = createMessage(7, "ABC", 3, 3, replyChannel, null);
this.aggregator.setReleaseStrategy(new SequenceSizeReleaseStrategy());
this.aggregator.handleMessage(message1);
this.aggregator.handleMessage(message3);

1
spring-integration-core/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java
View File

@@ -284,6 +284,7 @@ public class ResequencerTests {
Message<?> message2 = createMessage("456", "ABC", 5, 1, null);
this.resequencer.setSendPartialResultOnExpiry(false);
this.resequencer.setDiscardChannel(discardChannel);
this.resequencer.setReleasePartialSequences(true); // force SequenceSizeReleaseStrategy
this.resequencer.handleMessage(message1);
this.resequencer.handleMessage(message2);
// this.resequencer.discardBarrier(this.resequencer.barriers.get("ABC"));

6
spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java
View File

@@ -33,7 +33,7 @@ import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.aggregator.MethodInvokingCorrelationStrategy;
import org.springframework.integration.aggregator.MethodInvokingReleaseStrategy;
import org.springframework.integration.aggregator.SequenceSizeReleaseStrategy;
import org.springframework.integration.aggregator.SimpleSequenceSizeReleaseStrategy;
import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.test.util.TestUtils;
@@ -52,7 +52,7 @@ public class AggregatorAnnotationTests {
new String[] { "classpath:/org/springframework/integration/config/annotation/testAnnotatedAggregator.xml" });
final String endpointName = "endpointWithDefaultAnnotation";
MessageHandler aggregator = this.getAggregator(context, endpointName);
assertTrue(getPropertyValue(aggregator, "releaseStrategy") instanceof SequenceSizeReleaseStrategy);
assertTrue(getPropertyValue(aggregator, "releaseStrategy") instanceof SimpleSequenceSizeReleaseStrategy);
assertNull(getPropertyValue(aggregator, "outputChannel"));
assertTrue(getPropertyValue(aggregator, "discardChannel") instanceof NullChannel);
assertEquals(-1L, getPropertyValue(aggregator, "messagingTemplate.sendTimeout"));
@@ -65,7 +65,7 @@ public class AggregatorAnnotationTests {
new String[] { "classpath:/org/springframework/integration/config/annotation/testAnnotatedAggregator.xml" });
final String endpointName = "endpointWithCustomizedAnnotation";
MessageHandler aggregator = this.getAggregator(context, endpointName);
assertTrue(getPropertyValue(aggregator, "releaseStrategy") instanceof SequenceSizeReleaseStrategy);
assertTrue(getPropertyValue(aggregator, "releaseStrategy") instanceof SimpleSequenceSizeReleaseStrategy);
assertEquals("outputChannel", getPropertyValue(aggregator, "outputChannelName"));
assertEquals("discardChannel", getPropertyValue(aggregator, "discardChannelName"));
assertEquals(98765432L, getPropertyValue(aggregator, "messagingTemplate.sendTimeout"));