Spring Integration聚合器：下游组件如何判断分组是否完整？

问题背景 #

我们使用Spring Integration 6.3.3的Aggregator模式，搭配JDBCMessageStore并设置过期超时，开启sendPartialResultOnExpiry将超时的部分分组作为异常场景发送并做日志/告警。目前面临两个核心问题：

• 下游PetGroupHandler接收聚合消息时，无法直接判断该消息是完整分组还是超时触发的部分分组；曾尝试通过OutputProcessor注入布尔消息头标记，但因AbstractCorrelatingMessageHandler.completeGroup()中OutputProcessor执行时机早于分组标记为完成的时机，导致方案无效；且实际业务中发布策略复杂，不希望下游依赖发布策略逻辑判断。
• 期望sendPartialResultOnExpiry能指定专属通道/处理器，但当前版本无此原生配置方式，需可行替代方案。

解决方案 #

一、区分完整/部分分组的可行方案 #

方案1：自定义聚合处理器，在分组处理后添加消息头标记

不依赖OutputProcessor，直接继承AbstractCorrelatingMessageHandler，重写completeGroup()和expireGroup()方法，在消息发送前添加专属标记头：

public class CustomAggregatingMessageHandler extends AbstractCorrelatingMessageHandler {

    public CustomAggregatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store) {
        super(processor, store);
    }

    @Override
    protected void completeGroup(MessageGroup group, boolean sendResult) {
        super.completeGroup(group, sendResult);
        if (sendResult) {
            Message<?> outputMessage = getOutputMessage(group);
            Message<?> markedMessage = getMessageBuilderFactory()
                    .fromMessage(outputMessage)
                    .setHeader("is_complete_group", Boolean.TRUE)
                    .build();
            sendOutputs(markedMessage);
        }
    }

    @Override
    protected void expireGroup(MessageGroup group) {
        if (isSendPartialResultOnExpiry()) {
            Message<?> partialResult = aggregatePayloads(group, new ArrayList<>(), null);
            Message<?> markedPartial = getMessageBuilderFactory()
                    .fromMessage(partialResult)
                    .setHeader("is_partial_group", Boolean.TRUE)
                    .build();
            sendOutputs(markedPartial);
            removeGroup(group);
        } else {
            super.expireGroup(group);
        }
    }
}

下游PetGroupHandler只需读取消息头is_complete_group或is_partial_group即可判断分组类型，完全脱离发布策略逻辑。

方案2：用ErrorMessage封装超时分组

将超时触发的部分分组封装为ErrorMessage，完整分组保持普通消息格式：

@Override
protected void expireGroup(MessageGroup group) {
    if (isSendPartialResultOnExpiry()) {
        Message<?> partialResult = aggregatePayloads(group, new ArrayList<>(), null);
        MessagingException timeoutException = new MessagingException(partialResult, "Group expired before completion");
        ErrorMessage errorMessage = new ErrorMessage(timeoutException);
        sendOutputs(errorMessage);
        removeGroup(group);
    } else {
        super.expireGroup(group);
    }
}

下游通过判断消息类型是否为ErrorMessage即可区分：如果是ErrorMessage则为超时部分分组，直接做日志告警；否则为完整分组，执行业务逻辑。

二、实现超时消息专属通道/处理器的方案 #

方案1：通过消息路由分流

在聚合器输出后添加MessageRouter，根据消息头标记路由到不同通道：

@Bean
public IntegrationFlow aggregatorMainFlow() {
    return IntegrationFlows.from("petInputChannel")
            .aggregate(a -> a.handler(new CustomAggregatingMessageHandler(groupProcessor(), jdbcMessageStore()))
                    .sendPartialResultOnExpiry(true)
                    .groupTimeout(10000))
            .route(Message.class, msg -> {
                if (Boolean.TRUE.equals(msg.getHeaders().get("is_partial_group"))) {
                    return "partialGroupAlertChannel";
                } else {
                    return "completeGroupProcessChannel";
                }
            })
            .get();
}

// 超时分组专属处理流
@Bean
public IntegrationFlow partialGroupAlertFlow() {
    return IntegrationFlows.from("partialGroupAlertChannel")
            .handle(msg -> {
                // 执行日志记录、告警推送逻辑
                log.warn("Received partial group due to timeout: {}", msg.getPayload());
            })
            .get();
}

// 完整分组业务处理流
@Bean
public IntegrationFlow completeGroupProcessFlow() {
    return IntegrationFlows.from("completeGroupProcessChannel")
            .handle(new PetGroupHandler())
            .get();
}

方案2：自定义聚合器直接发送到专属通道

在expireGroup()方法中跳过默认输出，直接将超时分组发送到指定通道：

@Override
protected void expireGroup(MessageGroup group) {
    if (isSendPartialResultOnExpiry()) {
        Message<?> partialResult = aggregatePayloads(group, new ArrayList<>(), null);
        // 直接发送到超时专属通道
        getApplicationContext().getBean("partialGroupAlertChannel", MessageChannel.class).send(partialResult);
        removeGroup(group);
    } else {
        super.expireGroup(group);
    }
}

此方案无需经过聚合器默认输出通道，实现完全分流。

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