Rx 操作符五

Posted 2021-03-01 liuxiaokun

• sample

不定期的对 Observable 取样

sample 操作符将不定期的对源 Observable 进行取样操作。通过第二个 Observable 来控制取样时机。一旦第二个 Observable 发出一个元素，就从源 Observable 中取出最后产生的元素。

• scan

持续的将 Observable 的每一个元素应用一个函数，然后发出每一次函数返回的结果

scan 操作符将对第一个元素应用一个函数，将结果作为第一个元素发出。然后，将结果作为参数填入到第二个元素的应用函数中，创建第二个元素。以此类推，直到遍历完全部的元素。

这种操作符在其他地方有时候被称作是 accumulator。

let disposeBag = DisposeBag()

Observable.of(10, 100, 1000)
    .scan(1) { aggregateValue, newValue in
        aggregateValue + newValue
    }
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)

• shareReplay

使观察者共享 Observable，观察者会立即收到最新的元素，即使这些元素是在订阅前产生的

shareReplay 操作符将使得观察者共享源 Observable，并且缓存最新的 n 个元素，将这些元素直接发送给新的观察者。

• single

限制 Observable 只有一个元素，否出发出一个 error 事件

single 操作符将限制 Observable 只产生一个元素。如果 Observable 只有一个元素，它将镜像这个 Observable 。如果 Observable 没有元素或者元素数量大于一，它将产生一个 error 事件。

• skip

跳过 Observable 中头 n 个元素

skip 操作符可以让你跳过 Observable 中头 n 个元素，只关注后面的元素。

let disposeBag = DisposeBag()

Observable.of("??", "??", "??", "??", "??", "??")
    .skip(2)
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)

• skipUntil 跳过 Observable 中头几个元素，直到另一个 Observable 发出一个元素

skipUntil 操作符可以让你忽略源 Observable 中头几个元素，直到另一个 Observable 发出一个元素后，它才镜像源 Observable。

let disposeBag = DisposeBag()

let sourceSequence = PublishSubject<String>()
let referenceSequence = PublishSubject<String>()

sourceSequence
    .skipUntil(referenceSequence)
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)

sourceSequence.onNext("??")
sourceSequence.onNext("??")
sourceSequence.onNext("??")

referenceSequence.onNext("??")

sourceSequence.onNext("??")
sourceSequence.onNext("??")
sourceSequence.onNext("??")

• skipWhile

跳过 Observable 中头几个元素，直到元素的判定为否

skipWhile 操作符可以让你忽略源 Observable 中头几个元素，直到元素的判定为否后，它才镜像源 Observable。

let disposeBag = DisposeBag()

Observable.of(1, 2, 3, 4, 3, 2, 1)
    .skipWhile { $0 < 4 }
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)

• startWith

将一些元素插入到序列的头部

startWith 操作符会在 Observable 头部插入一些元素。

（如果你想在尾部加入一些元素可以用concat）

let disposeBag = DisposeBag()

Observable.of("??", "??", "??", "??")
    .startWith("1")
    .startWith("2")
    .startWith("3", "???", "???")
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)

• subscribeOn

指定 Observable 在那个 Scheduler 执行

ReactiveX 使用 Scheduler 来让 Observable 支持多线程。你可以使用 subscribeOn 操作符，来指示 Observable 在哪个 Scheduler 执行。

observeOn 操作符非常相似。它指示 Observable 在哪个 Scheduler 发出通知。

默认情况下，Observable 创建，应用操作符以及发出通知都会在 Subscribe 方法调用的 Scheduler 执行。subscribeOn 操作符将改变这种行为，它会指定一个不同的 Scheduler 来让 Observable 执行，observeOn 操作符将指定一个不同的 Scheduler 来让 Observable 通知观察者。

如上图所示，subscribeOn 操作符指定 Observable 在那个 Scheduler 开始执行，无论它处于链的那个位置。 另一方面 observeOn 将决定后面的方法在哪个 Scheduler 运行。因此，你可能会多次调用 observeOn 来决定某些操作符在哪个线程运行。