RxJava2.0 操作符之 -- 算数操作符

rxjava-math 模块的操作符 此类库只有 1.x 的支持，2.0 代码真没法写

averageInteger( ) — 求序列平均数并发射

averageLong( ) — 求序列平均数并发射

averageFloat( ) — 求序列平均数并发射

averageDouble( ) — 求序列平均数并发射

max( ) — 求序列最大值并发射

maxBy( ) — 求最大 key 对应的值并发射

min( ) — 求最小值并发射

minBy( ) — 求最小 Key 对应的值并发射

sumInteger( ) — 求和并发射

sumLong( ) — 求和并发射

sumFloat( ) — 求和并发射

sumDouble( ) — 求和并发射

Concat

不交错的发射两个或多个 Observable 的发射物

	Observable observable1 = Observable.just(1, 2, 3, 4);
  Observable observable2 = Observable.just(5, 6, 7, 8);
  Observable.concat(observable1,observable2).subscribe(RxUtils.getObserver());

	  onSubscribe
	Thread:Thread[main,5,main]
	onNext:1
	Thread:Thread[main,5,main]
	onNext:2
	Thread:Thread[main,5,main]
	onNext:3
	Thread:Thread[main,5,main]
	onNext:4
	Thread:Thread[main,5,main]
	onNext:5
	Thread:Thread[main,5,main]
	onNext:6
	Thread:Thread[main,5,main]
	onNext:7
	Thread:Thread[main,5,main]
	onNext:8
	Thread:Thread[main,5,main]
	onComplete

Reduce

按顺序对 Observable 发射的每项数据应用一个函数并发射最终的值

Observable.just(1, 2, 3, 4, 5).reduce(new BiFunction, Integer, Integer>() {
    public Integer apply(@NonNull Integer integer, @NonNull Integer integer2) throws Exception {
        return integer + integer2;
  }
}).toObservable().subscribe(RxUtils.getObserver());

	onSubscribe
	Thread:Thread[main,5,main]
	onNext:15
	Thread:Thread[main,5,main]
	onComplete
	Thread:Thread[main,5,main]

Count

计算数据项的个数并发射结果

	Observable.just(1,2,3,4)
			.count().toObservable()
			.subscribe(RxUtils.getObserver());

	onSubscribe
	Thread:Thread[main,5,main]
	onNext:4
	Thread:Thread[main,5,main]
	onComplete
	Thread:Thread[main,5,main]

collect

Collect 操作符类似于 Reduce，但是其目的不同，collect 用来将源 Observable 发射的数据给收集到一个数据结构里面，需要使用两个参数：

• 一个产生收集数据结构的函数。
• 一个接收第一个函数产生的数据结构和源 Observable 发射的数据作为参数的函数。

List list=new ArrayList();
list.add(1);
list.add(2);
list.add(3);
list.add(4);
list.add(5);
Observable.fromIterable(list).collect(new Callable>() {
    public ArrayList call() throws Exception {
        return new ArrayList();
  }
}, new BiConsumer, Integer>() {
    public void accept(ArrayList integers, Integer integer) throws Exception {
        integers.add(integer);
  }
}).toObservable().subscribe(RxUtils.>getObserver());

  onSubscribe
  Thread:Thread[main,5,main]
  onNext:[1, 2, 3, 4, 5]
  Thread:Thread[main,5,main]
  onComplete
  Thread:Thread[main,5,main]