Spring Integration 5.x Java DSL聚合器实现及过期处理问询

你提到的通过实现Consumer<AggregatorSpec>来配置聚合器的方式完全可行，下面是完整的可运行示例，包含分组聚合、消息数量校验、24小时过期清理的逻辑：

1. 实体类定义 #

先给出基础的实体类实现（可根据你的业务需求调整）：

import lombok.Data;
import java.util.UUID;

@Data
public class JobInfo {
    private UUID jobId;
    private String jobName;
    // 其他业务字段
}

@Data
public class MiniJob {
    private UUID jobInfoUuid;
    private MiniJobStatus status;
    private String data;
    // 其他业务字段
}

public enum MiniJobStatus {
    STEP1, STEP2, STEP3, STEP4
}

2. 聚合器配置与集成流 #

核心配置包含分组策略、释放规则、过期管理及聚合逻辑：

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.dsl.MessageChannels;
import org.springframework.integration.aggregator.AggregatorSpec;
import org.springframework.integration.aggregator.CorrelationStrategy;
import org.springframework.integration.aggregator.ReleaseStrategy;
import org.springframework.integration.store.MessageGroupStore;
import org.springframework.integration.store.SimpleMessageStore;
import org.springframework.messaging.Message;

import java.time.Duration;
import java.util.List;
import java.util.UUID;
import java.util.stream.Collectors;

@Configuration
public class JobIntegrationConfig {

    // 输入通道：接收分散的MiniJob消息
    @Bean
    public DirectChannel miniJobInputChannel() {
        return MessageChannels.direct().get();
    }

    // 输出通道：聚合完成后的JobInfo发送至此
    @Bean
    public DirectChannel aggregatedJobInfoOutputChannel() {
        return MessageChannels.direct().get();
    }

    // 消息存储：用于暂存待聚合的消息分组，支持过期清理
    @Bean
    public MessageGroupStore messageGroupStore() {
        SimpleMessageStore messageStore = new SimpleMessageStore();
        // 设置分组过期时间：24小时
        messageStore.setGroupTimeout(Duration.ofHours(24).toMillis());
        return messageStore;
    }

    // 聚合器配置Consumer
    @Bean
    public Consumer<AggregatorSpec> jobAggregatorConsumer() {
        return spec -> spec
                // 按消息头的job-info-uuid分组
                .correlationStrategy(correlationStrategy())
                // 分组内凑齐4条消息时触发聚合
                .releaseStrategy(releaseStrategy())
                // 绑定消息存储，实现过期管理
                .messageStore(messageGroupStore())
                // 自定义聚合逻辑：将4个MiniJob合并为JobInfo
                .processor(this::aggregateMiniJobsToJobInfo)
                // 聚合完成后自动删除分组
                .expireGroupsUponCompletion(true)
                // 过期时不发送部分聚合结果，直接清理分组
                .sendPartialResultOnExpiry(false);
    }

    // 分组策略实现
    private CorrelationStrategy correlationStrategy() {
        return message -> message.getHeaders().get("job-info-uuid", UUID.class);
    }

    // 释放策略实现
    private ReleaseStrategy releaseStrategy() {
        return group -> group.size() == 4;
    }

    // 聚合逻辑：合并同一Job的MiniJob为JobInfo
    private JobInfo aggregateMiniJobsToJobInfo(List<Message<MiniJob>> messages) {
        UUID jobId = messages.get(0).getPayload().getJobInfoUuid();
        JobInfo aggregatedJob = new JobInfo();
        aggregatedJob.setJobId(jobId);
        aggregatedJob.setJobName("AGGREGATED_JOB_" + jobId);

        // 可根据业务需求添加更多校验或数据整合逻辑
        List<MiniJobStatus> completedSteps = messages.stream()
                .map(msg -> msg.getPayload().getStatus())
                .collect(Collectors.toList());

        return aggregatedJob;
    }

    // 集成流定义：串联输入、聚合、输出环节
    @Bean
    public IntegrationFlow jobAggregationFlow() {
        return IntegrationFlows.from(miniJobInputChannel())
                .aggregate(jobAggregatorConsumer())
                .channel(aggregatedJobInfoOutputChannel())
                .get();
    }
}

3. 测试示例 #

验证聚合和过期逻辑的测试代码：

import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;

import java.util.UUID;

@SpringBootTest
public class JobAggregationTest {

    @Autowired
    private DirectChannel miniJobInputChannel;

    @Autowired
    private DirectChannel aggregatedJobInfoOutputChannel;

    @Test
    public void testFullAggregation() {
        UUID jobUuid = UUID.randomUUID();

        // 发送同一Job的4个MiniJob
        for (MiniJobStatus status : MiniJobStatus.values()) {
            MiniJob miniJob = new MiniJob();
            miniJob.setJobInfoUuid(jobUuid);
            miniJob.setStatus(status);
            miniJob.setData("DATA_FOR_" + status);

            Message<MiniJob> message = MessageBuilder.withPayload(miniJob)
                    .setHeader("job-info-uuid", jobUuid)
                    .build();

            miniJobInputChannel.send(message);
        }

        // 验证聚合结果
        Message<JobInfo> aggregatedMessage = (Message<JobInfo>) aggregatedJobInfoOutputChannel.receive(1000);
        assert aggregatedMessage != null;
        assert aggregatedMessage.getPayload().getJobId().equals(jobUuid);
    }

    @Test
    public void testGroupExpiry() throws InterruptedException {
        UUID jobUuid = UUID.randomUUID();

        // 只发送3个MiniJob，不满足聚合条件
        MiniJob miniJob1 = new MiniJob();
        miniJob1.setJobInfoUuid(jobUuid);
        miniJob1.setStatus(MiniJobStatus.STEP1);
        miniJobInputChannel.send(MessageBuilder.withPayload(miniJob1)
                .setHeader("job-info-uuid", jobUuid)
                .build());

        MiniJob miniJob2 = new MiniJob();
        miniJob2.setJobInfoUuid(jobUuid);
        miniJob2.setStatus(MiniJobStatus.STEP2);
        miniJobInputChannel.send(MessageBuilder.withPayload(miniJob2)
                .setHeader("job-info-uuid", jobUuid)
                .build());

        MiniJob miniJob3 = new MiniJob();
        miniJob3.setJobInfoUuid(jobUuid);
        miniJob3.setStatus(MiniJobStatus.STEP3);
        miniJobInputChannel.send(MessageBuilder.withPayload(miniJob3)
                .setHeader("job-info-uuid", jobUuid)
                .build());

        // 测试时建议将messageStore的超时改为5秒，等待过期
        Thread.sleep(6000);

        // 验证输出通道无消息（分组已被清理）
        Message<?> message = aggregatedJobInfoOutputChannel.receive(1000);
        assert message == null;
    }
}

关键配置说明 #

• 分组策略：通过correlationStrategy绑定job-info-uuid头，确保同一Job的MiniJob进入同一分组
• 释放策略：releaseStrategy设置为凑齐4条消息时触发聚合，符合你的业务需求
• 过期机制：通过messageStore.setGroupTimeout()设置24小时超时，sendPartialResultOnExpiry(false)确保过期时仅清理分组，不发送不完整结果
• 聚合处理器：processor方法可根据实际业务扩展，比如校验所有步骤状态、整合业务数据等

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