RxKotlin 例子不超过15行教程 4----Observ

作者: 蔡洪宇 | 来源:发表于2018-02-24 20:53 被阅读250次

本节代码中的 observer 就是第二节中的

Observer 与 Subscribe

按照惯例,先来看两段能跑的代码

两段能跑的代码

// 4.1.kt
import io.reactivex.Observable

fun main(args: Array<String>) {
    val observable: Observable<Int> = Observable.range(1, 3)
    observable.subscribe({  // 我知道你要问我为什么 subscribe 后面还可以接三个 Lambda,先看例子,下面说
        //onNext method
        println("Next $it")
    }, {
        //onError Method
        println("Error ${it.message}")
    }, {
        //onComplete Method
        println("Done")
    })
}

输出

Next 1
Next 2
Next 3
All Completed

再来一段(上一节用过的例子)

// 3.11.kt
import io.reactivex.Observable

fun main(args: Array<String>) {
    Observable.range(1, 3).subscribe(observer)
}

输出

New Subscription  // 上面那个例子没有这一行
Next 1
Next 2
Next 3
All Completed

上面代码的主旨

Observer

之前(第二节)我们说过,一个 Observer 需要实现四个方法(它们的作用参见第二节)

onNext
onComplete
onError
onSubscribe

当我们把 Observable 连接到 Observer 上的时候,系统会调用这四个方法并把相应的值传给它们。

subscribe 的参数都能是什么

subscribe 在 ReactiveX 中有几个重载方法,这里不列出。基本模式有这两个

subscribe(onNext,onError,onComplete,onSubscribe)
这几个参数都可以省略,但是只能从后往前省略(这句是废话)
是废话也要说,因为 subscribe 是在 Java 文件中定义的,不能使用 Kotlin 的命名参数
4.1.kt 中省略了 onSubscribe
subscribe(observer)
3.11.kt 已经很清晰,这里不展开了

除了 `subscribe` 方法,还有 RxKotlin 提供的小语法糖 `subscribeBy`

这个函数是 RxKotlin 为 Observable (等可以 subscribe 的对象)定义的扩展函数,函数定义如下

fun <T : Any> Observable<T>.subscribeBy(
        onError: (Throwable) -> Unit = onErrorStub,
        onComplete: () -> Unit = onCompleteStub,
        onNext: (T) -> Unit = onNextStub
        ): Disposable = subscribe(onNext, onError, onComplete)  // 好的好的,我知道你要问 Disposable 是什么,稍等。

因为被定义在 Kotlin 文件中,它可以使用命名参数(例子见第一节 1.kt)

从之前的例子可知,subscribe 可以连接 Observable 与 Observer。
它有两种形式(上面说过,这里再概括一下)

把 onNext 等,以参数的形式传进去
直接传入一个 Observer 对象

如果你选择第一种形式,那么 subscribe 方法是有返回值的,返回值类型是 Disposable (不要急,它的介绍马上就到了)
如果你选择第二种形式,那么 subscribe 方法是没有返回值的
这两种形式中的 onSubscribe 都是一个 (d:Disposable):Unit 类型的函数。
那么 Disposable 有什么用呢?

Disposable

disposable: 一次性的，可任意处理的; 用后就抛弃的; 免洗的; 可供使用的。讲真,这几个中文翻译放在这里我觉得并不是很合适,我也没有想到合适的中文翻译(如果有合适的欢迎指出)。我就一直用英文了。
Disposable 对象的 dispose 方法可以停止本次订阅
看一个例子
我保证这个例子是为数不多的长例子之一,真的不能再精简了
下面示例用到了 lateinit 可以自行 Google 下,此处不介绍。(如果有好的链接欢迎发给我,加在这里)

// 4.2.kt
import io.reactivex.Observable
import io.reactivex.Observer
import io.reactivex.disposables.Disposable
import java.util.concurrent.TimeUnit

fun main(args: Array<String>) {

    val observale: Observable<Long> = Observable.interval(100, TimeUnit.MILLISECONDS)

    val observer: Observer<Long> = object : Observer<Long> {
        lateinit var disposable: Disposable

        override fun onSubscribe(d: Disposable) {
            disposable = d
        }

        override fun onNext(item: Long) {
            if (item >= 5 && !disposable.isDisposed) {
                disposable.dispose()
                println("Disposed")
            }
            println("Received $item")
        }

        override fun onError(e: Throwable) {
            println("Error ${e.message}")
        }

        override fun onComplete() {
            println("Complete")
        }

    }

    observale.subscribe(observer)
    Thread.sleep(1000)
}

输出

Received 0
Received 1
Received 2
Received 3
Received 4
Disposed    // 注释1
Received 5 // 注释2
// 注释3

注释1
dispose 处理后不会执行 observer 的 onComplete 方法(所以 Complete 没有输出)
注释2
disposable.dispose() 之后,observer 不会再处理其它值(所以 Received 6 Received 7 等等并没有输出)
但是当前值依然会继续处理(所以 Received 5 依然被输出)
总结一下, Disposable 是用来控制订阅的

下面我们回到 Observable,看看它的分类

Hot/Cold Observable

在本教程前面所有示例中，如果多次订阅同一个 Observable,则所有订阅都会得到从一开始的所有值。
例子

// 4.3.kt
import io.reactivex.Observable
import io.reactivex.rxkotlin.toObservable

