Spring Integration聚合配置异常：超时分组耗时过长，需按序发送

问题根源 #

1. 全局单消息组导致消息累积
   你配置的correlationStrategy(message -> true)将所有消息归入同一个消息组，若消息持续涌入，组会不断累积消息。由于新消息持续进入，组始终处于活跃状态，释放策略的超时条件无法触发，只有当消息数达到1000条时才会释放——这就是首次聚合耗时40秒的核心原因，后续消息量提升后，攒够1000条的时间缩短，聚合耗时才降至1-2秒。

2. 释放策略逻辑未生效
   TimeoutCountSequenceSizeReleaseStrategy(600, 1000)的超时参数（600ms）未发挥作用，因为单消息组持续有新消息加入，释放策略的超时检查被持续的新消息加入打断，实际只有计数条件（1000条）在生效。

3. 异步配置放大调度延迟
   聚合器和处理器的async(true)配置，若线程池资源不足，会导致早期消息的调度延迟，进一步拉长首次聚合的耗时。

修正方案 #

要实现按消息到达顺序分组发送（达到指定数量或超时即释放分组，分组内消息保持到达顺序），需调整以下配置：

IntegrationFlows.from(MessageChannels.queue(MailPushGateway.SEND_MAIL_NOTIFICATION_CHANNEL, Integer.MAX_VALUE))
        .enrichHeaders(headerEnricherSpec -> headerEnricherSpec.headerExpression(TracingConstants.USER_ID, "payload.getUser()"))
        .transform(Message.class, mailPushTransformer::toPushNotification)
        .aggregate(aggregatorSpec -> aggregatorSpec
                .expireGroupsUponTimeout(true)
                .expireGroupsUponCompletion(true)
                .sendPartialResultOnExpiry(true)
                // 组超时：1秒内无新消息则释放当前组
                .groupTimeout(1000)
                .requiresReply(false)
                .messageStore(new SimpleMessageStore(Integer.MAX_VALUE))
                // 绑定自定义线程池，避免调度延迟
                .async(true)
                .taskExecutor(aggregationTaskExecutor())
                // 计数释放：达到1000条立即释放
                .releaseStrategy(new CountReleaseStrategy(1000))
                // 按时间窗口生成组ID，确保同一窗口的消息进入同一组
                .correlationStrategy(message -> {
                    long timestamp = message.getHeaders().getTimestamp();
                    // 每1000ms生成一个组ID，实现按时间窗口分组
                    return timestamp / 1000;
                })
                .notPropagatedHeaders(TracingConstants.USER_ID)
                .headersFunction(messageGroup -> Map.of(TracingConstants.MESSAGES_IDS, MailPushFlow.collectMessageIds(messageGroup)))
                .outputProcessor(MailPushFlow::extractPayload)
                .advice(requestHandlerRetryAdvice)
                .releaseLockBeforeSend(true)
        )
        .handle(mailPushService, "sendMailPushNotificationChunk",
                e -> e.id(MAIL_PUSH_NOTIFICATION_GATEWAY)
                        .advice(this.requestHandlerRetryAdvice)
                        .async(true)
                        .requiresReply(false))
        .get();

// 自定义异步线程池（可选，优化调度性能）
@Bean
public TaskExecutor aggregationTaskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(10);
    executor.setMaxPoolSize(20);
    executor.setQueueCapacity(100);
    executor.setThreadNamePrefix("aggregation-");
    executor.initialize();
    return executor;
}

关键配置说明 #

• 关联策略：通过消息时间戳生成组ID，每1秒内的消息进入同一个组，避免全局单组的累积问题，确保分组按时间窗口隔离。
• 释放逻辑：结合CountReleaseStrategy(1000)和groupTimeout(1000)，实现“达到1000条立即释放”或“1秒无新消息自动释放”的双条件触发，避免消息长时间等待。
• 组过期配置：expireGroupsUponTimeout(true)和expireGroupsUponCompletion(true)确保分组释放后立即过期，新消息进入新的分组，保持分组独立性。
• 线程池优化：自定义线程池避免异步调度的资源瓶颈，提升处理效率。

验证效果 #

修正后，消息会按时间窗口（每1秒）或数量（每1000条）进行分组，分组内消息严格保持到达顺序，即使消息量较少，也会在1秒后自动释放分组，不会出现长时间等待的情况。

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