具有并行流的散点聚集（聚合器中的超时）

Question

我一直在尝试在聚集中添加超时，以避免等待每个流都完成。但是当我添加超时时，它不起作用，因为聚合器等待每个流完成。

@Bean
public IntegrationFlow queueFlow(LogicService service) {    
        return f -> f.scatterGather(scatterer -> scatterer
                                .applySequence(true)
                                .recipientFlow(aFlow(service))
                                .recipientFlow(bFlow(service))
                        , aggregatorSpec -> aggregatorSpec.groupTimeout(2000L)) 

E、 在我的流中，其中一个有2秒的延迟，另一个有4秒的延迟

public IntegrationFlow bFlow(LogicService service) {
        return IntegrationFlows.from(MessageChannels.executor(Executors.newCachedThreadPool()))
                .handle(service::callFakeServiceTimeout2)
                .transform((MessageDomain.class), message -> {
                    message.setMessage(message.getMessage().toUpperCase());
                    return message;
                }).get();
    } 

我使用遗嘱执行人。newCachedThreadPool（）以并行运行。我想释放包含的每条消息，直到超时完成

我一直在测试的另一种方法是使用默认的gatherer，并在scatterGather中设置GathereTimeout，但我不知道我是否缺少一些方法GathereTimeout

更新

评论中给出的所有方法都经过测试并正常工作，唯一的问题是每个操作都在消息组创建过程中进行评估。并且消息组是在第一条消息到达之前创建的。理想的方法是在散射器分发请求消息时有一个有效的选项。

我的临时解决方案是使用一个发布策略，应用一个分组条件提供程序，它读取我通过网关发送消息时创建的自定义标头。唯一担心的是，只有在到达新消息或我设置了组超时时，才会执行发布策略。

Answer 1

聚合器上的组超时不足以释放组。如果你没有在那个超时时间得到整个组，那么它将被丢弃。请参阅sendPartialResultOnExpiry选项：https://docs.spring.io/spring-integration/reference/html/message-routing.html#agg-并分组到

如果到期时发送部分结果为true，则（部分）消息组中的现有消息将作为正常聚合器回复消息发布到输出通道。否则，它将被丢弃。

如果您根本不希望gatherer回复，那么GathereTimeout是一个不错的选择。因此，这样您就不会永远阻止散聚线程：https://docs.spring.io/spring-integration/reference/html/message-routing.html#scatter-收集错误处理