Rx 相当于 COUNT 和 GROUP BY？答案

【问题标题】：Rx equivalent of COUNT with GROUP BY?Rx 相当于 COUNT 和 GROUP BY？
【发布时间】：2016-09-15 01:03:35
【问题描述】：

我有一个重复元素的列表，比如说：

Observable<String> source = Observable.just("A", "B", "A", "C", "C", "A");

我想按它们的值以及它们出现的次数对它们进行分组，因此输出将是成对的：

{"A", 3}, {"B", 1}, {"C", 2}

基本上相当于SELECT x, COUNT(1) GROUP BY x;这样的SQL语句

我只能在他们身上调用 groupBy：

source.groupBy(x -> x, x -> 1)

但这会将流转换为 GroupedObservables，我找不到如何继续使用它们的好例子。我试过reduce()，但这里不好，因为在groupBy()之后它想减少GroupedObservables，而不是每个组内的元素。

GroupedObservables 可以做到这一点吗？有没有其他方法可以达到预期的效果？

【问题讨论】：

您是否明白，如果您有一个 hot observable 或一个没有 OnComplete 信号的您无法进行此计算？
不，我没有想到，但似乎可以理解，谢谢。实际上，我将列表作为一个整体发出，而不是逐个元素发出（我只是简化了问题的代码），所以我想这不是问题，对吧？
是的，但是如果它是一个冷的 observable，它会发出它的所有值，你可能不需要 Rx。你这样做是为了什么？

标签： rx-java reactive-programming

【解决方案1】：

以下代码：

source.groupBy(val -> val)
    .flatMap(
        gr -> gr.count()
                .map(count -> new Pair<>(gr.getKey(), count)
    )
).subscribe(System.out::println);

会打印出来：

A=3
B=1
C=2

【讨论】：

【解决方案2】：

另一种方法是使用collect或collectInto方法，如下所示。

Observable<String> source = Observable.just("A", "B", "A", "C", "C", "A");

source.collectInto(new HashMap<String, MutableInt>(), (map, elem) -> {
    if (map.containsKey(elem)) {
    map.get(elem).increment();
    } else {
    map.put(elem, new MutableInt(1));
    }

}).subscribe(System.out::println);

顺便说一句，如果我们使用Reactor，collect 是这样做的建议方式，因为在大量组的情况下，groupBy 之后的 flatMap 会挂起。

来自javadoc of Reactor

请注意，groupBy 最适用于基数较低的组，因此请相应地选择您的 keyMapper 函数。
...
值得注意的是，当标准产生大量组时，如果这些组没有在下游适当地消耗（例如，由于 flatMap 的 maxConcurrency 参数设置得太低），可能会导致挂起。

还有一个与此相关的githubissue。

【讨论】：

【解决方案3】：

    Observable<String> source = Observable.just("A", "B", "A", "C", "C", "A");
    Observable<KeyValue<String, Integer>> countStream = source
            .groupBy(val -> val)
            .flatMap(obs -> obs.count().flatMap(cnt -> Observable.just(new KeyValue<>(obs.getKey(), cnt))));

  private static class KeyValue<K, V> {

    private final K key;
    private final V val;

    public KeyValue(K key, V val) {
        this.key = key;
        this.val = val;
    }

    public K getKey() {
        return key;
    }

    public V getVal() {
        return val;
    }
}

【讨论】：