RxSwift学习笔记

最近在学习RxSwift相关的内容，在这里记录一些基本的知识点，以便今后查阅。

Observable

在RxSwift中，最关键的一个概念是可观察序列（Observable Sequence），它相当于Swift中的序列（Sequence），可观察序列中的每个元素都是一个事件，我们知道Swift的序列中可以包含任意多个元素，类似的，可观察序列会不断产生新的事件直到发生错误或正常结束为止。订阅者（Observer）通过订阅（subscribe）一个可观察队列来接收序列所产生的新事件，只有在有观察者的情况下序列才可以发送事件。

例如，使用of操作创建一个可观察序列：

let seq = Observable.of(1, 2, 3)

of是一种用来创建Observable的简便操作，在上面的代码中创建了一个类型为Observable<Int>的Observable，里面包含了三个元素：1，2，3。

来看看Observable中都提供了哪些操作，可观察序列是一个实现了ObservableType协议的类型，ObservableType协议的定义非常简单：

protocol ObservableType : ObservableConvertibleType {
    associatedtype E
    func subscribe<O: ObserverType>(_ observer: O) -> Disposable where O.E == E
}

其中E是一个关联类型，表示序列中元素的类型，除此之外协议只定义了一个方法：subscribe，用于向可观察序列添加一个观察者（ObserverType类型）：

// 接收闭包的subscribe函数是通过协议扩展提供的简便方法
seq.subscribe { (event) in
    print(event)
}

subscribe相当于Swift序列中的遍历操作（makeIterator），如上，向seq序列添加一个观察者，在序列中有新的事件时调用该闭包，上面的代码会输出1，2，3。

Observer

观察者是实现了ObserverType协议的对象，ObserverType协议同样十分简单：

public protocol ObserverType {
    associatedtype E
    func on(_ event: Event<E>)
}

E为观察者所观察序列中的元素类型，当序列中有新的事件产生时，会调用on方法来接收新的事件。其中事件的类型Event是一个枚举，其中包含3个类型：

enum Event<Element> {
    case next(Element)
    case error(Swift.Error)
    case completed
}
  1. .next：表示序列中产生了下一个事件，关联值Element保存了该事件的值。
  2. .error：序列产生了一个错误，关联值Error保存了错误类型，在这之后序列会直接结束(不再产生新的next事件)。
  3. .completed：序列正常结束。

Dispose

除了产生错误和自然结束以外，还可以手动结束观察，在使用subscribe订阅一个可观察序列时，会返回一个Disposable类型的对象。这里的Disposable是一个协议，只定义了一个方法：

protocol Disposable {
    func dispose()
}

dispose方法用来结束此次订阅并释放可观察序列中的相关资源，通常来说你并不需要直接调用该方法，而是通过调用其扩展方法addDisposableTo将Disposable添加到一个DisposeBag对象中。DisposeBag对象会自动管理所有添加到其中的Disposable对象，在DisposeBag对象销毁的时候会自动调用其中所有Disposable的dispose方法释放资源。

也可以使用takeUntil来自动结束订阅：

seq.takeUntil(otherSeq)
    .subscribe({ (event) in
        print(event)
    })

在otherSeq序列发出任意类型的事件之后，自动结束本次订阅。

创建序列

通过Observable类型提供的方法create可以创建一个自定义的可观察序列：

let seq = Observable<Int>.create { (observer) -> Disposable in
    observer.on(.next(1))
    observer.on(.completed)
    return Disposables.create {
        // do some cleanup
    }
}

create方法使用一个闭包来创建自定义的序列，闭包接收一个ObserverType的参数observer，并通过observer来发送相应的事件。如上面的代码，创建了一个Observable<Int>类型的可观察序列，订阅该序列的观察者会收到事件1和一个完成事件。最后create方法返回一个自己创建的Disposable对象，可以在这里进行一些相关的资源回收操作。

除了create方法之外，RxSwift中提供了很多中简便的方法用于创建序列，常用的有：

  • just：创建一个只包含一个值的可观察序列：

    let justSeq = Observable.just(1)
justSeq.subscribe { (event) in
    print(event)
}
---- example output ----
next(1)
completed

  • of：of和just有点类似，不同的是of可以将一系列元素创建成事件队列，该Observable依次发送相应事件和结束事件：

    let ofSeq = Observable.of(1, 2, 3)
