This version is still in development and is not considered stable yet. For the latest stable version, please use Spring Integration 6.3.4!spring-doc.cn

This version is still in development and is not considered stable yet. For the latest stable version, please use Spring Integration 6.3.4!spring-doc.cn

An Aggregator is conceptually the opposite of a Splitter. It aggregates a sequence of individual messages into a single message and is necessarily more complex. By default, an aggregator returns a message that contains a collection of payloads from incoming messages. The same rules are applied for the Resequencer. The following example shows a canonical example of the splitter-aggregator pattern:spring-doc.cn

@Bean
public IntegrationFlow splitAggregateFlow() {
    return IntegrationFlow.from("splitAggregateInput")
            .split()
            .channel(MessageChannels.executor(this.taskExecutor()))
            .resequence()
            .aggregate()
            .get();
}

The split() method splits the list into individual messages and sends them to the ExecutorChannel. The resequence() method reorders messages by sequence details found in the message headers. The aggregate() method collects those messages.spring-doc.cn

However, you can change the default behavior by specifying a release strategy and correlation strategy, among other things. Consider the following example:spring-doc.cn

.aggregate(a ->
        a.correlationStrategy(m -> m.getHeaders().get("myCorrelationKey"))
            .releaseStrategy(g -> g.size() > 10)
            .messageStore(messageStore()))

The preceding example correlates messages that have myCorrelationKey headers and releases the messages once at least ten have been accumulated.spring-doc.cn

Similar lambda configurations are provided for the resequence() EIP method.spring-doc.cn