关于RxJava最友好的文章

作者: 拉丁吴 | 来源:发表于2016-10-20 21:20 被阅读4422次

本文首发于掘金-关于RxJava最友好的文章

RxJava到底是什么?让我们直接跳过官方那种晦涩的追求精确的定义,其实初学RxJava只要把握两点:观察者模式异步,就基本可以熟练使用RxJava了。

异步在这里并不需要做太多的解释,因为在概念和使用上,并没有太多高深的东西。大概就是你脑子里想能到的那些多线程,线程切换这些东西。我会在后面会讲解它的用法。

我们先把观察者模式说清楚

“按下开关,台灯灯亮”

在这个事件中,台灯作为观察者,开关作为被观察者,台灯透过电线来观察开关的状态来并做出相应的处理

观察上图,其实已经很明了了,不过需要指出一下几点(对于下面理解RxJava很重要):

  • 开关(被观察者)作为事件的产生方(生产“开”和“关”这两个事件),是主动的,是整个开灯事理流程的起点
  • 台灯(观察者)作为事件的处理方(处理“灯亮”和“灯灭”这两个事件),是被动的,是整个开灯事件流程的终点
  • 在起点和终点之间,即事件传递的过程中是可以被加工,过滤,转换,合并等等方式处理的(上图没有体现,后面对会讲到)。

我必须苦口婆心的告诉你:我们总结的这三点对于我们理解RxJava非常重要。因为上述三条分别对应了RxJava中被观察者(Observable),观察者(Observer)和操作符的职能。而观察者模式又是RxJava程序运行的骨架

好了,我假设你已经完全理解了我上面讲述的东西。我们正式进入RxJava!

RxJava也是基于观察者模式来组建自己的程序逻辑的,就是构建被观察者(Observable),观察者(Observer/Subscriber),然后建立二者的订阅关系(就像那根电线,连接起台灯和开关)实现观察,在事件传递过程中还可以对事件做各种处理

Tips: Observer是观察者的接口, Subscriber是实现这个接口的抽象类,因此两个类都可以被当做观察者,由于Subscriber在Observe的基础上做了一些拓展,加入了新的方法,一般会更加倾向于使用Subscriber。


创建被观察者

  • 正常模式:
 Observable switcher=Observable.create(new Observable.OnSubscribe<String>(){

            @Override
            public void call(Subscriber<? super String> subscriber) {
                subscriber.onNext("On");
                subscriber.onNext("Off");
                subscriber.onNext("On");
                subscriber.onNext("On");
                subscriber.onCompleted();
            }
        });
  

这是最正宗的写法,创建了一个开关类,产生了五个事件,分别是:开,关,开,开,结束。

  • 偷懒模式1
Observable switcher=Observable.just("On","Off","On","On");

  • 偷懒模式2
String [] kk={"On","Off","On","On"};
Observable switcher=Observable.from(kk);  

偷懒模式是一种简便的写法,实际上也都是被观察者把那些信息"On","Off","On","On",包装成onNext("On")这样的事件依次发给观察者,当然,它自己补上了onComplete()事件。

以上是最常用到的创建方式,好了,我们就创建了一个开关类。


