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

INT-3386 Improve Aggregator Performance

Browse Source 
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)
This commit is contained in:
Gary Russell
2014-05-20 18:08:38 -04:00
parent 39ef91ff84
commit f396b2a4b4
5 changed files with 172 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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) {

16
spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java
View File

@@ -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() {

34
spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java
View File

@@ -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;

11
spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java
View File

@@ -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<String>("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<String>("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);

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

@@ -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<String>("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<Message<?>> messages = new ArrayList<Message<?>>(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<Object> payloads = new ArrayList<Object>(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<String>("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 {