INT-4193: Large Group Aggregation Performance
JIRA: https://jira.spring.io/browse/INT-4193 When using the internal `SequenceAwareMessageGroup` within an correlating message handler, the messages were copied to a new collection before checking for duplicate sequences in `canAdd()`. This was unnecessary since we never add anything to this group, if `canAdd()` returns true, the message is added to the store; this group is discarded. Instead, use the message collection from the original group; although it is not modifiable, this is not an issue because we don't need to modify it. Conflicts: spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroupFactory.java src/reference/asciidoc/aggregator.adoc * Polishing `aggregator.adoc` to reflect reality
This commit is contained in:
Gary Russell
committed by
Artem Bilan
Artem Bilan
parent
168d91cbaa
commit
318bb4c4b7
@@ -753,13 +753,20 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
|
||||
protected static class SequenceAwareMessageGroup extends SimpleMessageGroup {
|
||||
|
||||
public SequenceAwareMessageGroup(MessageGroup messageGroup) {
|
||||
super(messageGroup);
|
||||
/*
|
||||
* Since this group is temporary, and never added to, we simply use the
|
||||
* supplied group's message collection for the lookup rather than creating a
|
||||
* new group.
|
||||
*/
|
||||
super(messageGroup.getMessages(), null, messageGroup.getGroupId(), messageGroup.getTimestamp(),
|
||||
messageGroup.isComplete(), true);
|
||||
}
|
||||
|
||||
/**
|
||||
* This method determines whether messages have been added to this group that supersede the given message based on
|
||||
* its sequence id. This can be helpful to avoid ending up with sequences larger than their required sequence size
|
||||
* or sequences that are missing certain sequence numbers.
|
||||
* This method determines whether messages have been added to this group that
|
||||
* supersede the given message based on its sequence id. This can be helpful to
|
||||
* avoid ending up with sequences larger than their required sequence size or
|
||||
* sequences that are missing certain sequence numbers.
|
||||
*/
|
||||
@Override
|
||||
public boolean canAdd(Message<?> message) {
|
||||
|
||||
@@ -67,20 +67,22 @@ public class SimpleMessageGroup implements MessageGroup {
|
||||
|
||||
public SimpleMessageGroup(Collection<? extends Message<?>> messages, Object groupId, long timestamp,
|
||||
boolean complete) {
|
||||
this(new LinkedHashSet<Message<?>>(), messages, groupId, timestamp, complete);
|
||||
this(new LinkedHashSet<Message<?>>(), messages, groupId, timestamp, complete, false);
|
||||
}
|
||||
|
||||
SimpleMessageGroup(Collection<Message<?>> internalStore, Collection<? extends Message<?>> messages, Object groupId,
|
||||
long timestamp, boolean complete) {
|
||||
protected SimpleMessageGroup(Collection<Message<?>> internalStore, Collection<? extends Message<?>> messages,
|
||||
Object groupId, long timestamp, boolean complete, boolean storePreLoaded) {
|
||||
Assert.notNull(internalStore, "'internalStore' must not be null");
|
||||
Assert.notNull(messages, "'messages' must not be null");
|
||||
this.messages = internalStore;
|
||||
this.groupId = groupId;
|
||||
this.timestamp = timestamp;
|
||||
this.complete = complete;
|
||||
for (Message<?> message : messages) {
|
||||
if (message != null) { //see INT-2666
|
||||
addMessage(message);
|
||||
if (!storePreLoaded) {
|
||||
Assert.notNull(messages, "'messages' must not be null");
|
||||
for (Message<?> message : messages) {
|
||||
if (message != null) { //see INT-2666
|
||||
addMessage(message);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -56,7 +56,7 @@ public class SimpleMessageGroupFactory implements MessageGroupFactory {
|
||||
@Override
|
||||
public MessageGroup create(Collection<? extends Message<?>> messages, Object groupId, long timestamp,
|
||||
boolean complete) {
|
||||
return new SimpleMessageGroup(this.type.get(), messages, groupId, timestamp, complete);
|
||||
return new SimpleMessageGroup(this.type.get(), messages, groupId, timestamp, complete, false);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
@@ -26,10 +26,12 @@ import java.lang.reflect.Constructor;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collection;
|
||||
import java.util.Collections;
|
||||
import java.util.HashSet;
|
||||
import java.util.List;
|
||||
|
||||
import org.junit.Test;
|
||||
|
||||
import org.springframework.beans.DirectFieldAccessor;
|
||||
import org.springframework.integration.support.MessageBuilder;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.support.GenericMessage;
|
||||
@@ -40,6 +42,7 @@ import org.springframework.util.StopWatch;
|
||||
* @author Oleg Zhurakousky
|
||||
* @author Dave Syer
|
||||
* @author Artem Bilan
|
||||
* @author Gary Russell
|
||||
*/
|
||||
public class SimpleMessageGroupTests {
|
||||
|
||||
@@ -54,11 +57,12 @@ public class SimpleMessageGroupTests {
|
||||
Constructor<SimpleMessageGroup> ctr = clazz.getDeclaredConstructor(MessageGroup.class);
|
||||
ctr.setAccessible(true);
|
||||
group = ctr.newInstance(group);
|
||||
new DirectFieldAccessor(group).setPropertyValue("messages", new HashSet<Message<?>>());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void shouldFindSupersedingMessagesIfSequenceAware() throws Exception {
|
||||
this.prepareForSequenceAwareMessageGroup();
|
||||
prepareForSequenceAwareMessageGroup();
|
||||
final Message<?> message1 = MessageBuilder.withPayload("test").setSequenceNumber(1).build();
|
||||
final Message<?> message2 = MessageBuilder.fromMessage(message1).setSequenceNumber(1).build();
|
||||
assertThat(group.canAdd(message1), is(true));
|
||||
@@ -69,7 +73,7 @@ public class SimpleMessageGroupTests {
|
||||
|
||||
@Test
|
||||
public void shouldIgnoreMessagesWithZeroSequenceNumberIfSequenceAware() throws Exception {
|
||||
this.prepareForSequenceAwareMessageGroup();
|
||||
prepareForSequenceAwareMessageGroup();
|
||||
final Message<?> message1 = MessageBuilder.withPayload("test").build();
|
||||
final Message<?> message2 = MessageBuilder.fromMessage(message1).build();
|
||||
assertThat(group.canAdd(message1), is(true));
|
||||
|
||||
@@ -221,6 +221,35 @@ As shown above, it is also the default strategy.
|
||||
NOTE: Before _version 5.0_, the default release strategy was `SequenceSizeReleaseStrategy` which does not perform well with large groups.
|
||||
With that strategy, duplicate sequence numbers are detected and rejected; this operation can be expensive.
|
||||
|
||||
If you are aggregating large groups, you don't need to release partial groups, and you don't need to detect/reject duplicate sequences, consider using the `SimpleSequenceSizeReleaseStrategy` instead - it is much more efficient for these use cases, and is the default since _version 5.0_ when partial group release is not specified.
|
||||
|
||||
===== Aggregating Large Groups
|
||||
|
||||
The 4.3 release changed the default `Collection` for messages in a `SimpleMessageGroup` to `HashSet` (it was previously a `BlockingQueue`).
|
||||
This was expensive when removing individual messages from large groups (an O(n) linear scan was required).
|
||||
Although the hash set is generally much faster for removing, it can be expensive for large messages because the hash has to be calculated (on both inserts and removes).
|
||||
If you have messages that are expensive to hash, consider using some other collection type.
|
||||
As discussed in <<message-group-factory>>, a `SimpleMessageGroupFactory` is provided so you can select the `Collection` that best suits your needs.
|
||||
You can also provide your own factory implementation to create some other `Collection<Message<?>>`.
|
||||
|
||||
Here is an example of how to configure an aggregator with the previous implementation and a `SimpleSequenceSizeReleaseStrategy`.
|
||||
|
||||
[source, xml]
|
||||
----
|
||||
<int:aggregator input-channel="aggregate"
|
||||
output-channel="out" message-store="store" release-strategy="releaser" />
|
||||
|
||||
<bean id="store" class="org.springframework.integration.store.SimpleMessageStore">
|
||||
<property name="messageGroupFactory">
|
||||
<bean class="org.springframework.integration.store.SimpleMessageGroupFactory">
|
||||
<constructor-arg value="BLOCKING_QUEUE"/>
|
||||
</bean>
|
||||
</property>
|
||||
</bean>
|
||||
|
||||
<bean id="releaser" class="SimpleSequenceSizeReleaseStrategy" />
|
||||
----
|
||||
|
||||
===== CorrelationStrategy
|
||||
|
||||
The `CorrelationStrategy` interface is defined as follows:
|
||||
@@ -635,9 +664,9 @@ For this purpose the `groupTimeout` option allows scheduling the `MessageGroup`
|
||||
[source,xml]
|
||||
----
|
||||
<aggregator input-channel="input" output-channel="output"
|
||||
send-partial-result-on-expiry="true"
|
||||
group-timeout-expression="size() ge 2 ? 10000 : -1"
|
||||
release-strategy-expression="[0].headers.sequenceNumber == [0].headers.sequenceSize"/>
|
||||
send-partial-result-on-expiry="true"
|
||||
group-timeout-expression="size() ge 2 ? 10000 : -1"
|
||||
release-strategy-expression="[0].headers.sequenceNumber == [0].headers.sequenceSize"/>
|
||||
----
|
||||
|
||||
With this example, the normal _release_ will be possible if the aggregator receives the last message in sequence as defined by the `release-strategy-expression`.
|
||||
|
||||
Reference in New Issue
Block a user