创建观察者

  • 正常模式
 Subscriber light=new Subscriber<String>() {
            @Override
            public void onCompleted() {
                //被观察者的onCompleted()事件会走到这里;
                Log.d("DDDDDD","结束观察...\n");
            }

            @Override
            public void onError(Throwable e) {
                    //出现错误会调用这个方法
            }
            @Override
            public void onNext(String s) {
                //处理传过来的onNext事件
                Log.d("DDDDD","handle this---"+s)
            }

这也是比较常见的写法,创建了一个台灯类。

  • 偷懒模式(非正式写法)
        Action1 light=new Action1<String>() {
                @Override
                public void call(String s) {
                    Log.d("DDDDD","handle this---"+s)
                }
            }

之所以说它是非正式写法,是因为Action1是一个单纯的人畜无害的接口,和Observer没有啥关系,只不过它可以当做观察者来使,专门处理onNext 事件,这是一种为了简便偷懒的写法。当然还有Action0,Action2,Action3...,0,1,2,3分别表示call()这个方法能接受几个参数。如果你还不懂,可以暂时跳过。后面我也会尽量使用new Subscriber方式,创建正统的观察者,便于你们理解。

订阅

现在已经创建了观察者和被观察者,但是两者还没有联系起来

switcher.subscribe(light);

我猜你看到这里应该有疑问了,为什么是开关订阅了台灯?应该是台灯订阅了开关才对啊。卧槽,到底谁观察谁啊!!

大家冷静,把刀放下,有话慢慢说,

是这样的,台灯观察开关,逻辑是没错的,而且正常来看就应该是light.subscribe(switcher)才对,之所以“开关订阅台灯”,是为了保证流式API调用风格

啥是流式API调用风格

//这就是RxJava的流式API调用
Observable.just("On","Off","On","On")
        //在传递过程中对事件进行过滤操作
         .filter(new Func1<String, Boolean>() {
                    @Override
                    public Boolean call(String s) {
                        return s!=null;
                    }
                })
        .subscribe(mSubscriber);

上面就是一个非常简易的RxJava流式API的调用:同一个调用主体一路调用下来,一气呵成。

由于被观察者产生事件,是事件的起点,那么开头就是用Observable这个主体调用来创建被观察者,产生事件,为了保证流式API调用规则,就直接让Observable作为唯一的调用主体,一路调用下去。

一句话,背后的真实的逻辑依然是台灯订阅了开关,但是在表面上,我们让开关“假装”订阅了台灯,以便于保持流式API调用风格不变。

好了,现在分解动作都完成了,已经架构了一个基本的RxJava事件处理流程。

我们再来按照观察者模式的运作流程和流式Api的写法复习一遍:

流程图如下:

结合流程图的相应代码实例如下:

//创建被观察者,是事件传递的起点
Observable.just("On","Off","On","On")
        //这就是在传递过程中对事件进行过滤操作
         .filter(new Func1<String, Boolean>() {
                    @Override
                    public Boolean call(String s) {
                        return s!=null;
                    }
                })
        //实现订阅
        .subscribe(
                //创建观察者,作为事件传递的终点处理事件    
                  new Subscriber<String>() {
                        @Override
                        public void onCompleted() {
                            Log.d("DDDDDD","结束观察...\n");
                        }
            
                        @Override
                        public void onError(Throwable e) {
                            //出现错误会调用这个方法
                        }
                        @Override
                        public void onNext(String s) {
                            //处理事件
                            Log.d("DDDDD","handle this---"+s)
                        }
        );

嗯,基本上我们就把RxJava的骨架就讲完了,总结一下:

  • 创建被观察者,产生事件
  • 设置事件传递过程中的过滤,合并,变换等加工操作。
  • 订阅一个观察者对象,实现事件最终的处理。

Tips: 当调用订阅操作(即调用Observable.subscribe()方法)的时候,被观察者才真正开始发出事件。


现在开始讲异步操作?别着急,事件的产生起点和处理的终点我们都比较详细的讲解了,接下来我们好好讲讲事件传递过程中发生的那些事儿...

RxJava的操作符

即使你已经看了我上面那段讲解,Rxjava可能还打动不了你,没关系,事件产生的起点和消费的终点其实没那么吸引人,真正有意思的是事件传递过程中的那些鬼斧神工的操作。

由于篇幅的限制,我只讲两三个操作,其他的操作请看我的RxJava操作的Demo记得在github给我点star或者follow一下,不然我就坐在地上不起来,哼

变换

Map操作

比如被观察者产生的事件中只有图片文件路径;,但是在观察者这里只想要bitmap,那么就需要类型变换

  Observable.just(getFilePath()
            //使用map操作来完成类型转换
            .map(new Func1<String, Bitmap>() {
              @Override
              public Bitmap call(String s) {
                //显然自定义的createBitmapFromPath(s)方法,是一个极其耗时的操作
                  return createBitmapFromPath(s);
              }
          })
            .subscribe(
                 //创建观察者,作为事件传递的终点处理事件    
                  new Subscriber<Bitmap>() {
                        @Override
                        public void onCompleted() {
                            Log.d("DDDDDD","结束观察...\n");
                        }
            
                        @Override
                        public void onError(Throwable e) {
                            //出现错误会调用这个方法
                        }
                        @Override
                        public void onNext(Bitmap s) {
                            //处理事件
                            showBitmap(s)
                        }
                    );
            
  • 实际上在使用map操作时,new Func1<String, Bitmap>() 就对应了类型的转你方向,String是原类型,Bitmap是转换后的类型。在call()方法中,输入的是原类型,返回转换后的类型

你认真看完上面的代码就会觉得,何必在过程中变换类型呢?我直接在事件传递的终点,在观察者中变换就行咯。老实说,你这个想法没毛病,但实际上,上面写的代码是不合理的。

我在代码中也提到,读取文件,创建bitmap可能是一个耗时操作,那么就应该在子线程中执行,主线程应该仅仅做展示。那么线程切换一般就会是比较复杂的事情了。但是在Rxjava中,是非常方便的。

 Observable.just(getFilePath()
           //指定了被观察者执行的线程环境
          .subscribeOn(Schedulers.newThread())
          //将接下来执行的线程环境指定为io线程
          .observeOn(Schedulers.io())
            //使用map操作来完成类型转换
            .map(new Func1<String, Bitmap>() {
              @Override
              public Bitmap call(String s) {
                //显然自定义的createBitmapFromPath(s)方法,是一个极其耗时的操作
                  return createBitmapFromPath(s);
              }
          })
            //将后面执行的线程环境切换为主线程
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(
                 //创建观察者,作为事件传递的终点处理事件    
                  new Subscriber<Bitmap>() {
                        @Override
                        public void onCompleted() {
                            Log.d("DDDDDD","结束观察...\n");
                        }
            
                        @Override
                        public void onError(Throwable e) {
                            //出现错误会调用这个方法
                        }
                        @Override
                        public void onNext(Bitmap s) {
                            //处理事件
                            showBitmap(s)
                        }
                    );
            

由上面的代码可以看到,使用操作符将事件处理逐步分解,通过线程调度为每一步设置不同的线程环境,完全解决了你线程切换的烦恼。可以说线程调度+操作符,才真正展现了RxJava无与伦比的魅力。

flatmap操作

先提出一个需求,查找一个学校每个班级的每个学生,并打印出来。

如果用老办法:先读出所有班级的数据,循环每个班级。再循环中再读取每个班级中每个学生,然后循环打印出来。

还是得说,这种想法,没毛病,就是嵌套得有点多。

Rxjava说:我不是针对谁...

//创建被观察者,获取所有班级
 Observable.from(getSchoolClasses())
                .flatMap(new Func1<SingleClass, Observable<Student>>() {
                    @Override
                    public Observable<Student> call(SingleClass singleClass) {
                        //将每个班级的所有学生作为一列表包装成一列Observable<Student>,将学生一个一个传递出去
                        return Observable.from(singleClass.getStudents());
                    }
                })
                .subscribe(
                //创建观察者,作为事件传递的终点处理事件    
                  new Subscriber<Student>() {
                        @Override
                        public void onCompleted() {
                            Log.d("DDDDDD","结束观察...\n");
                        }
            
                        @Override
                        public void onError(Throwable e) {
                            //出现错误会调用这个方法
                        }
                        @Override
                        public void onNext(Student student) {
                            //接受到每个学生类
                            Log.d("DDDDDD",student.getName())
                        }
                    );

好了,基本上按照RxJava的骨架搭起来就能完成需求。你说棒不棒??

其实FlatMap是比较难懂的一个操作符,作为初学者其实会用就好,所以我推荐的对于FlatMap的解释是:将每个Observable产生的事件里的信息再包装成新的Observable传递出来,

那么为什么FlatMap可以破除嵌套难题呢?

就是因为FlatMap可以再次包装新的Observable,而每个Observable都可以使用from(T[])方法来创建自己,这个方法接受一个列表,然后将列表中的数据包装成一系列事件。


异步(线程调度)

异步是相对于主线程来讲的子线程操作,在这里我们不妨使用线程调度这个概念更加贴切。

首先介绍一下RxJava的线程环境有哪些选项:

在讲解Map操作符时,已经提到了线程调度,在这里我用更加简介的代码代替:

            //new Observable.just()执行在新线程
Observable.just(getFilePath()
           //指定在新线程中创建被观察者
          .subscribeOn(Schedulers.newThread())
          //将接下来执行的线程环境指定为io线程
          .observeOn(Schedulers.io())
            //map就处在io线程
          .map(mMapOperater)
            //将后面执行的线程环境切换为主线程,
            //但是这一句依然执行在io线程
          .observeOn(AndroidSchedulers.mainThread())
          //指定线程无效,但这句代码本身执行在主线程
          .subscribeOn(Schedulers.io())
          //执行在主线程
          .subscribe(mSubscriber);

实际上线程调度只有subscribeOn()和observeOn()两个方法。对于初学者,只需要掌握两点:

  • subscribeOn()它指示Observable在一个指定的调度器上创建(只作用于被观察者创建阶段)。只能指定一次,如果指定多次则以第一次为准

  • observeOn()指定在事件传递(加工变换)和最终被处理(观察者)的发生在哪一个调度器。可指定多次,每次指定完都在下一步生效。

线程调度掌握到这个程度,在入门阶段时绝对够用的了。

结尾

好了,对于RxJava整个入门文章到这里就完全结束了,现在再来回看RxJava,你会发现,它就是在观察者模式的骨架下,通过丰富的操作符和便捷的异步操作来完成对于复杂业务的处理

我相信你对于整个RxJava的骨架,以及执行流程应该有了相当的了解,现在就只需要多练习一下操作符的用法了。

本文没有介绍太多的操作符,很多没来得及介绍的操作符的用法实例都放在github上的RxJavaDemo项目上了,后期还会继续加上更多操作符的使用,欢迎大家上去看看,对照代码,手机运行一下。

大家多给点star!!顺便follow一下,接下来,我也会慢慢整理出一些别的有用的项目分享给大家。

勘误

暂无

后记

如果你还有不太理解,或者觉得文章在某些地方还有提升的空间,欢迎在下方留言,帮助我一起把这篇文章改得更加简单,生动,透彻,实用。这也是我写Blog一直追求的目标。

而且,我希望这篇文章能成为RxJava入门上手最好的文章。

附源码

文章涉及和未涉及的代码范例

个人Github主页

star ~~

相关文章

网友评论

  • 闪闪发光的神经病o:rxjava2.x变动很大,博主能出一篇2,x的嘛
    拉丁吴:@闪闪发光的神经病o 有的,你在我主页上看看
  • Today_Now:博主写的很好,点赞
  • 哈喽jv:良心
  • e2f11fd62dd4:好文,写的通俗易懂,同时对重点把握的很好
  • 积极向上的老王:隔一段时间看一次,每次都有新的收获:smile:
  • shy程序: Observable.create(new Observable.just(getFilePath())
    //使用map操作来完成类型转换
    .map
    这个。。是不是写错了哇
    拉丁吴:@长腿小程序员 是的,这个错误简书这边忘了改了,已更正
  • a66779e45b1e:请问我有两个Subscriber想同时订阅一个Observable如何实现
    拉丁吴: @汪汪汪_e3d8 这本身就支持啊,你先创建observable,然后再创建两个subscriber,然后分开订阅那个observable就行了
  • 夏侯撒:学习啊,刚入行
  • Kopite__:@拉丁吴
    Observable.create(new Observable.just(getFilePath())
    //指定在新线程中创建被观察者
    .subscribeOn(Schedulers.newThread())
    //将接下来执行的线程环境指定为io线程
    .observeOn(Schedulers.io())
    //map就处在io线程
    .map(mMapOperater)
    //将后面执行的线程环境切换为主线程,
    //但是这一句依然执行在io线程
    .observeOn(AndroidSchedulers.mainThread())
    //指定线程无效,但这句代码本身执行在主线程
    .subscribeOn(Schedulers.io())
    //执行在主线程
    .subscribe(mSubscriber);

    哦……我好像理解了,observeOn(Schedulers.io()) 切换的是下面map操作符的线程。
    那楼主,如果把map操作符的代码移到subscribeOn(Schedulers.newThread())上面是不是可行?这样是不是observeOn(Schedulers.io()) 可以省略了?
    Kopite__:@拉丁吴 嗯,结合掘金中的评论 有点知道了。非常感谢作者的文章,对入门RxJava很有帮助。
    拉丁吴:@Kopite__ 其实你放在哪里都行,我的例子使用了过多的线程调度只是为了说明,实际应用是按需调配,你希望哪个地方用哪个线程,就做出相应的调配即可
  • Kopite__:楼主,我还是对线程切换有点疑问,一般看rxjava+retrofit 的教程都是类似下面的写法,和楼主你写的哪个是正确的啊?

    service.login(phone, password)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(...)
    拉丁吴:@Kopite__ 没区别啊?
  • 1bb9a89c22ef:太好了 作者是我的救星
    拉丁吴:@null_null_null 哈哈,很荣幸
  • b0d5bbdb5374:很清晰,适合初学者。
  • Rye2016:文章写的很好,容易理解,第二个Tips那里” Tips:当调用订阅操作(即调用Observer.subscribe()方法)的时候,被观察者才真正开始发出事件。“是不是应该是Observable.subscribe()方法?楼主说为了体现流式API调用风格
    拉丁吴:@Rye2016 嗯,是为了保证流式API调用的风格。。
    拉丁吴:@Rye2016 嗯嗯,是的,我写错了,
  • 会理发的店小二:不黑不吹,这篇文章很良心.
  • 会理发的店小二:楼主下面这句话是不是有点问题?"Tips: 当调用订阅操作(即调用Observer.subscribe()方法)的时候,被观察者才真正开始发出事件"。应该是Observable.subscribe()吧?
    拉丁吴:@有理想的狗剩哥 嗯嗯,你说得对,是这样
  • cdfd8e7a3fbe:已关注,希望楼主后面能出篇rxjava结合retrofit的文章
  • 汪简书:博主写的太好的,自己认识rxjava是对evenBus的进一步抽象,概况。基础是各种回调方法。期待博主rxjava与retrofit的文章(o´ω`o)
    拉丁吴:@然而喵星人早已看穿了一切 嗯嗯,后面会总结出来的。欢迎关注

本文标题:关于RxJava最友好的文章

本文链接:https://www.haomeiwen.com/subject/llcquttx.html