IT数码 购物 网址 头条 软件 日历 阅读 图书馆
TxT小说阅读器
↓语音阅读,小说下载,古典文学↓ 		图片批量下载器
↓批量下载图片,美女图库↓ 		图片自动播放器
↓图片自动播放器↓ 		一键清除垃圾
↓轻轻一点,清除系统垃圾↓
开发: C++知识库 Java知识库 JavaScript Python PHP知识库 人工智能 区块链 大数据 移动开发 嵌入式 开发工具 数据结构与算法 开发测试 游戏开发 网络协议 系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑 笔记本 显卡 显示器 固态硬盘 硬盘 耳机 手机 iphone vivo oppo 小米 华为 单反 装机 图拉丁
 
   -> 移动开发 -> 探索Android开源框架 - 3. RxJava使用及源码解析 -> 正文阅读

[移动开发]探索Android开源框架 - 3. RxJava使用及源码解析

相关概念

Android多线程编程的原则:

  1. 不要阻塞UI线程;
  2. 不要在UI线程之外访问UI组件;

ReactiveX

  • Reactive Extensions的缩写，一般简写为Rx;
  • 是一个使用可观察数据流进行异步编程的编程接口，ReactiveX结合了观察者模式、迭代器模式和函数式编程的精华;

RxJava

  • Reactive Extensions for the JVM:，RxJava就是ReactiveX在JVM平台的实现;
  • 基于事件流的链式调用，进行耗时任务，线程切换，其本质是一个异步操作库;

原理

  • 基于观察者模式, 一个方面的操作依赖于另一个方面的状态变化，当一个对象必须通知其他对象，又希望这个对象和其他被通知的对象是松散耦合的；

三个要素

  • 被观察者（Observable），观察者(Subscriber)，订阅(subscribe);

被观察者

1. Observable
  • 可多次发送事件(onNext)，直到 onComplete 或 onError 被调用结束订阅;
  • 不支持背压：当被观察者快速发送大量数据时，下游不会做其他处理，即使数据大量堆积，调用链也不会报MissingBackpressureException,消耗内存过大只会OOM。（官方给出以1000个事件为分界线作为参考）;
val observable = Observable.create(ObservableOnSubscribe<Int> {
    it.onNext(1)
    it.onNext(3)
    it.onNext(5)
    it.onComplete()
    LjyLogUtil.d("${Thread.currentThread().name}_subscribe")
})
//Disposable通过 dispose() ?方法来让上游停?止?工作，达到「丢弃」的效果
var disposable: Disposable? = null
val observer = object : Observer<String> {
    override fun onSubscribe(d: Disposable) {
        //订阅后发送数据之前, 回调这个方法，Disposable可用于取消订阅
        disposable = d
        LjyLogUtil.d("${Thread.currentThread().name}_onSubscribe")
    }

    override fun onNext(t: String) {
        LjyLogUtil.d("${Thread.currentThread().name}_onNext:$t")
    }

    override fun onError(e: Throwable) {
        LjyLogUtil.d("${Thread.currentThread().name}_onError:${e.message}")
    }

    override fun onComplete() {
        LjyLogUtil.d("${Thread.currentThread().name}_onComplete")
    }
}

//rxJava的整体结构是一条链，链的最上游是被观察者observable，最下游是观察者observer，
//链的中间各个节点，既是其下游的observable，又是其上游的observer
observable
    //subscribeOn: 切换起源 Observable 的线程,
    // 当多次调?用 subscribeOn() 的时候，只有最上?面的会对起源 Observable 起作?用，
    // 原因subscribeOn底层通过新建observable实现
    .subscribeOn(Schedulers.io())
    .subscribeOn(AndroidSchedulers.mainThread())
    //observeOn指定的是它之后的操作所在的线程，通过observeOn的多次调用，程序实现了线程的多次切换
    //影响范围observeOn是它下面的每个observer，除非又遇到新的observeOn
    .observeOn(Schedulers.io())
    .map {
        LjyLogUtil.d("map：${Thread.currentThread().name}")
        "num_$it"
    }
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(observer)
2. Flowable:
  • 和 Observable 一样, 且支持Reactive-Streams和背压;
  • 上游的被观察者会响应下游观察者的数据请求，下游调用request(n)来告诉上游发送多少个数据。这样避免了大量数据堆积在调用链上，使内存一直处于较低水平;
var sub: Subscription? = null
Flowable.create(FlowableOnSubscribe<Int> {
    it.onNext(1)
    it.onNext(3)
    it.onNext(5)
    it.onComplete()
    LjyLogUtil.d("${Thread.currentThread().name}_subscribe")
    //使用create创建Flowable，需要指定背压策略
}, BackpressureStrategy.BUFFER)

Flowable.range(0, 5)
    .subscribe(object : Subscriber<Int> {
        override fun onSubscribe(s: Subscription?) {
            //当订阅后，会首先调用这个方法，其实就相当于onStart()，
            //传入的Subscription s参数可以用于请求数据或者取消订阅
            LjyLogUtil.d("${Thread.currentThread().name}_onSubscribe start")
            sub = s
            sub?.request(1)
            LjyLogUtil.d("${Thread.currentThread().name}_onSubscribe end")
        }

        override fun onNext(t: Int?) {
            LjyLogUtil.d("${Thread.currentThread().name}_onNext:$t")
            sub?.request(1)
        }

        override fun onError(t: Throwable?) {
            LjyLogUtil.d("${Thread.currentThread().name}_onError:${t?.message}")
        }

        override fun onComplete() {
            LjyLogUtil.d("${Thread.currentThread().name}_onComplete")
        }

    })
3. Single
  • 单次发送事件（onSuccess、onError），发完即结束订阅;
  • 总是只发射一个值，或者一个错误通知，而不是发射一系列的值（当然就不存在背压问题）;
  • 一般一个接口只是一次请求一次返回，所以使用Single 与 Retrofit 配合是更为合理;
Single.just(1)
    .map { "num_$it" }
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe { num ->
        LjyLogUtil.d(num)
    }
Single.create(SingleOnSubscribe<String> { emitter ->
    LjyLogUtil.d("${Thread.currentThread().name}_subscribe")
    emitter.onSuccess("str1")
    emitter.onSuccess("str2")//错误写法，重复调用也不会处理，因为只会调用一次
}).subscribe(object : SingleObserver<String> {
    override fun onSubscribe(d: Disposable) {
        LjyLogUtil.d("${Thread.currentThread().name}_onSubscribe")
    }

    override fun onSuccess(t: String) {
        LjyLogUtil.d("${Thread.currentThread().name}_onSuccess:$t")
    }

    override fun onError(e: Throwable) {
        LjyLogUtil.d("${Thread.currentThread().name}_onError:${e.message}")
    }

})
4. Completable
  • 单次发送事件（onError、onComplete），发完即结束订阅;
  • 如果你的观察者连onNext事件都不关心，可以使用Completable，它只有onComplete和onError两个事件,要转换成其他类型的被观察者，也是可以使用toFlowable()、toObservable()等方法去转换;
Completable.create { emitter ->
    LjyLogUtil.d("${Thread.currentThread().name}_subscribe")
    emitter.onComplete()//单一onComplete或者onError
}.subscribe(object : CompletableObserver {
    override fun onSubscribe(d: Disposable) {
        LjyLogUtil.d("${Thread.currentThread().name}_onSubscribe")
    }

    override fun onComplete() {
        LjyLogUtil.d("${Thread.currentThread().name}_onComplete")
    }

    override fun onError(e: Throwable) {
        LjyLogUtil.d("${Thread.currentThread().name}_onError:${e.message}")
    }

})
5. Maybe
  • 单次发送事件（onSuccess、onError、onComplete），发完即结束订阅。相当于 Completable 和 Single 结合;
  • 如果可能发送一个数据或者不会发送任何数据，这时候你就需要Maybe，它类似于Single和Completable的混合体;
  • onSuccess和onComplete是互斥的存在;
Maybe.create(MaybeOnSubscribe<String> {
    it.onSuccess("str1")
    it.onComplete()
}).subscribe(object : MaybeObserver<String> {
    override fun onSubscribe(d: Disposable) {
        LjyLogUtil.d("${Thread.currentThread().name}_onSubscribe")
    }

    override fun onSuccess(t: String) {
        LjyLogUtil.d("${Thread.currentThread().name}_onSuccess:$t")
    }

    override fun onError(e: Throwable) {
        LjyLogUtil.d("${Thread.currentThread().name}_onError:${e.message}")
    }

    override fun onComplete() {
        LjyLogUtil.d("${Thread.currentThread().name}_onComplete")
    }

})
  • 五种被观察者可通过toObservable，toFlowable,toSingle,toCompletable,toMaybe相互转换;

