From f396b2a4b412ec7da7e7dd1ed4a9c22c2f2e77b3 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 20 May 2014 18:08:38 -0400 Subject: [PATCH] INT-3386 Improve Aggregator Performance JIRA: https://jira.spring.io/browse/INT-3386 Eliminate unnecessary group copy during retrieval in the aggregator. This is expensive with large groups. Ensure that the reaper uses a snapshot of the group at the time it is considered for reaping. AggregatorTests.testAggPerf() - before: AggregatorTests [main] : Sent 0 in 0.0 (10k in 0ms) AggregatorTests [main] : Sent 10000 in 4.872 (10k in 4872ms) AggregatorTests [main] : Sent 20000 in 16.86 (10k in 11988ms) AggregatorTests [main] : Sent 30000 in 36.809 (10k in 19949ms) AggregatorTests [main] : Sent 40000 in 65.045 (10k in 28236ms) AggregatorTests [main] : Sent 50000 in 100.329 (10k in 35284ms) AggregatorTests [main] : Received 60000 AggregatorTests [main] : Sent 60000 in 143.106 (10k in 42777ms) AggregatorTests [main] : Sent 70000 in 147.027 (10k in 3921ms) AggregatorTests [main] : Sent 80000 in 158.664 (10k in 11637ms) AggregatorTests [main] : Sent 90000 in 177.956 (10k in 19292ms) AggregatorTests [main] : Sent 100000 in 205.045 (10k in 27089ms) AggregatorTests [main] : Sent 110000 in 240.005 (10k in 34960ms) AggregatorTests [main] : Received 60000 AggregatorTests [main] : Sent 120000 in 282.566 (10k in 42561ms) After: AggregatorTests [main] : Sent 0 in 0.0 (10k in 0ms) AggregatorTests [main] : Sent 10000 in 0.612 (10k in 612ms) AggregatorTests [main] : Sent 20000 in 0.925 (10k in 313ms) AggregatorTests [main] : Sent 30000 in 1.131 (10k in 206ms) AggregatorTests [main] : Sent 40000 in 1.288 (10k in 157ms) AggregatorTests [main] : Sent 50000 in 1.361 (10k in 73ms) AggregatorTests [main] : Received 60000 AggregatorTests [main] : Sent 60000 in 1.522 (10k in 161ms) AggregatorTests [main] : Sent 70000 in 1.619 (10k in 97ms) AggregatorTests [main] : Sent 80000 in 1.696 (10k in 77ms) AggregatorTests [main] : Sent 90000 in 1.75 (10k in 54ms) AggregatorTests [main] : Sent 100000 in 1.81 (10k in 60ms) AggregatorTests [main] : Sent 110000 in 1.851 (10k in 41ms) AggregatorTests [main] : Received 60000 AggregatorTests [main] : Sent 120000 in 1.918 (10k in 67ms) --- .../AbstractCorrelatingMessageHandler.java | 2 +- .../store/AbstractMessageGroupStore.java | 16 ++- .../integration/store/SimpleMessageStore.java | 34 +++++ ...bstractCorrelatingMessageHandlerTests.java | 11 +- .../aggregator/AggregatorTests.java | 116 ++++++++++++++++++ 5 files changed, 172 insertions(+), 7 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java index 8a49022b97..9fec2f6612 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java @@ -142,7 +142,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH } public AbstractCorrelatingMessageHandler(MessageGroupProcessor processor) { - this(processor, new SimpleMessageStore(0), null, null); + this(processor, SimpleMessageStore.fastMessageStore(0), null, null); } public void setLockRegistry(LockRegistry lockRegistry) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java index 75fb50dd11..a99ccefa70 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java @@ -110,12 +110,26 @@ public abstract class AbstractMessageGroupStore implements MessageGroupStore, It if (timestamp <= threshold) { count++; - expire(group); + expire(copy(group)); } } return count; } + /** + * Used by expireMessageGroups. We need to return a snapshot of the group + * at the time the reaper runs, so we can properly detect if the + * group changed between now and the attempt to expire the group. + * Not necessary for persistent stores, so the default behavior is + * to just return the group. + * @param group The group. + * @return The group, or a copy. + * @since 4.0.1 + */ + protected MessageGroup copy(MessageGroup group) { + return group; + } + @Override @ManagedAttribute public int getMessageCountForAllMessageGroups() { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java index d9fef25c1e..433a0f750d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java @@ -57,6 +57,8 @@ public class SimpleMessageStore extends AbstractMessageGroupStore private volatile boolean isUsed; + private volatile boolean copyOnGet = true; // TODO: default false in 4.1 + /** * Creates a SimpleMessageStore with a maximum size limited by the given capacity, or unlimited size if the given * capacity is less than 1. The capacities are applied independently to messages stored via @@ -105,6 +107,28 @@ public class SimpleMessageStore extends AbstractMessageGroupStore this(0); } + /** + * Factory method to return a simple message store that does not + * copy the group in {@link #getMessageGroup(Object)}. + * @param capacity the capacity (0 for unlimited). + * @return the store. + * @since 4.0.1 + */ + public static SimpleMessageStore fastMessageStore(int capacity) { + SimpleMessageStore store = new SimpleMessageStore(capacity); + store.setCopyOnGet(false); + return store; + } + + /** + * Set to false to disable copying the group in {@link #getMessageGroup(Object)}. + * @param copyOnGet True to copy, false to not. + * @since 4.0.1 + */ + public void setCopyOnGet(boolean copyOnGet) { + this.copyOnGet = copyOnGet; + } + public void setLockRegistry(LockRegistry lockRegistry) { Assert.notNull(lockRegistry, "The LockRegistry cannot be null"); Assert.isTrue(!(this.isUsed), "Cannot change the lock registry after the store has been used"); @@ -152,6 +176,16 @@ public class SimpleMessageStore extends AbstractMessageGroupStore if (group == null) { return new SimpleMessageGroup(groupId); } + if (this.copyOnGet) { + return copy(group); + } + else { + return group; + } + } + + @Override + protected MessageGroup copy(MessageGroup group) { SimpleMessageGroup simpleMessageGroup = new SimpleMessageGroup(group); simpleMessageGroup.setLastModified(group.getLastModified()); return simpleMessageGroup; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java index 765e2d1592..279257a6e2 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java @@ -36,15 +36,16 @@ import java.util.concurrent.TimeUnit; import org.junit.Test; import org.springframework.beans.DirectFieldAccessor; -import org.springframework.messaging.Message; -import org.springframework.messaging.MessageChannel; import org.springframework.integration.channel.QueueChannel; -import org.springframework.messaging.support.GenericMessage; import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageGroupStore; +import org.springframework.integration.store.SimpleMessageGroup; import org.springframework.integration.store.SimpleMessageStore; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.test.util.TestUtils; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.support.GenericMessage; /** * @author Gary Russell @@ -349,7 +350,7 @@ public class AbstractCorrelatingMessageHandlerTests { handler.setOutputChannel(outputChannel); MessageGroupStore mgs = TestUtils.getPropertyValue(handler, "messageStore", MessageGroupStore.class); mgs.addMessageToGroup("foo", new GenericMessage("foo")); - MessageGroup group = mgs.getMessageGroup("foo"); + MessageGroup group = new SimpleMessageGroup(mgs.getMessageGroup("foo")); mgs.completeGroup("foo"); mgs = spy(mgs); new DirectFieldAccessor(handler).setPropertyValue("messageStore", mgs); @@ -383,7 +384,7 @@ public class AbstractCorrelatingMessageHandlerTests { handler.setOutputChannel(outputChannel); MessageGroupStore mgs = TestUtils.getPropertyValue(handler, "messageStore", MessageGroupStore.class); mgs.addMessageToGroup("foo", new GenericMessage("foo")); - MessageGroup group = mgs.getMessageGroup("foo"); + MessageGroup group = new SimpleMessageGroup(mgs.getMessageGroup("foo")); mgs = spy(mgs); new DirectFieldAccessor(handler).setPropertyValue("messageStore", mgs); Method forceComplete = AbstractCorrelatingMessageHandler.class.getDeclaredMethod("forceComplete", MessageGroup.class); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java index fde65e3ab7..9b333ad107 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java @@ -20,19 +20,32 @@ import static org.junit.Assert.assertNull; import static org.junit.Assert.assertThat; import static org.mockito.Mockito.mock; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.concurrent.locks.ReentrantLock; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.junit.Before; import org.junit.Ignore; import org.junit.Test; import org.springframework.beans.factory.BeanFactory; +import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.handler.AbstractMessageHandler; import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.SimpleMessageStore; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.MessagingException; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.util.StopWatch; /** * @author Mark Fisher @@ -42,6 +55,8 @@ import org.springframework.messaging.MessageHeaders; */ public class AggregatorTests { + private static final Log logger = LogFactory.getLog(AggregatorTests.class); + private AggregatingMessageHandler aggregator; private final SimpleMessageStore store = new SimpleMessageStore(50); @@ -54,6 +69,107 @@ public class AggregatorTests { this.aggregator.afterPropertiesSet(); } + @Test + public void testAggPerf() { + AggregatingMessageHandler handler = new AggregatingMessageHandler(new DefaultAggregatingMessageGroupProcessor()); + handler.setCorrelationStrategy(new CorrelationStrategy() { + + @Override + public Object getCorrelationKey(Message message) { + return "foo"; + } + }); + handler.setReleaseStrategy(new MessageCountReleaseStrategy(60000)); + handler.setExpireGroupsUponCompletion(true); + handler.setSendPartialResultOnExpiry(true); + DirectChannel outputChannel = new DirectChannel(); + handler.setOutputChannel(outputChannel); + outputChannel.subscribe(new MessageHandler() { + + @Override + public void handleMessage(Message message) throws MessagingException { + logger.warn("Received " + ((Collection) message.getPayload()).size()); + } + + }); + Message message = new GenericMessage("foo"); + StopWatch stopwatch = new StopWatch(); + stopwatch.start(); + for (int i=0; i < 120000; i++) { + if (i % 10000 == 0) { + stopwatch.stop(); + logger.warn("Sent " + i + " in " + stopwatch.getTotalTimeSeconds() + " (10k in " + stopwatch.getLastTaskTimeMillis() + "ms)"); + stopwatch.start(); + } + handler.handleMessage(message); + } + stopwatch.stop(); + logger.warn("Sent " + 120000 + " in " + stopwatch.getTotalTimeSeconds() + " (10k in " + stopwatch.getLastTaskTimeMillis() + "ms)"); + } + + @Test + public void testCustomAggPerf() { + class CustomHandler extends AbstractMessageHandler { + + // custom aggregator, only handles a single correlation + + private final ReentrantLock lock = new ReentrantLock(); + + private final Collection> messages = new ArrayList>(60000); + + private final MessageChannel outputChannel; + + private CustomHandler(MessageChannel outputChannel) { + this.outputChannel = outputChannel; + } + + @Override + public void handleMessageInternal(Message requestMessage) { + lock.lock(); + try { + this.messages.add(requestMessage); + if (this.messages.size() == 60000) { + List payloads = new ArrayList(this.messages.size()); + for (Message message : this.messages) { + payloads.add(message.getPayload()); + } + this.messages.clear(); + outputChannel.send(getMessageBuilderFactory().withPayload(payloads) + .copyHeaders(requestMessage.getHeaders()) + .build()); + } + } + finally { + lock.unlock(); + } + } + + }; + + DirectChannel outputChannel = new DirectChannel(); + CustomHandler handler = new CustomHandler(outputChannel); + outputChannel.subscribe(new MessageHandler() { + + @Override + public void handleMessage(Message message) throws MessagingException { + logger.warn("Received " + ((Collection) message.getPayload()).size()); + } + + }); + Message message = new GenericMessage("foo"); + StopWatch stopwatch = new StopWatch(); + stopwatch.start(); + for (int i=0; i < 120000; i++) { + if (i % 10000 == 0) { + stopwatch.stop(); + logger.warn("Sent " + i + " in " + stopwatch.getTotalTimeSeconds() + " (10k in " + stopwatch.getLastTaskTimeMillis() + "ms)"); + stopwatch.start(); + } + handler.handleMessage(message); + } + stopwatch.stop(); + logger.warn("Sent " + 120000 + " in " + stopwatch.getTotalTimeSeconds() + " (10k in " + stopwatch.getLastTaskTimeMillis() + "ms)"); + } @Test public void testCompleteGroupWithinTimeout() throws InterruptedException {