Android框架式编程之RxJava-白红宇的个人博客

Android框架式编程之RxJava

发布日期：2021-05-09 06:09:15 浏览次数：19 分类：博客文章

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

RxJava是一个很好的开源库，将链式编程和异步很好的结合在一起。在介绍RxJava之前我们先介绍一下观察者模式，这样我们能更好的学习和理解RxJava。

简单介绍一下，A和B两个，A是被观察者，B是观察者，B对A进行观察，B并不是需要时刻盯着A，而是A如果发生了变化，会主动通知B，B会对应做一些变化。

如上图，假设A是连载小说，B是读者，读者订阅了连载小说，当小说出现了新的连载的时候，会推送给读者。读者不用时刻盯着小说连载，而小说有了新的连载会主动推送给读者。这就是观察者模式。而RxJava正是基于观察者模式开发的。

一、RxJava的基础使用

目前主流的RxJava使用的版本为RxJava2系列，在使用之前我们需要先引入RxJava2.0相关的类库：

implementation 'io.reactivex.rxjava2:rxjava:2.2.13' implementation 'io.reactivex.rxjava2:rxandroid:2.1.1'

使用方式：

1). 创建被观察者：

//  创建被观察者Observable novelObservable = Observable.create(new ObservableOnSubscribe
      
       () {    @Override    public void subscribe(ObservableEmitter
       
         observableEmitter) throws Exception {        observableEmitter.onNext("新章节1001");        observableEmitter.onNext("新章节1002");        observableEmitter.onNext("新章节1003");        observableEmitter.onComplete();    }});

Observable中文意思就是被观察者，通过create方法生成对象，里面放的参数ObservableOnSubscribe<T>，可以理解为一个计划表，泛型T是要操作对象的类型，重写subscribe方法，里面写具体的计划，本文的例子就是推送新章节1001、新章节1002和新章节1003，在subscribe中的ObservableEmitter<String>对象的Emitter是发射器的意思。

ObservableEmitter有三种发射的方法，分别是void onNext(T value)、void onError(Throwable error)、void onComplete()，onNext方法可以无限调用，Observer（观察者）所有的都能接收到，onError和onComplete是互斥的，Observer（观察者）只能接收到一个，OnComplete可以重复调用，但是Observer（观察者）只会接收一次，而onError不可以重复调用，第二次调用就会报异常。

2). 创建观察者

// 创建观察者Observer
      
        reader = new Observer
       
        () {    @Override    public void onSubscribe(Disposable disposable) {        doLog("subscribe success");    }    @Override    public void onNext(String s) {        doLog("收到最新章节 ===》" + s);    }    @Override    public void onError(Throwable throwable) {    }    @Override    public void onComplete() {        doLog("接受结束！");    }};novelObservable.subscribe(reader);

通过new创建接口，并实现其内部的方法，看方法其实就应该差不多知道干嘛的，onNext、onError、onComplete都是跟被观察者发射的方法一一对应的，这里就相当于接收了。onSubscribe（Disposable d）里面的Disposable对象要说一下，Disposable英文意思是可随意使用的，这里就相当于读者和连载小说的订阅关系，如果读者不想再订阅该小说了，可以调用 mDisposable.dispose()取消订阅，此时连载小说更新的时候就不会再推送给读者了。

3). 将订阅者和被订阅者建立关系

novelObservable.subscribe(reader);

运行代码，查看输出结果如下：

总结：以上就是RxJava2最简单的用法，总结一下就是：

1). 创建被订阅者

2). 创建订阅者

3). 建立订阅关系

二、RxJava异步链式编程

RxJava 是支持异步的，核心就是 Scheduler （调度器），RxJava用 Scheduler 来控制线程。

当我们没有设置 Scheduler 的时候，RxJava遵循在哪个线程启动，就在哪个线程运行的原则。

例：一般情况下我们在使用RxJava的时候都是后台执行，前台调用的，本着这个原则，我们需要调用observeOn(AndroidSchedulers.mainThread())，subscribeOn(Schedulers.io())。

这里observeOn是事件回调的线程，AndroidSchedulers.mainThread()一看就知道是主线程；subscribeOn是事件执行的线程，Schedulers.io()是子线程，这里也可以用Schedulers.newThread()，只不过io线程可以重用空闲的线程，因此多数情况下 io() 比 newThread() 更有效率。

下面我们改造一下第一节的代码，改造后如下：

Log.e("111", "主Thread Hash = " + Thread.currentThread().hashCode());Observable.create(new ObservableOnSubscribe
      
       () {    @Override    public void subscribe(ObservableEmitter
       
         observableEmitter) throws Exception {        Log.e("111", "被订阅Thread Hash = " + Thread.currentThread().hashCode());        observableEmitter.onNext("新章节1001");        observableEmitter.onNext("新章节1002");        observableEmitter.onNext("新章节1003");        observableEmitter.onComplete();    }}).subscribeOn(Schedulers.io())  // 执行在io线程        .observeOn(AndroidSchedulers.mainThread())  //  回调在主线程        .subscribe(new Observer
        
         () {            @Override            public void onSubscribe(Disposable disposable) {                doLog("subscribe success");            }            @Override            public void onNext(String s) {                Log.e("111", "订阅Thread Hash = " + Thread.currentThread().hashCode());                doLog("收到最新章节 ===》" + s);            }            @Override            public void onError(Throwable throwable) {            }            @Override            public void onComplete() {                doLog("接受结束！");            }        });

输出结果：

E/111: 主Thread Hash = 72810649E/111: 日志 -----> subscribe successE/111: 被订阅Thread Hash = 68064815E/111: 订阅Thread Hash = 72810649E/111: 日志 -----> 收到最新章节 ===》新章节1001    订阅Thread Hash = 72810649    日志 -----> 收到最新章节 ===》新章节1002    订阅Thread Hash = 72810649    日志 -----> 收到最新章节 ===》新章节1003    日志 -----> 接受结束！

三、RxJava的应用场景

1). RxJava + Retrofit 封装网络请求框架

2). 基于RxJava的开源类库 Rxpermissions、RxBinding、RxBus

四、参考资料

1). RxJava2.0——从放弃到入门：

2). 打造终极MVP+Retrofit2+okhttp3+Rxjava2网络请求：

3). 我为什么不再推荐RxJava：

上一篇：FFmpeg命令行工具学习(五)：FFmpeg 调整音视频播放速度

下一篇：Android Gradle 学习笔记（一）：Gradle 入门

发表评论

关于作者

喝酒易醉，品茶养心，人生如梦，品茶悟道，何以解忧？唯有杜康！

-- 愿君每日到此一游！