观察者

val observer = object : Observer<Int> {
        override fun onSubscribe(d: Disposable) {
            LjyLogUtil.d("${Thread.currentThread().name}_onSubscribe")
        }

        override fun onNext(t: Int) {
            LjyLogUtil.d("${Thread.currentThread().name}_onNext:$t")
        }

        override fun onError(e: Throwable) {
            LjyLogUtil.d("${Thread.currentThread().name}_onError:${e.message}")
        }

        override fun onComplete() {
            LjyLogUtil.d("${Thread.currentThread().name}_onComplete")
        }
    }

线程调度器（Schedulers）

  • 指定 被观察者（Observable）/ 观察者（Observer）的工作线程,简化了异步操作;
  • 默认在创建自身的线程;
  • 配合操作符 subscribeOn , observeOn 使用;
AndroidSchedulers.mainThread()
  • 需要引用RxAndroid, 切换到UI线程（Android的主线程）, 为Android开发定制;
Schedulers.io()
  • 用于IO密集型任务，如读写SD卡文件，查询数据库，访问网络等;
  • 具有线程缓存机制，默认是一个CacheThreadScheduler;
Schedulers.newThread()
  • 为每一个任务创建一个新线程;
  • 不具有线程缓存机制，虽然使用Schedulers.io的地方，都可以使用Schedulers.newThread，但是，Schedulers.newThread的效率没有Schedulers.io高;
Schedulers.computation()
  • 用于CPU 密集型计算任务，即不会被 I/O 等操作限制性能的耗时操作，例如xml,json文件的解析，Bitmap图片的压缩取样等，具有固定的线程池，大小为CPU的核数。不可以用于I/O操作，因为I/O操作的等待时间会浪费CPU。
Schedulers.trampoline()
  • 在当前线程立即执行任务，如果当前线程有任务在执行，则会将其暂停，等插入进来的任务执行完之后，再将未完成的任务接着执行;
Schedulers.single()
  • 拥有一个线程单例，所有的任务都在这一个线程中执行，当此线程中有任务执行时，其他任务将会按照先进先出的顺序依次执行;
Scheduler.from(executor)
  • 指定一个线程调度器，由此调度器来控制任务的执行策略;

背压(Backpressure)

  • 背压是指在异步场景中，被观察者发送事件速度远快于观察者的处理速度的情况下，一种告诉上游的被观察者降低发送速度的策略;
  • 支持背压的被观察者为Flowable；
  • Android中很少用到，除非在线视频流，直播等场景，当画面卡顿已取得的数据失效了，需要抛弃等;
背压策略模式
BackpressureStrategy.MISSING
  • 在此策略下，通过Create方法创建的Flowable相当于没有指定背压策略，不会对通过onNext发射的数据做缓存或丢弃处理，需要下游通过背压操作符处理
BackpressureStrategy.ERROR:
  • 在此策略下，如果放入Flowable的异步缓存池中的数据超限了，则会抛出MissingBackpressureException异常;
BackpressureStrategy.BUFFER:
  • 内部维护了一个缓存池SpscLinkedArrayQueue，其大小不限，此策略下，如果Flowable默认的异步缓存池满了，会通过此缓存池暂存数据，它与Observable的异步缓存池一样，可以无限制向里添加数据，不会抛出MissingBackpressureException异常，但会导致OOM;
  • 当缓存区大小存满（默认缓存区大小 = 128）、被观察者仍然继续发送下1个事件时，将缓存区大小设置成无限大，被观察者可无限发送事件 观察者，但实际上是存放在缓存区，但要注意内存情况，防止出现OOM;
BackpressureStrategy.DROP
  • 在此策略下，如果Flowable的异步缓存池满了，会丢掉上游发送的数据;
BackpressureStrategy.LATEST
  • 与Drop策略一样，如果缓存池满了，会丢掉将要放入缓存池中的数据，不同的是，不管缓存池的状态如何，LATEST都会将最后一条数据强行放入缓存池中，来保证观察者在接收到完成通知之前，能够接收到Flowable最新发射的一条数据;
  • 即如果发送了150个事件，缓存区里会保存129个事件（第1-第128 + 第150事件）;
RxJava 2.0内部提供 封装了背压策略模式的方法
  • 默认采用BackpressureStrategy.ERROR模式;
onBackpressureBuffer()
onBackpressureDrop()
onBackpressureLatest()

Flowable.interval(1, TimeUnit.MILLISECONDS)
    //添加背压策略封装好的方法，此处选择Buffer模式，即缓存区大小无限制
    .onBackpressureBuffer()
    .observeOn(Schedulers.newThread())
    .subscribe(subscriber)

冷热流

Cold Observable

  • subscribe时才会发射数据;
  • 常见的工厂方法提供的都是ColdObservable,包括just(),fromXX,create(),interval(),defer();
  • 当你有多个Subscriber的时候，他们的事件是独立的,示例代码如下：
val interval = Observable.interval(1, TimeUnit.SECONDS)
var disposable1: Disposable? = null
val observer1 = object : Observer<Long> {
    override fun onSubscribe(d: Disposable) {
        disposable1 = d
    }

    override fun onNext(t: Long) {
        LjyLogUtil.d("观察者1：$t")
    }

    override fun onError(e: Throwable) {

    }

    override fun onComplete() {

    }
}

var disposable2: Disposable? = null
val observer2 = object : Observer<Long> {
    override fun onSubscribe(d: Disposable) {
        disposable1 = d
    }

    override fun onNext(t: Long) {
        LjyLogUtil.d("观察者2：$t")
    }

    override fun onError(e: Throwable) {

    }

    override fun onComplete() {

    }
}

findViewById<Button>(R.id.btn_subscribe1)
    .clicks().subscribe {
        interval.subscribe(observer1)
    }
findViewById<Button>(R.id.btn_dispose1)
    .clicks().subscribe {
        disposable1?.dispose()
    }


findViewById<Button>(R.id.btn_subscribe2)
    .clicks().subscribe {
        interval.subscribe(observer2)
    }
findViewById<Button>(R.id.btn_dispose2)
    .clicks().subscribe {
        disposable2?.dispose()
    }

Hot Observable

  • 对于Hot Observable的所有subscriber,他们会在同一时刻收到相同的数据;
  • 通常使用publish()操作符来将ColdObservable变为Hot。或者 使用 Subjects 也是Hot Observable;
  • 如果他开始传输数据，你不主动喊停(dispose()/cancel())，那么他就不会停，一直发射数据，即使他已经没有Subscriber了;
val interval = Observable.interval(1, TimeUnit.SECONDS).publish()
var disposable: Disposable? = null
findViewById<Button>(R.id.btn_connect)
    .clicks().subscribe {
        disposable = interval.connect()
    }
findViewById<Button>(R.id.btn_dispose)
    .clicks().subscribe {
        disposable?.dispose()
    }
findViewById<Button>(R.id.btn_subscribe3)
    .clicks().subscribe {
        interval.subscribe {
            LjyLogUtil.d("观察者3：$it")
        }
    }
findViewById<Button>(R.id.btn_subscribe4)
    .clicks().subscribe {
        interval.subscribe {
            LjyLogUtil.d("观察者4：$it")
        }
    }

操作符

  • 操作符很多，不用完全背下来，浏览一遍，用时知道在哪找（关注+收藏本文[狗头]）即可

线程切换

  • subscribeOn：指定被观察者Observable的工作线程;
  • observeOn：指定观察者的observer工作线程;
observable
    //subscribeOn: 切换起源 Observable 的线程,
    // 当多次调?用 subscribeOn() 的时候，只有最上?面的会对起源 Observable 起作?用，
    // 原因subscribeOn底层通过新建observable实现
    .subscribeOn(Schedulers.io())
    .subscribeOn(AndroidSchedulers.mainThread())
    //observeOn指定的是它之后的操作所在的线程，通过observeOn的多次调用，程序实现了线程的多次切换
    //影响范围observeOn是它下面的每个observer，除非又遇到新的observeOn
    .observeOn(Schedulers.io())
    .map {
        LjyLogUtil.d("map：${Thread.currentThread().name}")
        "num_$it"
    }
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(observer)

创建操作

基本创建
create
  • 通过调用观察者的方法从头创建一个Observable;
