## 前言 最近很多小伙伴私信我，說自己很懊惱，對于 RxJava 2.x 系列一看就能明白，但自己寫卻又寫不出來。如果 LZ 能放上實戰情景教程就最好不過了。也是哈，單講我們的操作符，也讓我們的教程不溫不火，但 LZ 自己選擇的路，那跪著也要走完呀。所以，也就讓我可憐的小伙伴們忍忍了，操作符馬上就講完了。 ## 正題 #### Single 顧名思義，`Single` 只會接收一個參數，而 `SingleObserver` 只會調用 `onError()` 或者 `onSuccess()`。 ~~~ Single.just(new Random().nextInt()) .subscribe(new SingleObserver<Integer>() { @Override public void onSubscribe(@NonNull Disposable d) { } @Override public void onSuccess(@NonNull Integer integer) { mRxOperatorsText.append("single : onSuccess : "+integer+"\n"); Log.e(TAG, "single : onSuccess : "+integer+"\n" ); } @Override public void onError(@NonNull Throwable e) { mRxOperatorsText.append("single : onError : "+e.getMessage()+"\n"); Log.e(TAG, "single : onError : "+e.getMessage()+"\n"); } }); ~~~ 輸出：  #### [distinct](http://reactivex.io/documentation/operators/distinct.html) 去重操作符，簡單的作用就是去重。  ~~~ Observable.just(1, 1, 1, 2, 2, 3, 4, 5) .distinct() .subscribe(new Consumer<Integer>() { @Override public void accept(@NonNull Integer integer) throws Exception { mRxOperatorsText.append("distinct : " + integer + "\n"); Log.e(TAG, "distinct : " + integer + "\n"); } }); ~~~ 輸出：  很明顯，發射器發送的事件，在接收的時候被去重了。 #### [debounce](http://reactivex.io/documentation/operators/debounce.html) 去除發送頻率過快的項，看起來好像沒啥用處，但你信我，后面絕對有地方很有用武之地。  ~~~ Observable.create(new ObservableOnSubscribe<Integer>() { @Override public void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Exception { // send events with simulated time wait emitter.onNext(1); // skip Thread.sleep(400); emitter.onNext(2); // deliver Thread.sleep(505); emitter.onNext(3); // skip Thread.sleep(100); emitter.onNext(4); // deliver Thread.sleep(605); emitter.onNext(5); // deliver Thread.sleep(510); emitter.onComplete(); } }).debounce(500, TimeUnit.MILLISECONDS) .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(new Consumer<Integer>() { @Override public void accept(@NonNull Integer integer) throws Exception { mRxOperatorsText.append("debounce :" + integer + "\n"); Log.e(TAG,"debounce :" + integer + "\n"); } }); ~~~ 輸出：  代碼很清晰，去除發送間隔時間小于 500 毫秒的發射事件，所以 1 和 3 被去掉了。 #### [defer](http://reactivex.io/documentation/operators/defer.html) 簡單地時候就是每次訂閱都會創建一個新的 `Observable`，并且如果沒有被訂閱，就不會產生新的 `Observable`。  ~~~ Observable<Integer> observable = Observable.defer(new Callable<ObservableSource<Integer>>() { @Override public ObservableSource<Integer> call() throws Exception { return Observable.just(1, 2, 3); } }); observable.subscribe(new Observer<Integer>() { @Override public void onSubscribe(@NonNull Disposable d) { } @Override public void onNext(@NonNull Integer integer) { mRxOperatorsText.append("defer : " + integer + "\n"); Log.e(TAG, "defer : " + integer + "\n"); } @Override public void onError(@NonNull Throwable e) { mRxOperatorsText.append("defer : onError : " + e.getMessage() + "\n"); Log.e(TAG, "defer : onError : " + e.getMessage() + "\n"); } @Override public void onComplete() { mRxOperatorsText.append("defer : onComplete\n"); Log.e(TAG, "defer : onComplete\n"); } }); ~~~ 輸出：  #### [last](http://reactivex.io/documentation/operators/last.html) `last` 操作符僅取出可觀察到的最后一個值，或者是滿足某些條件的最后一項。  ~~~ Observable.just(1, 2, 3) .last(4) .subscribe(new Consumer<Integer>() { @Override public void accept(@NonNull Integer integer) throws Exception { mRxOperatorsText.append("last : " + integer + "\n"); Log.e(TAG, "last : " + integer + "\n"); } }); ~~~ 輸出：  #### [merge](http://reactivex.io/documentation/operators/merge.html) `merge` 顧名思義，熟悉版本控制工具的你一定不會不知道 merge 命令，而在 Rx 操作符中，`merge` 的作用是把多個 `Observable` 結合起來，接受可變參數，也支持迭代器集合。注意它和 `concat` 的區別在于，不用等到 發射器 A 發送完所有的事件再進行發射器 B 的發送。  ~~~ Observable.merge(Observable.just(1, 2), Observable.just(3, 4, 5)) .subscribe(new Consumer<Integer>() { @Override public void accept(@NonNull Integer integer) throws Exception { mRxOperatorsText.append("merge :" + integer + "\n"); Log.e(TAG, "accept: merge :" + integer + "\n" ); } }); ~~~ 輸出：  #### [reduce](http://reactivex.io/documentation/operators/reduce.html) `reduce` 操作符每次用一個方法處理一個值，可以有一個 `seed` 作為初始值。  ~~~ Observable.just(1, 2, 3) .reduce(new BiFunction<Integer, Integer, Integer>() { @Override public Integer apply(@NonNull Integer integer, @NonNull Integer integer2) throws Exception { return integer + integer2; } }).subscribe(new Consumer<Integer>() { @Override public void accept(@NonNull Integer integer) throws Exception { mRxOperatorsText.append("reduce : " + integer + "\n"); Log.e(TAG, "accept: reduce : " + integer + "\n"); } }); ~~~ 輸出：  可以看到，代碼中，我們中間采用 reduce ，支持一個 function 為兩數值相加，所以應該最后的值是：1 + 2 = 3 + 3 = 6 ， 而Log 日志完美解決了我們的問題。 #### [scan](http://reactivex.io/documentation/operators/scan.html) `scan` 操作符作用和上面的 `reduce` 一致，唯一區別是 `reduce` 是個只追求結果的壞人，而 `scan` 會始終如一地把每一個步驟都輸出。  ~~~ Observable.just(1, 2, 3) .scan(new BiFunction<Integer, Integer, Integer>() { @Override public Integer apply(@NonNull Integer integer, @NonNull Integer integer2) throws Exception { return integer + integer2; } }).subscribe(new Consumer<Integer>() { @Override public void accept(@NonNull Integer integer) throws Exception { mRxOperatorsText.append("scan " + integer + "\n"); Log.e(TAG, "accept: scan " + integer + "\n"); } }); ~~~ 輸出：  看日志，沒毛病。 #### [window](http://reactivex.io/documentation/operators/window.html) 按照實際劃分窗口，將數據發送給不同的 `Observable`  ~~~ mRxOperatorsText.append("window\n"); Log.e(TAG, "window\n"); Observable.interval(1, TimeUnit.SECONDS) // 間隔一秒發一次 .take(15) // 最多接收15個 .window(3, TimeUnit.SECONDS) .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(new Consumer<Observable<Long>>() { @Override public void accept(@NonNull Observable<Long> longObservable) throws Exception { mRxOperatorsText.append("Sub Divide begin...\n"); Log.e(TAG, "Sub Divide begin...\n"); longObservable.subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(new Consumer<Long>() { @Override public void accept(@NonNull Long aLong) throws Exception { mRxOperatorsText.append("Next:" + aLong + "\n"); Log.e(TAG, "Next:" + aLong + "\n"); } }); } }); ~~~ 輸出：  ## 寫在最后 至此，大部分 RxJava 2.x 的操作符就告一段落了，當然還有一些沒有提到的操作符，不是說它們不重要，而是 LZ 也要考慮大家的情況，接下來就會根據實際應用場景來對 RxJava 2.x 發起沖鋒。如果想看更多的數據，請移步 GitHub：[https://github.com/nanchen2251/RxJava2Examples](https://github.com/nanchen2251/RxJava2Examples) 作者：南塵2251 鏈接：http://www.jianshu.com/p/c08bfc58f4b6 來源：簡書 著作權歸作者所有。商業轉載請聯系作者獲得授權，非商業轉載請注明出處。