Rx는 GROUP BY의 COUNT와 (과) 동일합니까?

Question

나는 말을 반복 요소의 목록을 가지고 :

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에서 가능합니까? 원하는 결과를 얻을 수있는 다른 방법이 있습니까?

Answer 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 

Answer 2

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; 
    } 
} 

Answer 3

또 다른 방법은 collect 또는 collectInto 방법을 사용하는 것입니다. 그룹의 많은 수의 경우, GROUPBY 후 flatMap이 중단됩니다 때문에 그런데

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이 그룹의 낮은 기수에서 가장 잘 작동하는 javadoc of Reactor

주에서

, 그래서 그에 따라 keyMapper 기능을 선택했다.
...
특히 기준이 많은 양의 그룹을 생성하면 그룹이 적절하게 다운 스트림으로 소비되지 않는 경우 (예 : 너무 낮게 설정된 maxConcurrency 매개 변수가있는 flatMap로 인해) 걸려 넘어 질 수 있습니다.

이와 관련된 github issue도 있습니다.