Observable.create(ObservableOnSubscribe<Int> {
    LjyLogUtil.d("${Thread.currentThread().name}_subscribe")
    it.onNext(1)
    it.onNext(3)
    it.onNext(5)
    it.onComplete()
})
快速创建
just
  • 将对象或者对象集合转换为一个会发射这些对象的Observable;
Observable.just(1, 2, 3, 4)
fromArray & fromIterable
  • 将其它的对象或数据结构转换为Observable;
Observable.fromArray(arrayOf(1,2,3))

Observable.fromIterable(listOf(4,5,6))
never
  • 创建的被观察者不发送任何事件, 观察者接收后什么都不调用;
Observable.never<Int>().subscribe(observer)
empty
  • 创建的被观察者仅发送Complete事件，直接通知完成, 观察者接收后会直接调用onCompleted;
Observable.empty<Int>().subscribe(observer)
error
  • 创建的被观察者仅发送Error事件，直接通知异常, 观察者接收后会直接调用onError;
Observable.error<Int>(RuntimeException()).subscribe(observer)
延迟创建
defer:
  • 在观察者订阅之前不创建这个Observable，为每一个观察者创建一个新的Observable;
var num=1
val observable=Observable.defer {Observable.just(num)}
num=2
observable.subscribe(observer)
timer
  • 创建在一个指定的延迟之后发射单个数据的Observable;
//延时3秒后，发送一个整数0
Observable.timer(3, TimeUnit.SECONDS)
interval & intervalRange
  • 创建一个定时发射整数序列的Observable;
//初始延时1秒，每3秒发一个自增整数
Observable.interval(1, 3, TimeUnit.SECONDS)

//初始延时2秒，后每1秒发一个从10开始的整数，发5个（发到14）停止
Observable.intervalRange(10, 5, 2, 1, TimeUnit.SECONDS);
range & rangeLong
  • 创建发射指定范围的整数序列的Observable;
Observable.range(0, 5)

Observable.rangeLong(0, 5)
Repeat
  • 创建重复发射特定的数据或数据序列的Observable;
//一直重复
Observable.fromArray(1, 2, 3, 4).repeat()
//重复发送5次
Observable.fromArray(1, 2, 3, 4).repeat(5)
//重复发送直到符合条件时停止重复
Observable.fromArray(1, 2, 3, 4).repeatUntil { false }
//
Observable.just(1, 2, 3, 4)
    .repeatWhen {
        it.flatMap { obj ->
            if (obj is NumberFormatException) {
                Observable.error(Throwable("repeatWhen终止"))
            } else {
                Observable.just(5, 6, 7)
            }
        }
    }.subscribe(observer)

2. 变换操作:

map
  • 映射，通过对序列的每一项都应用一个函数变换Observable发射的数据，实质是对序列中的每一项执行一个函数，函数的参数就是这个数据项;
Observable.just("1", "2", "3").map { it.toInt() }.subscribe(observer)
flatMap & concatMap
  • 扁平映射，将Observable发射的数据变换为Observables集合，然后将这些Observable发射的数据平坦化的放进一个单独的Observable，可以认为是一个将嵌套的数据结构展开的过程;
//concatMap与flatMap的区别: concatMap是有序的，flatMap是无序的
Observable.just("A", "B", "C")
    .flatMap { x ->
        Observable.intervalRange(1, 3, 0, 1, TimeUnit.SECONDS).map { y ->
            "($x,$y)"
        }
    }
Observable.just("A", "B", "C")
    .concatMap { m ->
        Observable.intervalRange(1, 3, 0, 1, TimeUnit.SECONDS).map { n ->
            "($m,$n)"
        }
    }
groupBy
  • 分组，将原来的Observable分拆为Observable集合，将原始Observable发射的数据按Key分组，每一个Observable发射一组不同的数据;
Observable.just(
    "Tiger",
    "Elephant",
    "Cat",
    "Chameleon",
    "Frog",
    "Fish",
    "Turtle",
    "Flamingo"
)
    .groupBy { it[0].uppercaseChar() }
    .concatMapSingle { it.toList() }
    .subscribe(observer4)
scan
  • 扫描，对Observable发射的每一项数据应用一个函数，然后按顺序依次发射这些值;
  • 对发射的数据和上一轮发射的数据进行函数处理，并返回的数据供下一轮使用，持续这个过程来产生剩余的数据流。其应用场景有简单的累加计算，判断所有数据的最小值等;
Observable.just(1, 2, 3, 4)
    .scan { t1, t2 -> t1 + t2 }
    .subscribe(observer)
buffer
  • 缓存，定期从Observable收集数据到一个集合，然后把这些数据集合打包发射，而不是一次发射一个
Observable.just(1, 2, 3, 4, 5, 6, 7, 8)
    .buffer(3)
    .subscribe(observer5)
Window
  • 窗口，定期将来自Observable的数据分拆成一些Observable窗口，然后发射这些窗口，而不是每次发射一项; 类似于Buffer，但Buffer发射的是数据，Window发射的是Observable，每一个Observable发射原始Observable的数据的一个子集;
//window操作符和buffer操作符在功能上实现的效果是一样的，但window操作符最大区别在于同样是缓存一定数量的数据项，
// window操作符最终发射出来的是新的事件流integerObservable，而buffer操作符发射出来的是新的数据流，
// 也就是说，window操作符发射出来新的事件流中的数据项，还可以经过Rxjava其他操作符进行处理。
Observable.just(1, 2, 3, 4)
    .window(2, 1)
    .subscribe(observer6)

3. 过滤操作:

Filter
  • 过滤，过滤掉没有通过谓词测试的数据项，只发射通过测试的;
Observable.just(1, 2, 3, 4, 5)
    .filter {
            it % 2 == 0
    }.subscribe(observer)
ofType
  • 过滤特定类型的数据;
Observable.just(1, 2.1, "3", 4L, 5.0)
    .ofType(Int::class.java)
    .subscribe(observer)
distinct & distinctUntilChanged
  • 去重，过滤掉重复数据项;
Observable.just(1, 2, 3, 4, 3, 2, 1)
    .distinct()
//  .distinctUntilChanged()//去掉相邻连续重复数据
    .subscribe(observer)
skip & skipLast
val observable = Observable.just(1, 2, 3, 4, 5, 6, 7, 8)
// 跳过, 跳过前面的若干项数据;
observable.skip(4).subscribe(observer)
//skipLast：跳过后面的若干项数据
observable.skipLast(4).subscribe(observer)
take & takeLast
val observable = Observable.just(1, 2, 3, 4, 5, 6, 7, 8)
//take：只保留前面的若干项数据
observable.take(4).subscribe(observer)
//takeLast：只保留后面的若干项数据
observable.takeLast(4).subscribe(observer)
debounce
  • 只有在空闲了一段时间后才发射数据，通俗的说，就是如果一段时间没有操作，就执行一次操作;
val observable2 = Observable.create<Int> {
    it.onNext(1)
    Thread.sleep(400)
    it.onNext(2)
    Thread.sleep(1200)
    it.onNext(3)
    Thread.sleep(1000)
    it.onNext(4)
    Thread.sleep(800)
    it.onNext(5)
    Thread.sleep(2000)
    it.onNext(6)
}
observable2
    .subscribeOn(Schedulers.io())
    .observeOn(Schedulers.io())
    .debounce(1, TimeUnit.SECONDS)
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(observer)
sample
  • 取样，定期发射最新的数据，等于是数据抽样;
//sample:与debounce的区别是，sample是以时间为周期的发射，一秒又一秒内的最新数据。而debounce是最后一个有效数据开始
observable2
    .subscribeOn(Schedulers.io())
    .observeOn(Schedulers.io())
    .sample(1, TimeUnit.SECONDS)
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(observer)
throttleFirst & throttleLast & throttleWithTimeout & throttleLatest
  • throttleFirst是指定周期内第一个数据，throttleLast与sample一致。throttleWithTimeout与debounce一致;
observable2
    .subscribeOn(Schedulers.io())
    .observeOn(Schedulers.io())
    .throttleFirst(1, TimeUnit.SECONDS)
//  .throttleLast(1, TimeUnit.SECONDS)
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(observer)
timeout
  • 后一个数据发射未在前一个元素发射后规定时间内发射则返回超时异常;
observable2
    .subscribeOn(Schedulers.io())
    .observeOn(Schedulers.io())
    .timeout(1, TimeUnit.SECONDS)
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(observer)
elementAt
  • 取值，取特定位置的数据项;
Observable.just(4, 3, 2, 1)
    .elementAt(1)
//出现越界时，抛出异常
//  .elementAtOrError(1)
    .subscribe(maybeObserver)
