Spring Integration Aggregator分组聚合：释放策略及代码示例咨询

Great question! Let's walk through how to implement this gender-based aggregation scenario step by step, focusing on the critical release strategy and core components tailored to your Spring Batch + File integration flow.

1. Core Concepts to Understand #

First, a quick recap of Aggregator fundamentals that matter here:

• Correlation Strategy: Determines which messages belong to the same group (we’ll use gender as our grouping key).
• Release Strategy: Decides when to finalize and release a group (this is what you’re asking about—we’ll cover two practical approaches based on whether you know group sizes upfront).
• Message Group Store: Persists in-flight message groups (use in-memory for testing, persistent stores like Redis/JDBC for production).

2. Define Your Domain Model #

Start with a simple Person class to represent your input data:

public class Person {
    private String gender;
    private String name;

    // Getters, setters, and constructor
    public Person(String gender, String name) {
        this.gender = gender;
        this.name = name;
    }
}

3. Configure the Aggregator #

Option 1: Release When Group Reaches Known Size #

If you know upfront how many records belong to each gender (e.g., you counted them during Spring Batch’s file reading step), use a size-based release strategy.

Correlation Strategy (Group by Gender)

This ensures all messages with the same gender are grouped together:

@Bean
public CorrelationStrategy genderCorrelationStrategy() {
    return message -> {
        Person person = (Person) message.getPayload();
        return person.getGender(); // Use gender as the grouping key
    };
}

Release Strategy (Trigger on Group Size)

This triggers release once a group reaches the expected number of records:

@Bean
public ReleaseStrategy genderSizeReleaseStrategy() {
    return messageGroup -> {
        // Replace 2 with your actual count per gender (e.g., 2 for Male/Female in your example)
        return messageGroup.size() == 2;
    };
}

Full Aggregator Flow

Put it all together with an integration flow:

@Bean
public IntegrationFlow genderAggregatorFlow() {
    return IntegrationFlows.from("personInputChannel")
            .aggregate(aggregatorSpec -> aggregatorSpec
                    .correlationStrategy(genderCorrelationStrategy())
                    .releaseStrategy(genderSizeReleaseStrategy())
                    .messageStore(new SimpleMessageStore()) // Swap with Redis/JDBC store for production
                    .outputProcessor(messageGroup -> {
                        // Transform the group into your desired output format
                        String gender = (String) messageGroup.getGroupId();
                        List<String> names = messageGroup.getMessages().stream()
                                .map(msg -> ((Person) msg.getPayload()).getName())
                                .collect(Collectors.toList());
                        return Map.of(gender, names);
                    }))
            .channel("aggregatedOutputChannel")
            .get();
}

Option 2: Release After Spring Batch Job Completes #

If you don’t know group sizes upfront, trigger release of all completed groups once the Batch job finishes processing all records.

Add a Job Completion Listener

This listener will tell the aggregator to release all groups when the job succeeds:

@Component
public class BatchJobCompletionListener implements JobExecutionListener {

    private final AggregatorMessageHandler aggregatorHandler;

    // Inject the aggregator handler (expose it as a bean first)
    public BatchJobCompletionListener(AggregatorMessageHandler aggregatorHandler) {
        this.aggregatorHandler = aggregatorHandler;
    }

    @Override
    public void afterJob(JobExecution jobExecution) {
        if (jobExecution.getStatus() == BatchStatus.COMPLETED) {
            // Release all pending message groups
            aggregatorHandler.releaseMessageGroups(null);
        }
    }
}

Update Aggregator for Dynamic Release

Modify the aggregator to use a no-op release strategy (since we’ll trigger release manually):

@Bean
public AggregatorMessageHandler aggregatorHandler() {
    AggregatorMessageHandler handler = new AggregatorMessageHandler(
            messageGroup -> {
                // Same output processing as Option 1
                String gender = (String) messageGroup.getGroupId();
                List<String> names = messageGroup.getMessages().stream()
                        .map(msg -> ((Person) msg.getPayload()).getName())
                        .collect(Collectors.toList());
                return Map.of(gender, names);
            },
            genderCorrelationStrategy(),
            new NeverReleaseStrategy() // Don't auto-release; wait for job completion
    );
    handler.setMessageStore(new SimpleMessageStore());
    return handler;
}

@Bean
public IntegrationFlow aggregatorFlow(AggregatorMessageHandler aggregatorHandler) {
    return IntegrationFlows.from("personInputChannel")
            .handle(aggregatorHandler)
            .channel("aggregatedOutputChannel")
            .get();
}

Attach Listener to Your Batch Job

Add the listener to your Batch job configuration:

@Bean
public Job personProcessingJob(JobBuilderFactory jobBuilderFactory, Step personProcessingStep, BatchJobCompletionListener listener) {
    return jobBuilderFactory.get("person-processing-job")
            .incrementer(new RunIdIncrementer())
            .listener(listener)
            .flow(personProcessingStep)
            .end()
            .build();
}

4. Connect Spring Batch to the Aggregator #

Create a MessagingGateway to send processed Person records from your Batch ItemWriter to the aggregator:

@MessagingGateway
public interface PersonGateway {
    @Gateway(requestChannel = "personInputChannel")
    void sendPerson(Person person);
}

Use this gateway in your Batch ItemWriter:

@Bean
public ItemWriter<Person> personItemWriter(PersonGateway gateway) {
    return items -> {
        for (Person person : items) {
            gateway.sendPerson(person);
        }
    };
}

Key Notes #

• Message Store: For production, replace SimpleMessageStore with RedisMessageStore or JdbcMessageStore to avoid data loss if the application restarts.
• Sequence Headers: If you prefer using Spring’s built-in SequenceSizeReleaseStrategy, add sequenceNumber and sequenceSize headers to each message in your ItemWriter (you’ll need to track counts per gender during the file reading step).

内容的提问来源于stack exchange，提问作者Sheep