ofSeq.subscribe { (event) in
    print(event)
}
---- example output ----
next(1)
next(2)
next(3)
completed

  • empty：这种类型的Observable只发送结束(Completed)事件

    let emptySequence = Observable<String>.empty()

  • error：该队列只发送一个error事件，传递一个自定义的错误类型。

    let errorSeq = Observable<TestError>.error(TestError.Error1)

Share

通常在我们在订阅一个可观察序列的时候，每一次的订阅行为都是独立的，也就是说：

let seq = Observable.of(1, 2)
// 1
seq.subscribe { (event) in
    print("sub 1: \(event)")
}
// 2
seq.subscribe { (event) in
    print("sub 2: \(event)")
}
---- example output ----
sub 1: next(1)
sub 1: next(2)
sub 1: completed 
sub 2: next(1)
sub 2: next(2)
sub 2: completed

我们连续订阅同一序列两次，每次都会接收到相同的事件，第二次订阅时并没有因为第一次订阅的行为导致元素"耗尽"。有些时候我们希望让所有的观察者都共享同一份事件，这个时候可以使用share

  • share：share是ObservableType协议的一个扩展方法，它返回一个可观察序列，该序列的所有观察者都会共享同一份订阅，上面的代码加上share之后：

    let seq = Observable.of(1, 2).share()
// 1
seq.subscribe { (event) in
    print("sub 1: \(event)")
}
// 2
seq.subscribe { (event) in
    print("sub 2: \(event)")
}
---- example output ----
sub 1: next(1)
sub 1: next(2)
sub 1: completed 
sub 2: completed

    可以看到，在第一次订阅时序列已经将所有的事件发送，后面再进行第二次订阅的时候只收到了一个完成事件。

  • shareReplay：shareReplay的用法与share类似，它的方法签名如下：

    func shareReplay(_ bufferSize: Int) -> Observable<Element>

    不同的地方在于，shareReplay接收一个整型参数bufferSize，指定缓冲区大小，订阅该序列的观察者会立即收到最近bufferSize条事件。

序列的变换和组合

在Swift的序列Sequence中，可以使用map、flatMap和reduce等常见的函数式方法对其中的元素进行变换，RxSwift中的可观察序列同样也支持这些方法。

变换

  • map：这是map方法的签名：

    func map<Result>(_ transform: @escaping (E) throws -> Result) -> Observable<Result>

    在一个自定义的闭包中对序列的每一个元素进行变换，返回一个包含转换后结果的可观察序列，与Swift中Sequence的map类似。

    let mappedSeq: Observable<String> = seq.map { (element) -> String in
  return "value: \(element)"
}

  • flatMap：先来看看flatMap的签名：

    func flatMap<O: ObservableConvertibleType>(_ selector: @escaping (E) throws -> O)
        -> Observable<O.E>

    关于flatMap的作用同样可以类比Sequence，Sequence中的flatMap闭包遍历每一个元素进行处理后返回一个新的序列，最后会将这些序列"展平"，得到一个包含所有序列元素的新序列：

    let array = [1, 2]
let res = array.flatMap { (n) -> [String] in
    return ["\(n)a", "\(n)b"]
}
// res: ["1a", "1b", "2a", "2b"]

    RxSwift中的flatMap用法与之类似，flatMap中的闭包会遍历可观察序列中的所有元素，并返回一个新的可观察序列，最后flatMap会返回一个包含所有元素的可观察序列：

    let seq = Observable.of(1, 2)
    .flatMap { (n) -> Observable<String> in
        return Observable.of("\(n)a", "\(n)b") // (1)
    }
    .subscribe { (event) in
        print(event)
    }
// 得到的seq类型为Observable<String>
---- example output ----
next(1a)
next(1b)
next(2a)
next(2b)
completed

    在闭包中创建了若干个可观察序列(1)，这些序列中发送的next事件都会被传递到seq序列中，其中任何一个序列发生错误（发送了error事件）时，seq序列都会直接结束，不再继续接收事件；但是只有所有序列都完成（发送了completed事件）后，seq序列才会正常结束。

  • flatMapLatest：作用与flatMap类似，但是对于闭包中生成的可观察序列，它并不会保留所有的序列的订阅，在遍历结束后，只保留最后创建的序列的订阅，之前创建的Observables都会取消订阅（相应序列的dispose方法也会被调用）：

    // 与上一个例子相同的代码，仅将flatMap改成flatMapLatest