first
  • 首项，只发射满足条件的第一条数据;
Observable.just(1, 2, 3, 4, 5)
//  .first(-1)
//  .firstOrError()
    .firstElement()
    .subscribe(maybeObserver)
last
  • 末项，只发射最后一条数据;
 Observable.just(1, 2, 3, 4, 5)
//  .last(-1)
//  .lastOrError()
    .lastElement()
    .subscribe(maybeObserver)
IgnoreElements
  • 忽略所有的数据，只保留终止通知(onError或onCompleted),ignoreElements 作用于Flowable、Observable。ignoreElement作用于Maybe、Single;
Observable.just(1, 2, 3, 4, 5)
    .ignoreElements()
    .subscribe(object : CompletableObserver {
        override fun onSubscribe(d: Disposable) {
            LjyLogUtil.d("${Thread.currentThread().name}_onSubscribe")
        }

        override fun onComplete() {
            LjyLogUtil.d("${Thread.currentThread().name}_onComplete")
        }

        override fun onError(e: Throwable) {
            LjyLogUtil.d("${Thread.currentThread().name}_onError:${e.message}")
        }
    })

4. 组合操作:

concat & concatArray
  • 不交错的连接多个Observable的数据;
  • 组合多个被观察者一起发送数据，合并后 按发送顺序串行执行;
val just1 = Observable.just(1, 2, 3)
val just2 = Observable.just("A", "B", "C")
val just3 = Observable.just(4, 5, 6)
Observable.concat(just1, just2).subscribe(observer3)

//concat组合被观察者数量<=4个，而concatArray则可>4个
Observable.concatArray(
    Observable.just(1, 2, 3),
    Observable.just(4, 5, 6),
    Observable.just(7, 8, 9)
).subscribe(observer3)
merge & mergeArray
  • 组合多个被观察者一起发送数据，合并后 按时间线并行执行;
  • merge和concat的区别: merge合并后发射的数据项是并行无序的，concat合并后发射的数据项是串行有序的;
Observable.merge(just1, just2).subscribe(observer3)
//mergeWith
just1.mergeWith(just3).subscribe(observer3)
zip
  • 打包，使用一个指定的函数将多个Observable发射的数据组合在一起，然后将这个函数的结果作为单项数据发射;
//zip操作符是将两个数据流进行指定的函数规则合并
Observable.zip(just1, just2, { t1, t2 -> "${t1}_${t2}" })
    .subscribe(observer3)
//zipWith
just1.zipWith(just2, { t1, t2 -> "${t1}_${t2}" })
    .subscribe(observer3)
startWith & startWithArray
  • 在发射原来的Observable的数据序列之前，先发射一个指定的数据序列或数据项;
Observable.just(1,2,3)
    .startWith(Observable.just(4,5,6))
    .startWithArray(7,8,9)
    .subscribe(observer3)
join
  • 无论何时，如果一个Observable发射了一个数据项，只要在另一个Observable发射的数据项定义的时间窗口内，就将两个Observable发射的数据合并发射;
just1.join(just2,
    //规定just2的过期期限
    { Observable.timer(3, TimeUnit.SECONDS) },
    //规定just1的过期期限
    { Observable.timer(8, TimeUnit.SECONDS) },
    //规定just1和just2的合并规则
    { t1, t2 -> "${t1}_${t2}" })
    .subscribe(observer3)
combineLatest
  • 当两个Observables中的任何一个发射了一个数据时，通过一个指定的函数组合每个Observable发射的最新数据（一共两个数据），然后发射这个函数的结果;
Observable.combineLatest(just1, just2,
    { t1, t2 -> "${t1}_${t2}" }).subscribe(observer3)
switch
  • 将一个发射Observable序列的Observable转换为这样一个Observable：它逐个发射那些Observable最近发射的数据;
Observable.switchOnNext(ObservableSource<Observable<Int>> { }, 12)
collect
  • 将被观察者Observable发送的数据事件收集到一个数据结构里;
Observable.just(1, 2, 3, 4, 5, 6)
    .collect(
        { ArrayList() },
        BiConsumer<ArrayList<Int?>, Int> { t1, t2 -> t1.add(t2) }
    ).subscribe(Consumer { LjyLogUtil.d("num: $it") })
count
  • 计算Observable发射的数据个数，然后发射这个结果;
Observable.just(1, 2, 3)
    .count()
    .subscribe(object : SingleObserver<Long>{
        override fun onSubscribe(d: Disposable) {
            LjyLogUtil.d("${Thread.currentThread().name}_onSubscribe")
        }

        override fun onSuccess(t: Long) {
            LjyLogUtil.d("${Thread.currentThread().name}_onSuccess:$it")
        }

        override fun onError(e: Throwable) {
            LjyLogUtil.d("${Thread.currentThread().name}_onError:${e.message}")
        }

    })

5. 错误处理:

cast
  • 将数据元素转型成其他类型,转型失败会抛出异常;
Observable.just(1, 4.0, 3f, 7, 12, 4.6, 5)
    .cast(Int::class.java)
    .subscribe(observer)
onErrorReturn
  • 调用数据源的onError函数后会回到该函数，可对错误进行处理，然后返回值，会调用观察者onNext()继续执行，执行完调用onComplete()函数结束所有事件的发射;
Observable.just(1, 2, 3.2, 4)
    .map { it.toInt() }
    .onErrorReturn {
        if (it is NumberFormatException) {
            0
        } else {
            throw IllegalArgumentException()
        }
    }.subscribe(observer)
onErrorReturnItem
  • 与onErrorReturn类似，onErrorReturnItem不对错误进行处理，直接返回一个值;
Observable.just(1, 2, 3.2, 4)
    .map { it.toInt() }
    .onErrorReturnItem(0)
    .subscribe(observer)
onErrorResumeNext & onExceptionResumeNext
  • 遇到错误时，发送1个新的Observable;
Observable.just(1, 2, 3.2, 4)
    .map { it.toInt() }
    .onErrorResumeNext { Observable.just(5,6,7) }
    .subscribe(observer)
retry
  • 重试，如果Observable发射了一个错误通知，重新订阅它，期待它正常终止;
//retry:当发生错误时，数据源重复发射item，直到没有异常或者达到所指定的次数
Observable.just(1, 2, 3, 4)
    .retry(3)
    .subscribe(observer)
retryUntil
  • 发生异常时，返回值是false表示继续执行(重复发射数据)，true不再执行,但会调用onError方法;
var temp = 0
Observable.just(1, 2, 3, 4)
    .map {
        temp = it
        it
    }
    .retryUntil {
        temp > 3
    }
    .subscribe(observer)
retryWhen
  • 遇到错误时，将发生的错误传递给一个新的被观察者（Observable），并决定是否需要重新订阅原始被观察者（Observable）& 发送事件;
Observable.just(1, 2, 3.2, 4)
    .map { it.toInt() }
    .retryWhen {
        it.flatMap { throwable ->
            if (throwable is NumberFormatException) {
                Observable.error(Throwable("retryWhen终止"))
            } else {
                Observable.just(5, 6, 7)
            }
        }
    }
    .subscribe(observer)

6. 辅助操作:

delay
  • 延迟一段时间发射结果数据;
Observable.just(1, 2, 3)
    .delay(1,TimeUnit.SECONDS)
    .subscribe(observer)
do
  • 在某个事件的生命周期中调用;
  • doOnEach：数据源（Observable）每发送一次数据，就调用一次;
  • doOnNext：数据源每次调用onNext() 之前都会先回调该方法;
  • doOnError：数据源每次调用onError() 之前会回调该方法;
  • doOnComplete：数据源每次调用onComplete() 之前会回调该方法;
  • doOnSubscribe：数据源每次调用onSubscribe() 之后会回调该方法;
  • doOnDispose：数据源每次调用dispose() 之后会回调该方法;
Observable.just(1, 2, 3, 4)
    .observeOn(Schedulers.io())
    .subscribeOn(AndroidSchedulers.mainThread())
    .doOnSubscribe { LjyLogUtil.d("${Thread.currentThread().name}_doOnSubscribe") }
    .doOnEach { LjyLogUtil.d("${Thread.currentThread().name}_doOnEach:$it") }
    .doOnNext { LjyLogUtil.d("${Thread.currentThread().name}_doOnNext:$it") }
    .doOnError { LjyLogUtil.d("${Thread.currentThread().name}_doOnError:${it.localizedMessage}") }
    .doOnComplete { LjyLogUtil.d("${Thread.currentThread().name}_doOnComplete") }
    .doOnDispose { LjyLogUtil.d("${Thread.currentThread().name}_doOnDispose") }
    .subscribe(observer)

