分类器的Spring集成组消息

Question

使用Java 8 streams，我可以通过分类器对消息进行分组：

Map<String, List<String>> grouped = Arrays.asList("a", "b", "b", "b", "c")
        .stream()
        .collect(Collectors.groupingBy(Function.identity()));

我想写一个聚合器，将消息进行相应的分组。在如上所示的五条有效载荷消息中，我想生成三条消息：第一条消息应将“a”作为有效载荷，第二条消息应将三条“b”作为有效载荷，第三条消息应将“c”作为有效载荷。

当达到序列大小时，应释放所有消息组。基于有效负载的分组工作正常，但消息组永远不会被释放。

在发布策略中，我可以访问序列大小，但我无法找到已处理的项目总数。如何发布分组邮件？

public interface StringGrouper {
    List<Message<?> groupSame(List<String> toGroup);
}   

@Bean
public IntegrationFlow groupStringsFlow() {
    return IntegrationFlows.from(StringGrouper.class)
        .split()
        .aggregate(agg -> agg
            .correlationStrategy(message -> message.getPayload())
            .releaseStrategy(group -> group.getSequenceSize() == /* what? */)) 
        .logAndReply();
}

@Test
public void shouldGroupMessages {
    List<Message<?> grouper
        .groupSame(Arrays.asList("a", "b", "b", "b", "c"));
}

解决方法是根本不使用聚合器，而是将传入列表分组到转换器中。但我希望我可以使用聚合器来实现这一点。

@Bean
public IntegrationFlow groupStringsFlow() {
    return IntegrationFlows.from(StringGrouper.class)
        .<List<String>, Collection<List<String>>>transform(source -> source.stream()
            .collect(Collectors.collectingAndThen(
                Collectors.groupingBy(Function.identity()), 
                grouped -> grouped.values())))
        .split()
        .log() // work with messages
        .aggregate()
        .get();
}

Answer 1

使用默认的序列大小释放策略将它们聚合为单个组，并使用自定义输出处理器（MessageGroupProcessor）在有效负载上重新分组，返回集合