聚合后缺少最后一组

Question

我遇到了聚合后丢失一些消息的问题。我需要将我的消息聚合到具有相同数量元素的组中。对于我当前的问题，我在messageChannel 中有 118 个元素。这些消息成功地组合成 11 个组，每个组有 10 个元素。但是最后8个已经丢失了

IntegrationFlows
                .from(messageChannel)
                .split(s -> s
                        .applySequence(false).get().getT2().setDelimiters("[\r\n]"))
                .aggregate(s -> s
                                .correlationExpression("payload")
                                .releaseExpression("size() >= 10")
                                .expireGroupsUponCompletion(true)
                )
                .handle(h ->
                        System.out.println(h))

                .get();

我希望在新组中收到 8 条丢失的消息

Answer 1

也许expireGroupsUponTimeout 可以在这里提供帮助：

IntegrationFlows
            .from(messageChannel)
            .split(s -> s
                    .applySequence(false).get().getT2().setDelimiters("[\r\n]"))
            .aggregate(s -> s
                            .correlationExpression("payload")
                            .releaseExpression("size() >= 10")
                            .expireGroupsUponCompletion(true)
                            .expireGroupsUponTimeout( 500 )
            )
            .handle(h ->
                    System.out.println(h))

            .get();