7. 条件和布尔操作:

all
  • 判断Observable发射的所有的数据项是否都满足某个条件;
Observable.just(1, -2, 3)
    .all {
        it > 0
    }.subscribe { it ->
        LjyLogUtil.d("${Thread.currentThread().name}_onSuccess:$it")
    }
takeWhile & skipWhile & takeUntil & skipUntil
// 从0开始每1s发送1个数据
Observable.interval(1, TimeUnit.SECONDS)
    //条件满足时，取数据
    .takeWhile {
        it<10
    }
Observable.interval(1, TimeUnit.SECONDS)
    //满足条件时，跳过数据
    .skipWhile {
        it<10
    }
Observable.interval(1, TimeUnit.SECONDS)
   //取数据，直到满足条件
    .takeUntil {
        it>10
    }
Observable.interval(1, TimeUnit.SECONDS)
    //等待直到传入的Observable开始发送数据
    .skipUntil (Observable.timer(5, TimeUnit.SECONDS))
SequenceEqual：
  • 判断两个Observable是否按相同的数据序列;
Observable.sequenceEqual(
    Observable.just(4, 5, 6),
    Observable.just(4, 5, 6)
).subscribe { it ->
    LjyLogUtil.d("${Thread.currentThread().name}_onSuccess:$it")
}
contains
  • 判断Observable是否会发射一个指定的数据项;
Observable.just(1, -2, 3)
    .contains(2)
    .subscribe { it ->
        LjyLogUtil.d("${Thread.currentThread().name}_onSuccess:$it")
    }
isEmpty
  • 判断发送的数据是否为空;
Observable.just(1, -2, 3)
    .isEmpty()
    .subscribe { it ->
        LjyLogUtil.d("${Thread.currentThread().name}_onSuccess:$it")
    }
amb

-给定多个Observable，只让第一个发射数据的Observable发射全部数据;

val list: MutableList<ObservableSource<Int>> = ArrayList()
list.add(Observable.just(1, 2, 3).delay(1, TimeUnit.SECONDS))
list.add(Observable.just(4, 5, 6))
Observable.amb(list).subscribe{
    LjyLogUtil.d("${Thread.currentThread().name}_onSuccess:$it")
}
defaultIfEmpty
  • 发射来自原始Observable的数据，如果原始Observable没有发射数据，就发射一个默认数据;
//在不发送onNext事件, 仅发送onComplete事件
Observable.empty<Int>()
    .defaultIfEmpty(-1)
    .subscribe { it ->
        LjyLogUtil.d("${Thread.currentThread().name}_onSuccess:$it")
    }
  • SkipUntil：丢弃原始Observable发射的数据，直到第二个Observable发射了一个数据，然后发射原始Observable的剩余数据;

实战

结合 RxBinding 使用

  • RxBinding是对 Android View 事件的扩展, 它使得开发者可以对 View 事件使用 RxJava 的各种操作；
1. 添加依赖
//RxBinding
implementation 'com.jakewharton.rxbinding4:rxbinding:4.0.0'
//Google 'material' library bindings:
implementation 'com.jakewharton.rxbinding4:rxbinding-material:4.0.0'
//AndroidX library bindings:
implementation 'com.jakewharton.rxbinding4:rxbinding-core:4.0.0'
implementation 'com.jakewharton.rxbinding4:rxbinding-appcompat:4.0.0'
implementation 'com.jakewharton.rxbinding4:rxbinding-drawerlayout:4.0.0'
implementation 'com.jakewharton.rxbinding4:rxbinding-leanback:4.0.0'
implementation 'com.jakewharton.rxbinding4:rxbinding-recyclerview:4.0.0'
implementation 'com.jakewharton.rxbinding4:rxbinding-slidingpanelayout:4.0.0'
implementation 'com.jakewharton.rxbinding4:rxbinding-swiperefreshlayout:4.0.0'
implementation 'com.jakewharton.rxbinding4:rxbinding-viewpager:4.0.0'
implementation 'com.jakewharton.rxbinding4:rxbinding-viewpager2:4.0.0'
2. 按钮防抖
val btn1 = findViewById<Button>(R.id.btn_1)
btn1.clicks()
    .throttleFirst(2, TimeUnit.SECONDS)
    .subscribeOn(AndroidSchedulers.mainThread())
    .subscribe {
        LjyLogUtil.d("点击按钮")
    }
3. editText输入监听
//也可用作联想搜索优化
val et1 = findViewById<EditText>(R.id.et_1)
et1.textChanges()
    .debounce(1, TimeUnit.SECONDS)
    //跳过第1次请求 因为初始输入框的空字符状态
    .skip(1)
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe{
        LjyLogUtil.d("${Thread.currentThread().name}_onSuccess:$it")
    }
4. 联合/表单判断
val etName = findViewById<EditText>(R.id.et_name)
val etPwd = findViewById<EditText>(R.id.et_pwd)
val obName = etName.textChanges()
val obPwd = etPwd.textChanges()
Observable.combineLatest(
    obName, obPwd, { name, pwd -> name == "ljy" && pwd == "123" })
    //跳过第1次请求 因为初始输入框的空字符状态
    .skip(1)
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe { isLogin -> LjyLogUtil.d(if (isLogin) "登录成功" else "登录失败") }
5. 定时器任务
val time = 10L
val btnLogin = findViewById<Button>(R.id.btn_login)
btnLogin.clicks()
    .throttleFirst(time, TimeUnit.SECONDS)
    .subscribeOn(AndroidSchedulers.mainThread())
    .doOnNext { btnLogin.isEnabled = false }
    .subscribe {
        LjyLogUtil.d("点击登录")
        Observable.intervalRange(
            0, time, 0, 1,
            TimeUnit.SECONDS, AndroidSchedulers.mainThread()
        )
            .subscribe(
                { btnLogin.text = "剩余${time - it}秒" },
                { LjyLogUtil.e(it.message) },
                {
                    btnLogin.text = "获取验证码"
                    btnLogin.isEnabled = true
                })
    }

利用RxLifecycle解决内存泄漏问题

1. 添加依赖
//RxLifecycle
    implementation 'com.trello.rxlifecycle4:rxlifecycle:4.0.2'
// If you want to bind to Android-specific lifecycles
    implementation 'com.trello.rxlifecycle4:rxlifecycle-android:4.0.2'
// If you want pre-written Activities and Fragments you can subclass as providers
    implementation 'com.trello.rxlifecycle4:rxlifecycle-components:4.0.2'
// If you want pre-written support preference Fragments you can subclass as providers
    implementation 'com.trello.rxlifecycle4:rxlifecycle-components-preference:4.0.2'
// If you want to use Android Lifecycle for providers
    implementation 'com.trello.rxlifecycle4:rxlifecycle-android-lifecycle:4.0.2'
// If you want to use Kotlin syntax
    implementation 'com.trello.rxlifecycle4:rxlifecycle-kotlin:4.0.2'
// If you want to use Kotlin syntax with Android Lifecycle
    implementation 'com.trello.rxlifecycle4:rxlifecycle-android-lifecycle-kotlin:4.0.2'
2. 绑定
Observable.intervalRange(0, 60, 0, 1, TimeUnit.SECONDS)
    //利用RxLifecycle来解决内存泄漏问题
    //bindToLifecycle的自动取消订阅示例
    //.compose(bindToLifecycle())
    //手动设置在activity onPause的时候取消订阅
    .compose(this.bindUntilEvent(ActivityEvent.PAUSE))
    .subscribe(
        { LjyLogUtil.d("剩余${60 - it}秒") },
        { LjyLogUtil.e(it.message) },
        { LjyLogUtil.d("完成") }
    )

网络请求

1. 网络请求嵌套回调
//使用observeOn多次切换线程
apiService.searchRepo(emptyMap()) // 请求搜索列表
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .doOnNext {
        //展示列表
    }
    .observeOn(Schedulers.io())
    .flatMap {
        Observable.fromIterable(it.items).map { repo -> repo.id }
    }
    .flatMap {
        apiService.getItem(it) // 请求详情
    }
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe {
        //展示详情
    }
2. 网络请求轮询
//每60秒一次，无限轮询
Observable.interval(60, TimeUnit.SECONDS)
    .doOnNext {
        apiService.searchRepo(emptyMap())
            .subscribe(Consumer { LjyLogUtil.d("accept: ${it.items}") })
    }.subscribe { LjyLogUtil.d("第 $it 次轮询") }