let seq = Observable.of(1, 2)
    .flatMapLatest { (n) -> Observable<String> in
        return Observable.of("\(n)a", "\(n)b") // (1)
    }
    .subscribe { (event) in
        print(event)
    }
---- example output ----
next(1a)
next(2a)
next(2b)
completed

    因为订阅关系的改变，现在只有当最后创建的那个Observable正常结束时，seq才会收到completed事件。

    在这种情况下，flatMapLatest会得到与flatMap相同的输出：

    let seq = Observable.of(1, 2)
    .flatMapLatest { (n) -> Observable<String> in
        return Observable<String>.create({ (observer) -> Disposable in
            observer.onNext("\(n)a")
            observer.onNext("\(n)b")

            return Disposables.create { }
        })
    }
    .subscribe { (event) in
        print(event)
    }

    这是因为在上面的这个例子中所创建的Observable是同步创建元素的，无法被打断。

    类似的方法还有flatMapFirst，使用方法可以类比flatMapLatest。

  • reduce和scan：reduce的作用与Sequence中定义的一样，它接收一个初始值和一个闭包，在Observable中的每个值上调用该闭包，并将每一步的结果作为下一次调用的输入：

    Observable.of(1, 2, 3).reduce(0) { (first, num) -> Float in
        return Float(first + num)
    }
    .subscribe { (event) in
        print(event)
    }
// 输出：next(6.0), completed

    在上面的代码中，提供了一个初始值0，在闭包中计算和，并将结果序列的元素类型改成Float，序列的观察者最后接收到所有元素的和。

    scan的作用类似于reduce，它跟reduce之间唯一的区别在于，scan会发送每一次调用闭包后的结果：

    Observable.of(1, 2, 3).scan(0) { (first, num) -> Float in
        return Float(first + num)
    }
    .subscribe { (event) in
        print(event)
    }
// 输出：next(1.0), next(3.0), next(6.0), completed

组合

  • startWith：在序列的开头加入一个指定的元素

    Observable.of(2, 3).startWith(1).subscribe { (event) in
    print(event)
}
---- example output ----
next(1)
next(2)
next(3)
completed

    订阅该序列之后，会立即收到startWith指定的事件，即使此时序列并没有开始发送事件。

  • merge：当你有多个类型相同的Observable，可以使用merge方法将它们合并起来，同时订阅所有Observable中的事件：

    let seq1 = Observable.just(1)
let seq2 = Observable.just(2)
let seq = Observable.of(seq1, seq2).merge()
seq.subscribe { (event) in
    print(event)
}
---- example output ----
next(1)
next(2)
completed

    只有当Observable中的元素也是Observable类型的时候才可以使用merge方法，当其中一个序列发生错误的时候，seq都会被终止，同样的只有所有序列都完成之后，seq才会收到完成事件。

  • zip：zip方法也可以将多个Observable合并在一起，与merge不同的是，zip提供了一个闭包用来对多个Observable中的元素进行组合变化，最后获得一个新的序列：

    let seq1 = Observable.just(1)
let seq2 = Observable.just(2)
let seq: Observable<String> = Observable.zip(seq1, seq2) { (num1, num2) -> String in
    return "\(num1 + num2)"
}
seq.subscribe { (event) in
    print(event)
}
---- example output ----
next(3)
completed

    zip方法按照参数个数的不同有多个版本，最多支持合并8个可观察序列，需要注意的一点是，闭包所接收的参数是各个序列中对应位置的元素。也就是说，如果seq1发送了一个事件，而seq2发送了多个事件，闭包也只会被执行一次，seq中只有一个元素。

    组合的Observable中任意一个发生错误，最后的seq都会直接出错终止，当所有的Observable都发出completed事件后，seq才会正常结束。

  • combineLatest：combineLatest同样用于将多个序列组合成一个，使用方法与zip一样，但是它的调用机制跟zip不同，每当其中一个序列有新元素时，combineLatest都会从其他所有序列中取出最后一个元素，传入闭包中生成新的元素添加到结果序列中。

Subject

