【知识整理】这可能是最好的RxJava 2.x 入门教程（四）

阅读量：4708 次

发布时间：2019-06-10

本文共 8501 字，大约阅读时间需要 28 分钟。

这可能是最好的RxJava 2.x入门教程系列专栏

文章链接：

GitHub 代码同步更新：

为了满足大家的饥渴难耐，GitHub将同步更新代码，主要包含基本的代码封装，RxJava 2.x所有操作符应用场景介绍和实际应用场景，后期除了RxJava可能还会增添其他东西，总之，GitHub上的Demo专为大家倾心打造。传送门：

一、前言

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

二、正题

16、Single

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

1 Single.just(new Random().nextInt()) 2                 .subscribe(new SingleObserver
     
      () { 3                     @Override 4                     public void onSubscribe(@NonNull Disposable d) { 5  6                     } 7  8                     @Override 9                     public void onSuccess(@NonNull Integer integer) {10                         mRxOperatorsText.append("single : onSuccess : "+integer+"\n");11                         Log.e(TAG, "single : onSuccess : "+integer+"\n" );12                     }13 14                     @Override15                     public void onError(@NonNull Throwable e) {16                         mRxOperatorsText.append("single : onError : "+e.getMessage()+"\n");17                         Log.e(TAG, "single : onError : "+e.getMessage()+"\n");18                     }19                 });

输出：

17、

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

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

输出：

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

18、

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

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、

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

1 Observable
     
       observable = Observable.defer(new Callable
      
       
        >() { 2             @Override 3             public ObservableSource
        
          call() throws Exception { 4                 return Observable.just(1, 2, 3); 5             } 6         }); 7  8  9         observable.subscribe(new Observer
         
          () {10             @Override11             public void onSubscribe(@NonNull Disposable d) {12 13             }14 15             @Override16             public void onNext(@NonNull Integer integer) {17                 mRxOperatorsText.append("defer : " + integer + "\n");18                 Log.e(TAG, "defer : " + integer + "\n");19             }20 21             @Override22             public void onError(@NonNull Throwable e) {23                 mRxOperatorsText.append("defer : onError : " + e.getMessage() + "\n");24                 Log.e(TAG, "defer : onError : " + e.getMessage() + "\n");25             }26 27             @Override28             public void onComplete() {29                 mRxOperatorsText.append("defer : onComplete\n");30                 Log.e(TAG, "defer : onComplete\n");31             }32         });

输出：

20、

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

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

输出：

21、

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

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

输出：

22、

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

1 Observable.just(1, 2, 3) 2                 .reduce(new BiFunction
     
      () { 3                     @Override 4                     public Integer apply(@NonNull Integer integer, @NonNull Integer integer2) throws Exception { 5                         return integer + integer2; 6                     } 7                 }).subscribe(new Consumer
      
       () { 8             @Override 9             public void accept(@NonNull Integer integer) throws Exception {10                 mRxOperatorsText.append("reduce : " + integer + "\n");11                 Log.e(TAG, "accept: reduce : " + integer + "\n");12             }13         });

输出：

可以看到，代码中，我们中间采用 reduce ，支持一个 function 为两数值相加，所以应该最后的值是：1 + 2 = 3 + 3 = 6 ，而Log 日志完美解决了我们的问题。

23、

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

1 Observable.just(1, 2, 3) 2                 .scan(new BiFunction
     
      () { 3                     @Override 4                     public Integer apply(@NonNull Integer integer, @NonNull Integer integer2) throws Exception { 5                         return integer + integer2; 6                     } 7                 }).subscribe(new Consumer
      
       () { 8             @Override 9             public void accept(@NonNull Integer integer) throws Exception {10                 mRxOperatorsText.append("scan " + integer + "\n");11                 Log.e(TAG, "accept: scan " + integer + "\n");12             }13         });

输出：

看日志，没毛病。

24、

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

1 mRxOperatorsText.append("window\n"); 2         Log.e(TAG, "window\n"); 3         Observable.interval(1, TimeUnit.SECONDS) // 间隔一秒发一次 4                 .take(15) // 最多接收15个 5                 .window(3, TimeUnit.SECONDS) 6                 .subscribeOn(Schedulers.io()) 7                 .observeOn(AndroidSchedulers.mainThread()) 8                 .subscribe(new Consumer
     
      
       >() { 9                     @Override10                     public void accept(@NonNull Observable
       
         longObservable) throws Exception {11                         mRxOperatorsText.append("Sub Divide begin...\n");12                         Log.e(TAG, "Sub Divide begin...\n");13                         longObservable.subscribeOn(Schedulers.io())14                                 .observeOn(AndroidSchedulers.mainThread())15                                 .subscribe(new Consumer
        
         () {16                                     @Override17                                     public void accept(@NonNull Long aLong) throws Exception {18                                         mRxOperatorsText.append("Next:" + aLong + "\n");19                                         Log.e(TAG, "Next:" + aLong + "\n");20                                     }21                                 });22                     }23                 });

输出：