//有条件轮询
var count = 0
apiService.searchRepo(emptyMap())
    .repeatWhen {
        it.flatMap {
            if (count > 3) {
                Observable.error(Throwable("轮询结束"))
            } else {
                LjyLogUtil.d("第 $count 次轮询")
                Observable.just(1).delay(60, TimeUnit.SECONDS)
            }
        }
    }
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe({
        LjyLogUtil.d("accept: ${it.items}")
        count++
    }, {
        LjyLogUtil.d(it.message)
    })
3. 网络请求出错重连
//最大重试次数
val maxConnectCount = 10
//已重试次数
var currentRetryCount = 0
apiService.searchRepo(emptyMap())
    .retryWhen {
        it.flatMap { throwable ->
            //根据异常类型选择是否重试
            if (throwable is IOException) {
                if (currentRetryCount < maxConnectCount) {
                    currentRetryCount++
                    //遇到的异常越多，重试延迟间隔时间越长
                    Observable.just(1).delay(currentRetryCount * 60L, TimeUnit.SECONDS)
                } else {
                    Observable.error(Throwable("重试结束"))
                }
            } else {
                Observable.error(Throwable("发生了非网络异常（非I/O异常）"))
            }
        }
    }
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe({
        LjyLogUtil.d("accept: ${it.items}")
    }, {
        LjyLogUtil.d(it.message)
    })
4. 内存，磁盘，网络 三级缓存
var memoryCache: String? = null
var diskCache: String? = null
private fun test10() {
    val memory = Observable.create<String> {
        if (memoryCache != null) {
            it.onNext(memoryCache)
        } else {
            it.onComplete()
        }
    }

    val disk = Observable.create<String> {
        if (diskCache != null) {
            it.onNext(diskCache)
        } else {
            it.onComplete()
        }
    }.doOnNext {
        memoryCache = "内存数据"
    }

    val net = Observable.just("网络数据")
        .doOnNext {
            // memoryCache="内存数据"
            diskCache = "磁盘数据"
        }
    //通过concat（）合并memory、disk、network 3个被观察者的事件（即检查内存缓存、磁盘缓存 & 发送网络请求）
    Observable.concat(memory, disk, net)
        //通过firstElement()，从串联队列中取出并发送第1个有效事件（Next事件），即依次判断检查memory、disk、network
        .firstElement()
        .subscribe {
            LjyLogUtil.d("accept: $it")
        }
}
5. 合并数据源 & 同时展示
var result = "数据来自："
val net = Observable.just("网络")
val disk = Observable.just("磁盘")
//使用merge，从网络和本地获取数据并展示
Observable.merge(net, disk)
    .subscribe({
        result += "$it, "
    }, {

    }, {
        LjyLogUtil.d("result: $result")
    })
//使用zip,合并2个网络请求向获取数据并展示
val repo1 = apiService.getItem(1001).subscribeOn(Schedulers.io())
val repo2 = apiService.getItem(1002).subscribeOn(Schedulers.io())
Observable.zip(
    repo1, repo2, { data1, data2 ->
        val repoList = ArrayList<RepoDetail>()
        repoList.add(data1)
        repoList.add(data2)
        repoList
    })
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe {
        for (repoDetail in it) {
            LjyLogUtil.d("result: ${repoDetail.name}")
        }
    }

源码解析

入口

  • 以下面代码为源码阅读入口，Single是最简单的被观察者；
Single.just(1)
    .subscribe(object : SingleObserver<String> {
        override fun onSubscribe(d: Disposable) {
            LjyLogUtil.d("${Thread.currentThread().name}_onSubscribe")
        }

        override fun onSuccess(t: String) {
            LjyLogUtil.d("${Thread.currentThread().name}_onSuccess:$t")
        }

        override fun onError(e: Throwable) {
            LjyLogUtil.d("${Thread.currentThread().name}_onError:${e.message}")
        }
    })

被观察者的创建

  • 先看上面代码中被观察者的创建方法：Single.just()；
public static <@NonNull T> Single<T> just(T item) {
    Objects.requireNonNull(item, "item is null");//判空
    return RxJavaPlugins.onAssembly(new SingleJust<>(item));
}
  • 其中第一行为判空，第二行RxJavaPlugins.onAssembly为一个钩子方法,方便添加一些额外操作，代码如下：
@Nullable
static volatile Function<? super Single, ? extends Single> onSingleAssembly;

public static <T> Single<T> onAssembly(@NonNull Single<T> source) {
    Function<? super Single, ? extends Single> f = onSingleAssembly;
    if (f != null) {
        return apply(f, source);
    }
    return source;
}
  • 其中onSingleAssembly默认为空，所以实际默认是直接返回source，也就是SingleJust；
  • 所以Single.just就是直接创建了一个被观察者的实现类SingleJust的实例并返回，那么我们看一下SingleJust的代码；
public final class SingleJust<T> extends Single<T> {
    final T value;

    public SingleJust(T value) {
        this.value = value;
    }

    @Override
    protected void subscribeActual(SingleObserver<? super T> observer) {
        observer.onSubscribe(Disposable.disposed());
        observer.onSuccess(value);
    }
}
  • 很简单吧，继承了Single，并重写了Single唯一的抽象方法subscribeActual，有个泛型变量value;
  • Single的代码很多，基本都是上面介绍的操作符方法的实现，用到时可以点进源码看一看；
public abstract class Single<@NonNull T> implements SingleSource<T> {
     ...
     protected abstract void subscribeActual(@NonNull SingleObserver<? super T> observer);
     ...
      public final void subscribe(@NonNull SingleObserver<? super T> observer) {
         ...
      }
}
  • Single实现了SingleSource接口，并实现subscribe方法；
public interface SingleSource<@NonNull T> {

    /**
     * Subscribes the given {@link SingleObserver} to this {@link SingleSource} instance.
     * @param observer the {@code SingleObserver}, not {@code null}
     * @throws NullPointerException if {@code observer} is {@code null}
     */
    void subscribe(@NonNull SingleObserver<? super T> observer);
}

订阅

  • 下面来看看开头例子中的第二行的subscribe方法：Single.subscribe()；
public final void subscribe(@NonNull SingleObserver<? super T> observer) {
    //判空
    Objects.requireNonNull(observer, "observer is null");
    //钩子方法，默认还是入参的SingleObserver
    observer = RxJavaPlugins.onSubscribe(this, observer);
    //判空
    Objects.requireNonNull(observer, "The RxJavaPlugins.onSubscribe hook returned a null SingleObserver. Please check the handler provided to RxJavaPlugins.setOnSingleSubscribe for invalid null returns. Further reading: https://github.com/ReactiveX/RxJava/wiki/Plugins");

    try {
        subscribeActual(observer);
    } catch (NullPointerException ex) {
        throw ex;
    } catch (Throwable ex) {
        Exceptions.throwIfFatal(ex);
        NullPointerException npe = new NullPointerException("subscribeActual failed");
        npe.initCause(ex);
        throw npe;
    }
}
  • 可以看到订阅方法subscribe中最有用的一行就是调用了subscribeActual，而这里的subscribeActual的实现正式上面的SingleJust中的实现；
@Override
protected void subscribeActual(SingleObserver<? super T> observer) {
    observer.onSubscribe(Disposable.disposed());
    observer.onSuccess(value);
}
  • 可以看到SingleJust中的subscribeActual直接连着调用了observer的onSubscribe和onSuccess方法，并且onSubscribe的入参为一个disposed的Disposable,是个已取消的Disposable，因为just是没有延迟的，无需取消和修改，而且onError是不会被调用的，因为瞬发一个也不会有出错的可能；

操作符实现

map

  • 以map为例，对之前的示例扩展一下，增加个map操作符对数据进行转换,map搞懂了，其他操作符的主要思路也是一样的；
Single.just(1)
    .map { "num_$it" }
    .subscribe { num ->
        LjyLogUtil.d(num)
    }
  • 那么看一下map的实现：
public final <@NonNull R> Single<R> map(@NonNull Function<? super T, ? extends R> mapper) {
    Objects.requireNonNull(mapper, "mapper is null");
    return RxJavaPlugins.onAssembly(new SingleMap<>(this, mapper));
}
  • map中也是有判空和钩子，创建了一个SingleMap并返回，其入参第一次参数为this，也就是just返回的SingleJust实例，第二个参数为数据转换的转换器；
  • 那么来看一下SingleMap：
public final class SingleMap<T, R> extends Single<R> {
    final SingleSource<? extends T> source;