// Cold Observables
fun main(args: Array<String>) {
    val observable: Observable<Int> = listOf(1, 2, 3, 4).toObservable()

    observable.subscribe(observer)

    observable.subscribe(observer)
}

输出

New Subscription
Next 1
Next 2
Next 3
Next 4
All Completed
New Subscription
Next 1
Next 2
Next 3
Next 4
All Completed

我们可以看到每一个 Observer 都被推送了了从 1-4 的所有值。
到目前为止,我们遇到的所有 Observable 都是这样的。
这样的 Observable 被称作 Cold Observable
我之前曾经比喻 Observable 为电台，这是有一些不恰当的。因为当你错过时间再打开电台会听不到原先的内容。
Cold Observable 更像是光盘(容量可能无限),随时打开都能从头开始听。

电台这个比喻更适合 Hot Observable,看下一个例子

// 4.4.kt
import io.reactivex.rxkotlin.toObservable

//Hot Observable
fun main(args: Array<String>) {
    val connectableObservable = listOf(1, 2, 3).toObservable().publish()  // 注释1
    connectableObservable.subscribe({ println("Subscription 1: $it") })  // 描点1
    connectableObservable.subscribe({ println("Subscription 2: $it") })  // 描点2
    connectableObservable.connect()  // 注释2
    connectableObservable.subscribe({ println("Subscription 3: $it") })  // 注释3
}

输出

Subscription 1: 1
Subscription 2: 1
Subscription 1: 2
Subscription 2: 2
Subscription 1: 3
Subscription 2: 3
// 并没有输出 Subscription 3

注释1
我们用 publish 方法把 Cold Observable 变成 ConnectableObservable (ConnectableObservable 是 Hot Observable 的一种)
注释2
ConnectableObservable 在 描点1 和 描点2 处都不会发送消息,它会在 注释2 处(调用 connect 方法时)开始发送消息
而 Cold Observable 会在调用 subscribe 时开始发送消息
如果订阅晚了(如 注释3),则会错过一些消息(在这里,注释3 错过了所有消息(计算机速度太快....),接下来有其他例子,不要急)
注释3
订阅3不会收到任何信息

我们来看下一个例子,在这个例子中,调用 connect 方法后我们又增加了新的订阅,这个订阅会丢失部分消息

import io.reactivex.Observable
import java.util.concurrent.TimeUnit

fun main(args: Array<String>) {
    val connectableObservable = Observable.interval(10, TimeUnit.MILLISECONDS).publish()
    connectableObservable.subscribe({ println("Subscription 1: $it") })
    connectableObservable.subscribe({ println("Subscription 2: $it") })
    connectableObservable.connect()  // ConnectableObservable 开始发送消息
    println("Sleep 1 starts")
    Thread.sleep(20)
    println("Sleep 1 ends")
    connectableObservable.subscribe({ println("Subscription 3: $it") })  // 不用再次调用 connect 方法
    println("Sleep 2 starts")
    Thread.sleep(30)
    println("Sleep 2 ends")
}

输出(有点长)

Sleep 1 starts
Subscription 1: 0
Subscription 2: 0
Subscription 1: 1
Subscription 2: 1 // 注释1
Sleep 1 ends      // 开始 订阅3
Sleep 2 starts
Subscription 1: 2
Subscription 2: 2
Subscription 3: 2  // 注释2
Subscription 1: 3
Subscription 2: 3
Subscription 3: 3
Subscription 1: 4
Subscription 2: 4
Subscription 3: 4
Sleep 2 ends

注释1
到这里我们没有开始 订阅3 所以没有输出任何 Subscription 3
注释2
订阅3 的输出是从 2 开始的,它错过了 0 和 1

这一节到这里就 OK 了,明天说 Subject

RxKotlin 例子不超过15行教程 1----环境配置与初体验

RxKotlin 例子不超过15行教程 2----Observable Observer 与 Subscribe 简介

RxKotlin 例子不超过15行教程 3----Observable 的创建

RxKotlin 例子不超过15行教程 4----Observer Subscribe 与 Hot/Cold Observable

RxKotlin 例子不超过15行教程 5----Subject

RxKotlin 例子不超过15行教程 6----Operator 与 Marble Diagram

RxKotlin 例子不超过15行教程 7----Backpressure Flowable 与 Subscriber 简介

RxKotlin 例子不超过15行教程 8----Error Handling

RxKotlin 例子不超过15行教程 4----Observ

Observer 与 Subscribe

两段能跑的代码

上面代码的主旨

Observer

subscribe 的参数都能是什么

除了 `subscribe` 方法,还有 RxKotlin 提供的小语法糖 `subscribeBy`

Subscribe

Disposable

Hot/Cold Observable

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

禅与计算机程序设计艺术

RxKotlin

Kotlin

程序员

反应式编程 · Reactive Programming · Reactor

Reactivex RxJava RxKotlin RxJS RxGo RxSwift

RxKotlin 例子不超过15行教程 4----Observ

Observer 与 Subscribe

两段能跑的代码

上面代码的主旨

Observer

subscribe 的参数都能是什么

除了 subscribe 方法,还有 RxKotlin 提供的小语法糖 subscribeBy

Subscribe

Disposable

Hot/Cold Observable

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

禅与计算机程序设计艺术

RxKotlin

Kotlin

程序员

反应式编程 · Reactive Programming · Reactor

Reactivex RxJava RxKotlin RxJS RxGo RxSwift

除了 `subscribe` 方法,还有 RxKotlin 提供的小语法糖 `subscribeBy`