Subject对象相当于一种中间的代理和桥梁的作用，它既是观察者又是可观察序列，在向一个Subject对象添加观察者之后，可以通过该Subject向其发送事件。Subject对象并不会主动发送completed事件，并且在发送了error或completed事件之后，Subject中的序列会直接终结，无法再发送新的消息。Subject同样也分为几种类型：

  • PublishSubject：PublishSubject的订阅者只会收到在其订阅（subscribe）之后发送的事件

    let subject = PublishSubject<Int>()
subject.onNext(1)
subject.subscribe { (event) in
    print(event)
}
subject.onNext(2)

---- example output ----
next(2)

    可以看到，观察者只收到了事件2，在订阅之前发送的事件1并没有接收到。

  • ReplaySubject：ReplaySubject在初始化时指定一个大小为n的缓冲区，里面会保存最近发送的n条事件，在订阅之后，观察者会立即收到缓冲区中的事件：

    let subject = ReplaySubject<Int>.create(bufferSize: 2)
subject.onNext(1)
subject.subscribe { (event) in
    print(event)
}
subject.onNext(2)

---- example output ----
next(1)
next(2)

  • BehaviorSubject：BehaviorSubject在初始化时需要提供一个默认值，在订阅时观察者会立刻收到序列上一次发送的事件，如果没有发送过事件则会收到默认值：

    let subject = BehaviorSubject(value: 1)
subject.subscribe { (event) in
    print(event)
}

---- example output ----
next(1)

  • Variable：Variable是对BehaviorSubject的一个封装，行为上与BehaviorSubject类似。Variable没有on之类的方法来发送事件，取而代之的是一个value属性，向value赋值可以向观察者发送next事件，并且访问value可以获取最后一次发送的数据：

    let variable = Variable(1)
variable.asObservable().subscribe { (event) in
    print(event)
}
variable.value = 2

---- example output ----
next(1)
next(2)
completed

    与其他Subject类型不同的是，Variable在释放的时候会发送completed事件，并且Variable对象永远不会发送error事件。

Scheduler

Scheduler是RxSwift中进行多线程编程的一种方式，一个Observable在执行的时候会指定一个Scheduler，这个Scheduler决定了在哪个线程对序列进行操作以及事件回调。默认情况下，在订阅Observable之后，观察者会在与调用subscribe方法时相同的线程收到通知，并且也会在该线程进行销毁（dispose）。

与GCD类似，Scheduler分为串行（serial）和并行（concurrent）两种类型，RxSwift中定义了几种Schedular：

  • CurrentThreadScheduler：这是默认的Scheduler，代表了当前的线程，serial类型。
  • MainScheduler：表示主线程，serial类型
  • SerialDispatchQueueScheduler：提供了一些快捷的方法来创建串行Scheduler，内部封装了DispatchQueue
  • ConcurrentDispatchQueueScheduler：提供了快捷的方法来创建并行Scheduler，同样封装了DispatchQueue

subscribeOn和observeOn

subscribeOn和observeOn是其中两个最重要的方法，它们可以改变Observable所在的Scheduler：

// main thread
let scheduler = ConcurrentDispatchQueueScheduler(qos: .default)
let seq = Observable.of(1, 2)
seq.subscribeOn(scheduler)
    .map {
        return $0 * 2 // 子线程
    }
    .subscribe { (event) in
        print(event) // 子线程
    }

在上面的代码中创建了一个并发的Scheduler，并在序列seq上调用subscribeOn指定了该Scheduler，可以看到，我们在主线程中订阅该序列，但是map方法以及事件的回调都是在创建的子线程中执行。

subscribeOn和observeOn都可以指定序列的Scheduler，它们之间的区别在于：

  • subscribeOn设定了整个序列开始的时候所在的Scheduler，序列在创建以及之后的操作都会在这个Scheduler上进行，subscribeOn在整个链式调用中只能调用一次，之后再次调用subscribeOn没有任何效果。
  • observeOn指定一个Scheduler，在这之后的操作都会被派发到这个Scheduler上执行，observeOn可以在链式操作的中间改变Scheduler
createObservable().
    .doSomething()
    .subscribeOn(scheduler1) // (1)
    .doSomethingElse()
    .observeOn(scheduler2) // (2)
    .doAnother()
    ...

