使用 Rx 运行直方图流答案

【问题标题】：Running histogram stream with Rx使用 Rx 运行直方图流
【发布时间】：2015-09-29 09:28:51
【问题描述】：

我有以下单字母流

A 
B 
C 
A
D 
B 
A 
C 
D

从这个流中，我想要一个每个字母的运行计数流

(A,1)
(A,1), (B,1)
(A,1), (B,1), (C,1)
(A,2), (B,1), (C,1)
(A,2), (B,1), (C,1), (D,1)
(A,2), (B,2), (C,1), (D,1)
(A,3), (B,2), (C,1), (D,1)     
(A,3), (B,2), (C,2), (D,1) 
(A,3), (B,2), (C,2), (D,2)

，即在每个新字母处，更新和发出总数。

我想这个问题与语言无关，所以不要犹豫，用您选择的语言提出解决方案。

【问题讨论】：

标签： system.reactive reactive-programming rx-java rxjs

【解决方案1】：

这就是使用RxJava 的方法：

final Observable<String> observable = Observable.just("A", "B", "C", "A", "D", "B", "A", "C", "D");
final Observable<LinkedHashMap<String, Integer>> histogram = observable.scan(new LinkedHashMap<>(), (state, value) -> {
  if (state.containsKey(value)) {
    state.put(value, state.get(value) + 1);
  } else {
    state.put(value, 1);
  }

  return state;
});

histogram.subscribe(state -> {
  System.out.println(state);
});

输出：

{}
{A=1}
{A=1, B=1}
{A=1, B=1, C=1}
{A=2, B=1, C=1}
{A=2, B=1, C=1, D=1}
{A=2, B=2, C=1, D=1}
{A=3, B=2, C=1, D=1}
{A=3, B=2, C=2, D=1}
{A=3, B=2, C=2, D=2}

【讨论】：

【解决方案2】：

在RxJS 中可能是这样的：

var letters   = Rx.Observable.of('A', 'B', 'C', 'A', 'D', 'B', 'A', 'C', 'D'),
    histogram = letters.scan(countL, Object.create(null));

histogram.subscribe(console.log.bind(console));

function countL(ls, l) {
    if (!ls[l]) ls[l] = 0;
    ls[l]++;
    return ls;
}

【讨论】：

或使用immutable map，如果您不希望下游观察者无意中将其结果从其下方更改出来。