    final Function<? super T, ? extends R> mapper;

    public SingleMap(SingleSource<? extends T> source, Function<? super T, ? extends R> mapper) {
        this.source = source;
        this.mapper = mapper;
    }

    @Override
    protected void subscribeActual(final SingleObserver<? super R> t) {
        source.subscribe(new MapSingleObserver<T, R>(t, mapper));
    }
    ...//省略内部类MapSingleObserver
}
  • 可以看到SingleMap的subscribeActual中只有一行代码，就是调用source的订阅方法subscribe，并传入一个观察者MapSingleObserver实例和SingleMap持有的数据转换器，其中source也就是前一步的被观察者SingleJust实例；
  • 而观察者MapSingleObserver则是负责对上下游数据进行转换 和传递，其入参下游subscribe传入的观察者SingleObserver实例，来看一下它的实现：
static final class MapSingleObserver<T, R> implements SingleObserver<T> {
    final SingleObserver<? super R> t;
    final Function<? super T, ? extends R> mapper;

    MapSingleObserver(SingleObserver<? super R> t, Function<? super T, ? extends R> mapper) {
        this.t = t;
        this.mapper = mapper;
    }

    @Override
    public void onSubscribe(Disposable d) {
        t.onSubscribe(d);
    }

    @Override
    public void onSuccess(T value) {
        R v;
        try {
            //mapper.apply数据转换器对数据进行转换
            v = Objects.requireNonNull(mapper.apply(value), "The mapper function returned a null value.");
        } catch (Throwable e) {
            Exceptions.throwIfFatal(e);
            onError(e);
            return;
        }

        t.onSuccess(v);
    }

    @Override
    public void onError(Throwable e) {
        t.onError(e);
    }
}
  • 可以看到MapSingleObserver的onSuccess中调用数据转换器mapper的apply方法对数据进行转换，
    并调用构造函数传入的SingleObserver的onSuccess将转换后的数据传递过去，而onSubscribe，onError方法则是直接调用了SingleObserver的方法
小结
  • 那么示例代码中map前后整个流程串起来就是：
  1. Single.just创建了一个被观察者SingleJust,负责发送原始数据；
  2. map创建了一个被观察者SingleMap，构造函数入参是上一步的SingleJust和数据转换器，也就是对SingleJust进行了包装，或者说接管，如何接管的呢，SingleMap中调用了SingleJust的订阅方法subscribe，并传入观察者MapSingleObserver，也就是对SingleJust说，小j啊，你这项目给我（SingleMap）了，后面的客户你不用跟了，把你手里的数据给我小弟MapSingleObserver就行了，以后需要啥他找你要；
  3. 后面再调用subscribe呢，实际是调用SingleMap的subscribe，也就是后面再有客户其实都是跟SingleMap签的单子了
  4. SingleMap收到单子呢转手就给小弟MapSingleObserver了，反正SingleJust的数据也在你手里呢你看看咋给客户服务吧
  5. MapSingleObserver就拿着SingleJust的数据进行转换，然后调用客户（下游subscribe传入的观察者）的回调方法进行服务
  • ps：
    • MapSingleObserver：同样是观察者，我也太累了，又负责接收上游的数据，又要进行处理，处理完还得反馈给下游的客户；
    • SingleJust：你累？同样是被观察者，我辛辛苦苦做的项目，还不是被SingleMap拿走了，还让你监视（观察）我，这点业绩都被内部消化了啊，而且项目开始了还把我工号留给客户（ t.onSubscribe(d)），那客户是说停就停啊，我这出错了你也是转头就给客户说啊（ t.onError(e)）；
    • 这哪是RxJava，这分明是职场啊；

subscribeOn & observeOn

  • 在对之前的示例扩展一下，加入线程切换，那就是用到了subscribeOn & observeOn 操作符了；
Single.just(1)
    .map { "num_$it" }
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe { num ->
        LjyLogUtil.d(num)
    }
subscribeOn
  • 先来看看subscribeOn的入参Schedulers.io()：
public static Scheduler io() {
    return RxJavaPlugins.onIoScheduler(IO);
}

static {
    SINGLE = RxJavaPlugins.initSingleScheduler(new SingleTask());

    COMPUTATION = RxJavaPlugins.initComputationScheduler(new ComputationTask());

    IO = RxJavaPlugins.initIoScheduler(new IOTask());

    TRAMPOLINE = TrampolineScheduler.instance();

    NEW_THREAD = RxJavaPlugins.initNewThreadScheduler(new NewThreadTask());
}

static final class IOTask implements Supplier<Scheduler> {
    @Override
    public Scheduler get() {
        return IoHolder.DEFAULT;
    }
}

static final class IoHolder {
    static final Scheduler DEFAULT = new IoScheduler();
}

public final class IoScheduler extends Scheduler {
    ...
}
  • 可以看到最后返回的是Scheduler的实现类IoScheduler实例；
  • 再来看看subscribeOn方法的实现：
public final Single<T> subscribeOn(@NonNull Scheduler scheduler) {
    Objects.requireNonNull(scheduler, "scheduler is null");
    return RxJavaPlugins.onAssembly(new SingleSubscribeOn<>(this, scheduler));
}
  • 还是判空和钩子，然后创建一个SingleSubscribeOn并返回，入参是上游的被观察者（this），和线程调度器Scheduler；
public final class SingleSubscribeOn<T> extends Single<T> {
    final SingleSource<? extends T> source;

    final Scheduler scheduler;

    public SingleSubscribeOn(SingleSource<? extends T> source, Scheduler scheduler) {
        this.source = source;
        this.scheduler = scheduler;
    }

    @Override
    protected void subscribeActual(final SingleObserver<? super T> observer) {
        //创建一个观察者SubscribeOnObserver的实例，传入上游的被观察者source
        final SubscribeOnObserver<T> parent = new SubscribeOnObserver<>(observer, source);
        //调用下游的观察者observer的onSubscribe，通知其开始订阅了，取消的话可以找SubscribeOnObserver
        observer.onSubscribe(parent);
        //调用IoScheduler的scheduleDirect方法切换线程
        Disposable f = scheduler.scheduleDirect(parent);

        parent.task.replace(f);

    }
    ...//省略内部类SubscribeOnObserver
}
  • 上面subscribeActual中创建了一个被观察者SubscribeOnObserver，并且调用scheduler.scheduleDirect切换线程；
public Disposable scheduleDirect(@NonNull Runnable run) {
    return scheduleDirect(run, 0L, TimeUnit.NANOSECONDS);
}

public Disposable scheduleDirect(@NonNull Runnable run, long delay, @NonNull TimeUnit unit) {
    final Worker w = createWorker();
    final Runnable decoratedRun = RxJavaPlugins.onSchedule(run);
    DisposeTask task = new DisposeTask(decoratedRun, w);
    w.schedule(task, delay, unit);
    return task;
}

//其中DisposeTask为Scheduler的内部类
static final class DisposeTask implements Disposable, Runnable, SchedulerRunnableIntrospection {
    ...
    DisposeTask(@NonNull Runnable decoratedRun, @NonNull Worker w) {
        this.decoratedRun = decoratedRun;
        this.w = w;
    }

    @Override
    public void run() {
        runner = Thread.currentThread();
        try {
            try {
                decoratedRun.run();
            } catch (Throwable ex) {
                // Exceptions.throwIfFatal(e); nowhere to go
                RxJavaPlugins.onError(ex);
                throw ex;
            }
        } finally {
            dispose();
            runner = null;
        }
    }
    ...
}
  • 其中scheduleDirect的入参是个Runnable，而SubscribeOnObserver实现了Runnable接口和SingleObserver接口,构造方法入参是下游的观察者和上游的被观察者，在自己的onSuccess,onError中直接传给下游的观察者，在run方法中订阅上游的被观察者;
static final class SubscribeOnObserver<T>
    extends AtomicReference<Disposable>
    implements SingleObserver<T>, Disposable, Runnable {
    ...
    SubscribeOnObserver(SingleObserver<? super T> actual, SingleSource<? extends T> source) {
        this.downstream = actual;
        this.source = source;
        this.task = new SequentialDisposable();
    }
    ...
    @Override
    public void onSuccess(T value) {
        downstream.onSuccess(value);
    }

    @Override
    public void onError(Throwable e) {
        downstream.onError(e);
    }
    ...
    @Override
    public void run() {
        source.subscribe(this);
    }
}
observeOn
  • 先来看看Android专用的线程调度器AndroidSchedulers.mainThread():
