通过键将GroupFlux收集到Hashmap中

Question

我最近开始使用Project Reactor，我想出了一个我似乎无法弄清楚的情景。

基本上我想分组某个流，然后获取哈希映射，如grouping key -> List of grouped values。我一直在玩API，但我得到的最远的是获取值，键或计数，而不是我想要的数据结构。这将是代码，例如，获取值：

var elements = new ArrayList<Integer>();

Flux.just(-1, -2, -3, 1, 2, 3)
        .groupBy(val -> val.compareTo(0))
        .flatMap(Flux::collectList)
        .subscribe(elements::addAll);

我想通过的测试如下：

@Test
public void groupBy() {
    var elements = new HashMap<Integer, List<Integer>>();

    Flux.just(-1, -2, -3, 1, 2, 3)
            .groupBy(val -> val.compareTo(0))
            // Do something here ...
            .subscribe(...);

    assertThat(elements).containsKeys(-1, 1);
    assertThat(elements.get(-1)).containsExactly(-1, -2, -3);
    assertThat(elements.get(1)).containsExactly(1, 2, 3);
}

我怎么能实现后者呢？

Answer 1

你考虑过使用Flux#connect吗？它接受Collector，与Stream使用的类型相同。还有Flux#collectMap。

另外，如果您需要流式传输此类地图，您可以使用Flux#scan。

当你需要按键“路由”你的信号并按键有一个groupBy时，Flux很有帮助，但它并不是用来创建数据集合的。