一、前言

最近很多小伙伴私信我，说自己很懊恼，对于RxJava 2.x 系列一看就能明白，但自己写却又写不出来。如果 LZ 能放上实战情景教程就最好不过了。也是哈，单讲我们的操作符，也让我们的教程不温不火，但 LZ 自己选择的路，那跪着也要走完呀。所以，也就让我可怜的小伙伴们忍忍了，操作符马上就讲完了。

二、正题

16、Single

顾名思义，Single 只会接收一个参数，而SingleObserver 只会调用onError 或者onSuccess。

Single.just(new Random().nextInt())
                .subscribe(new SingleObserver() {                    @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");
                    }
                });

输出：

17、distinct

去重操作符，简单的作用就是去重。

Observable.just(1, 1, 1, 2, 2, 3, 4, 5)
                .distinct()
                .subscribe(new Consumer() {                    @Override
                    public void accept(@NonNull Integer integer) throws Exception {
                        mRxOperatorsText.append("distinct : " + integer + "n");
                        Log.e(TAG, "distinct : " + integer + "n");
                    }
                });

输出：

很明显，发射器发送的事件，在接收的时候被去重了。

18、debounce

去除发送频率过快的项，看起来好像没啥用处，但你信我，后面绝对有地方很有用武之地。

Observable.create(new ObservableOnSubscribe() {            @Override
            public void subscribe(@NonNull ObservableEmitter 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() {                    @Override
                    public void accept(@NonNull Integer integer) throws Exception {
                        mRxOperatorsText.append("debounce :" + integer + "n");
                        Log.e(TAG,"debounce :" + integer + "n");
                    }
                });

输出：

代码很清晰，去除发送间隔时间小于500毫秒的发射事件，所以1 和 3 被去掉了。

19、defer

简单地时候就是每次订阅都会创建一个新的Observable，并且如果没有被订阅，就不会产生新的Observable

Observable observable = Observable.defer(new Callable>() {            @Override
            public ObservableSource call() throws Exception {                return Observable.just(1, 2, 3);
            }
        });


        observable.subscribe(new Observer() {            @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 : onCompleten");
                Log.e(TAG, "defer : onCompleten");
            }
        });

输出：

20、last

last 操作符仅取出可观察到的最后一个值，或者是满足某些条件的最后一项。

Observable.just(1, 2, 3)
                .last(4)
                .subscribe(new Consumer() {                    @Override
                    public void accept(@NonNull Integer integer) throws Exception {
                        mRxOperatorsText.append("last : " + integer + "n");
                        Log.e(TAG, "last : " + integer + "n");
                    }
                });

输出：

21、merge

merge 顾名思义，熟悉版本控制工具的你一定不会不知道 merge 命令，而在 Rx 操作符中，merge 的作用是把多个 Observable 结合起来，接受可变参数，也支持迭代器集合。注意它和 concat 的区别在于，不用等到发射器 A 发送完所有的事件再进行发射器 B 的发送。

Observable.merge(Observable.just(1, 2), Observable.just(3, 4, 5))
                .subscribe(new Consumer() {                    @Override
                    public void accept(@NonNull Integer integer) throws Exception {
                        mRxOperatorsText.append("merge :" + integer + "n");
                        Log.e(TAG, "accept: merge :" + integer + "n" );
                    }
                });

输出：

22、reduce

reduce 操作符每次用一个方法处理一个值，可以有一个 seed 作为初始值。

Observable.just(1, 2, 3)
               .reduce(new BiFunction() {                   @Override
                   public Integer apply(@NonNull Integer integer, @NonNull Integer integer2) throws Exception {                       return integer + integer2;
                   }
               }).subscribe(new Consumer() {           @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 日志完美解决了我们的问题。

23、scan

scan 操作符作用和上面的 reduce 一致，唯一区别是 reduce 是个只追求结果的坏人，而 scan 会始终如一地把每一个步骤都输出。

Observable.just(1, 2, 3)
                .scan(new BiFunction() {                    @Override
                    public Integer apply(@NonNull Integer integer, @NonNull Integer integer2) throws Exception {                        return integer + integer2;
                    }
                }).subscribe(new Consumer() {            @Override
            public void accept(@NonNull Integer integer) throws Exception {
                mRxOperatorsText.append("scan " + integer + "n");
                Log.e(TAG, "accept: scan " + integer + "n");
            }
        });

输出：

看日志，没毛病。

24、window

按照实际划分窗口，将数据发送给不同的Observable

mRxOperatorsText.append("windown");
       Log.e(TAG, "windown");
       Observable.interval(1, TimeUnit.SECONDS) // 间隔一秒发一次
               .take(15) // 最多接收15个
               .window(3, TimeUnit.SECONDS)
               .subscribeOn(Schedulers.io())
               .observeOn(AndroidSchedulers.mainThread())
               .subscribe(new Consumer>() {                   @Override
                   public void accept(@NonNull Observable 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() {                                   @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

原文链接：http://www.apkbus.com/blog-898535-68136.html

ITPub博客

这可能是最好的RxJava 2.x 入门教程（四）