//通过静态内部类提供HandlerScheduler的单例
public final class AndroidSchedulers {
    //开放的入口方法
    public static Scheduler mainThread() {
        //钩子
        return RxAndroidPlugins.onMainThreadScheduler(MAIN_THREAD);
    }
    //私有的静态变量
    private static final Scheduler MAIN_THREAD =
          RxAndroidPlugins.initMainThreadScheduler(() -> MainHolder.DEFAULT);
    //私有的静态内部类
    private static final class MainHolder {
        //本质还说离不开Handler,通过Looper.getMainLooper()切换到UI线程
        static final Scheduler DEFAULT
            = new HandlerScheduler(new Handler(Looper.getMainLooper()), true);
    }
    //私有的构造方法，并抛出异常，因为这里是创建HandlerScheduler的单例，而不是AndroidSchedulers本身
    private AndroidSchedulers() {
        throw new AssertionError("No instances.");
    }
}

final class HandlerScheduler extends Scheduler {
    private final Handler handler;
    private final boolean async;

    HandlerScheduler(Handler handler, boolean async) {
        this.handler = handler;
        this.async = async;
    }

    public Disposable scheduleDirect(Runnable run, long delay, TimeUnit unit) {
        ...//省略判空和钩子方法
        ScheduledRunnable scheduled = new ScheduledRunnable(handler, run);
        Message message = Message.obtain(handler, scheduled);
        if (async) {
            message.setAsynchronous(true);
        }
        //通过主线程的handler发送消息
        handler.sendMessageDelayed(message, unit.toMillis(delay));
        return scheduled;
    }

    @Override
    public Worker createWorker() {
        return new HandlerWorker(handler, async);
    }

    private static final class HandlerWorker extends Worker {
        private final Handler handler;
        private final boolean async;

        private volatile boolean disposed;

        HandlerWorker(Handler handler, boolean async) {
            this.handler = handler;
            this.async = async;
        }

        public Disposable schedule(Runnable run, long delay, TimeUnit unit) {
            ...
            if (disposed) {
                return Disposable.disposed();
            }
            ...
            ScheduledRunnable scheduled = new ScheduledRunnable(handler, run);
            Message message = Message.obtain(handler, scheduled);
            message.obj = this;
            ...
            handler.sendMessageDelayed(message, unit.toMillis(delay));

            if (disposed) {
                handler.removeCallbacks(scheduled);
                return Disposable.disposed();
            }

            return scheduled;
        }

        @Override
        public void dispose() {
            disposed = true;
            handler.removeCallbacksAndMessages(this /* token */);
        }

        @Override
        public boolean isDisposed() {
            return disposed;
        }
    }

    private static final class ScheduledRunnable implements Runnable, Disposable {
        private final Handler handler;
        private final Runnable delegate;

        private volatile boolean disposed;

        ScheduledRunnable(Handler handler, Runnable delegate) {
            this.handler = handler;
            this.delegate = delegate;
        }

        @Override
        public void run() {
            try {
                delegate.run();
            } catch (Throwable t) {
                RxJavaPlugins.onError(t);
            }
        }

        @Override
        public void dispose() {
            handler.removeCallbacks(this);
            disposed = true;
        }

        @Override
        public boolean isDisposed() {
            return disposed;
        }
    }
}
  • 然后来看observeOn方法:
public final Single<T> observeOn(@NonNull Scheduler scheduler) {
    Objects.requireNonNull(scheduler, "scheduler is null");
    return RxJavaPlugins.onAssembly(new SingleObserveOn<>(this, scheduler));
}
  • 本质还是创建了一个被观察者:
public final class SingleObserveOn<T> extends Single<T> {

    final SingleSource<T> source;

    final Scheduler scheduler;

    public SingleObserveOn(SingleSource<T> source, Scheduler scheduler) {
        this.source = source;
        this.scheduler = scheduler;
    }

    @Override
    protected void subscribeActual(final SingleObserver<? super T> observer) {
        //调用上游被观察者source的订阅方法subscribe，传入一个观察者ObserveOnSingleObserver，其入参是下游的观察者observer
        source.subscribe(new ObserveOnSingleObserver<>(observer, scheduler));
    }
    ...//省略内部类ObserveOnSingleObserver
}
  • ObserveOnSingleObserver代码如下, 在自己的onSubscribe调用下游观察者的onSubscribe，在自己的onSuccess和onError中调用scheduler.scheduleDirect切换线程，scheduleDirect的入参是Runnable，会调用自己的run方法，而在run方法中调用了下游观察者的onSuccess和onError传递结果，以此实现改变下游观察者的线程;
static final class ObserveOnSingleObserver<T> extends AtomicReference<Disposable>
implements SingleObserver<T>, Disposable, Runnable {
    ...
    ObserveOnSingleObserver(SingleObserver<? super T> actual, Scheduler scheduler) {
        this.downstream = actual;
        this.scheduler = scheduler;
    }

    @Override
    public void onSubscribe(Disposable d) {
        if (DisposableHelper.setOnce(this, d)) {
            downstream.onSubscribe(this);
        }
    }

    @Override
    public void onSuccess(T value) {
        this.value = value;
        Disposable d = scheduler.scheduleDirect(this);
        DisposableHelper.replace(this, d);
    }

    @Override
    public void onError(Throwable e) {
        this.error = e;
        Disposable d = scheduler.scheduleDirect(this);
        DisposableHelper.replace(this, d);
    }

    @Override
    public void run() {
        Throwable ex = error;
        if (ex != null) {
            downstream.onError(ex);
        } else {
            downstream.onSuccess(value);
        }
    }
   ...
}

番外：Single.create

  • 看了上面的Single.just, 可能有种被忽悠的感觉，也太low了吧，这么简单么，那么我们不妨再来看看Single.create：
public static <@NonNull T> Single<T> create(@NonNull SingleOnSubscribe<T> source) {
    Objects.requireNonNull(source, "source is null");
    return RxJavaPlugins.onAssembly(new SingleCreate<>(source));
}
  • 同样的create方法中创建了一个被观察者SingleCreate并返回：
public final class SingleCreate<T> extends Single<T> {

    final SingleOnSubscribe<T> source;

    public SingleCreate(SingleOnSubscribe<T> source) {
        this.source = source;
    }

    @Override
    protected void subscribeActual(SingleObserver<? super T> observer) {
        Emitter<T> parent = new Emitter<>(observer);
        observer.onSubscribe(parent);

        try {
            source.subscribe(parent);
        } catch (Throwable ex) {
            Exceptions.throwIfFatal(ex);
            parent.onError(ex);
        }
    }
    ...//省略了内部类Emitter
}
  • 看了上面代码，create和just有什么区别呢，just{}中的代码是直接赋值给value的，不需要下游subscribe就会执行，而create是通过匿名内部类传入一个SingleOnSubscribe的实例,并在实现的subscribe方法中产生数据，在SingleCreate.subscribeActual中调用SingleOnSubscribe.subscribe，SingleCreate.subscribeActual则是在下游subscribe时才会执行，也就是Single.create在下游订阅后才开始产生数据；那么显然，Observable.just()不适合封装网络数据，因为我们通常不想在subscribe之前做网络请求。

参考

我是今阳，如果想要进阶和了解更多的干货，欢迎关注微信公众号 “今阳说” 接收我的最新文章
  移动开发 最新文章
Vue3装载axios和element-ui
android adb cmd
【xcode】Xcode常用快捷键与技巧
Android开发中的线程池使用
Java 和 Android 的 Base64
Android 测试文字编码格式
微信小程序支付
安卓权限记录
知乎之自动养号
【Android Jetpack】DataStore
上一篇文章      下一篇文章      查看所有文章
加:2021-10-01 17:00:30  更:2021-10-01 17:02:35 
 
开发: C++知识库 Java知识库 JavaScript Python PHP知识库 人工智能 区块链 大数据 移动开发 嵌入式 开发工具 数据结构与算法 开发测试 游戏开发 网络协议 系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑 笔记本 显卡 显示器 固态硬盘 硬盘 耳机 手机 iphone vivo oppo 小米 华为 单反 装机 图拉丁

360图书馆 购物 三丰科技 阅读网 日历 万年历 2024年11日历 -2024/11/23 19:39:59-

图片自动播放器
↓图片自动播放器↓ 		TxT小说阅读器
↓语音阅读,小说下载,古典文学↓ 		一键清除垃圾
↓轻轻一点,清除系统垃圾↓ 		图片批量下载器
↓批量下载图片,美女图库↓
  网站联系: qq:121756557 email:121756557@qq.com  IT数码