如上代码，在(1)处执行了subscribeOn之后，之前的操作createObservable()和doSomething()都会在scheduler1中执行，随后的doSomethingElse()同样也在scheduler1中执行，随后用observeOn指定了另外一个scheduler2，之后的doAnother()会在scheduler2上执行。

为原有代码添加Rx扩展

RxSwift中提供了一种扩展机制，可以很方便的为原有的代码添加上Rx扩展。首先来看一个结构体Reactive：

public struct Reactive<Base> {
    /// base是扩展的对象实例
    public let base: Base
    
    public init(_ base: Base) {
        self.base = base
    }
}

Reactive是一个泛型结构体，只定义了一个属性base，并且在初始化结构体的时候传入该属性的值。

此外还定义了一个协议ReactiveCompatible：

public protocol ReactiveCompatible {
    associatedtype CompatibleType

    static var rx: Reactive<CompatibleType>.Type { get set }
    var rx: Reactive<CompatibleType> { get set }
}

该协议中分别为类对象和实例对象定义一个名字相同的属性：rx，类型为上面定义的Reactive，随后通过协议扩展为其提供了get的默认的实现：

extension ReactiveCompatible {
    public static var rx: Reactive<Self>.Type {
        get {
            return Reactive<Self>.self
        }
        set {
            // this enables using Reactive to "mutate" base type
        }
    }

    public var rx: Reactive<Self> {
        get {
            return Reactive(self)
        }
        set {
            // this enables using Reactive to "mutate" base object
        }
    }
}

关联类型CompatibleType被自动推导为实现该协议的类本身，使用self初始化一个Reactive对象。

最后通过协议扩展为所有的NSObject类型实现了ReactiveCompatible协议：

extension NSObject: ReactiveCompatible { }

这样一来，代码中所有继承自NSObject的类型实例中都会有一个类型为Reactive的属性rx，当我们要为自己的类型添加Rx扩展时，只需要通过扩展向Reactive中添加方法就可以了，例如向UIButton类型添加扩展：

extension Reactive where Base: UIButton { // 为Reactive<UIButton>添加扩展
    public var tap: ControlEvent<Void> {
        return controlEvent(.touchUpInside) // 通过base可以访问该实例本身
    }
}

由于Reactive是一个泛型类型，我们可以通过where语句指定泛型的类型，这样一来，我们就可以在UIButton实例的rx中访问tap属性了：

let button = UIButton(...)
button.rx.tap

类似RxCocoa这样的RxSwift扩展库都是通过这种方式进行Rx扩展的。

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 206,126评论 6赞 481
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 88,254评论 2赞 382
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 152,445评论 0赞 341
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 55,185评论 1赞 278
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 64,178评论 5赞 371
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 48,970评论 1赞 284
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 38,276评论 3赞 399
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 36,927评论 0赞 259
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 43,400评论 1赞 300
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,883评论 2赞 323
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,997评论 1赞 333
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 33,646评论 4赞 322
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 39,213评论 3赞 307
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,204评论 0赞 19
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,423评论 1赞 260
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 45,423评论 2赞 352
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 42,722评论 2赞 345

推荐阅读更多精彩内容

  • RXSwift的使用用法
    发现 关注 消息 RxSwift入坑解读-你所需要知道的各种概念 沸沸腾关注 2016.11.27 19:11*字...
    枫叶1234阅读 2,780评论 0赞 2
  • Rxjava
    我从去年开始使用 RxJava ，到现在一年多了。今年加入了 Flipboard 后，看到 Flipboard 的...
    Jason_andy阅读 5,456评论 7赞 62
  • RxSwift基础之一 - Observables & Subjects
    本文章内部分图片资源来自RayWenderlich.com 本文结合自己的理解来总结介绍一下RxSwift最基本的...
    FKSky阅读 2,853评论 4赞 14
  • 吃吃的快乐（2）
    吃吃的快乐，吃吃的爱 01 经典的主食以外，新加坡的面点和小吃，也是毫不逊色。融合东南亚各国的特色，在这里交汇，组...
    洪一念阅读 482评论 3赞 5
  • 开学后第一节体育课
    已经开学一个周了，今天下午是我们开学的第一节体育课。今天体育课的内容是:学习蹲打篮球。 上课了，...
    zhoukewen阅读 590评论 0赞 0