开启RxSwift之旅——开篇

浏览: 204 发布日期: 2017-08-05 分类: swift

RxSwift 是 ReactiveX 在 Swift 下的实现。ReactiveX 是一个通过使用可观察序列来组合异步和基于事件的程序的库。

很多地方通常把 ReactiveX 称为 “函数响应式编程” ,其实这是不恰当的。ReactiveX 可以是函数式的,可以是响应式的,但是和“函数响应式编程”是不同的概览。一个主要的不同点是“函数响应式编程”是对随着时间不停变化的值进行操作的,而 ReactiveX 对随时间发射的离散值进行操作。

我们先不急着去看 RxSwift 的源码,在这之前,我们有必要先了解一下什么是响应式编程。

"什么是响应式编程"

响应式编程是一种面向数据流和变化传播的编程范式。

在某种程度上,这并不是什么新东西。用户输入、单击事件、变量值等都可以看做一个流,你可以观察这个流,并基于这个流做一些操作。响应式就是基于这种想法。

一个流就是一个将要发生的以时间为序的事件序列。它能发射出三种不同的东西:一个数据值(某种类型的),一个错误(error)或者一个“完成(completed)”的信号。比如说,当前按钮所在的窗口或视图关闭时,“单击”事件流也就“完成”了。

以一个单击事件流为例:定义一个针对数据值的函数,在发出一个值时,该函数就会异步地执行,还有一个针对发出错误时的函数,最后还有针对发出‘完成’时的函数。“监听”流的行为叫做订阅。我们定义的这些函数就是观察者。这个流就是被观察的主体(subject)(或“可观察的(observable)”)。这正是观察者设计模式。

在你使用 RxSwift 时,你就会发现它正是按照这种模式来进行设计的。在 RxSwift 中,一个流可以被称为序列(Sequences)。序列的生产者就是 Observable 。

在 RxSwift 的 playground 中就有这么一句话:

Every Observable instance is just a sequence.

Observable

如果你在学习 RxSwift 之前就使用过 ReactiveCocoa 的话,你会发现 RxSwift 和 ReactiveCocoa 完全是两个不同的物种。在 RxSwift 的世界里,所有的东西都是 Observable 的。你可以创造它们、操作它们,然后订阅它们来响应变化。

理解 Observable 还有一件很重要的事情:

Observables will not execute their subscription closure unless there is a subscriber.

可以这么理解,如果你只是调用一个返回一个 Observable 的方法,生成序列不会被执行。Observable 只是一个解释序列如何被生成和什么参数被使用于生成元素的定义。生成序列开始于 subscribe 方法被调用的时候。

下面的例子中,Observable 的闭包永远不会执行:

example("Observable with no subscribers") {
    _ = Observable.create { observer -> Disposable in
        print("This will never be printed")
        observer.on(.next("????"))
        observer.on(.completed)
        return Disposables.create()
    }
}

只有当我们调用 subscribe(_:) 时,Observable 的闭包才会执行:

example("Observable with subscriber") {
  _ = Observable.create { observer in
            print("Observable created")
            observer.on(.next("????"))
            observer.on(.completed)
            return Disposables.create()
        }
        .subscribe { event in
            print(event)
    }
}

上面例子中从传入闭包创建一个 Observable ,到调用 subscribe(_:) 这个过程中 RxSwift 到底做了什么?我们可以先从简单的 empty 开始。

empty

empty 就是创建一个空的 sequence, 它只能发出一个 completed 事件。

example(of: "empty") {
    Observable.empty()
        .subscribe({
            print($0)
    }) 
}

// 打印结果
--- Example of: empty ---
completed

上面代码中通过 Observable 的 empty 方法创建了一个 Observable, 打开 Observable+Creation.swift 文件,可以看到 empty() 的实现:

public static func empty() -> Observable {
    return EmptyProducer()
}

这里返回了一个 EmptyProducer 的实例,点进去看看EmptyProducer是个什么东西:

final class EmptyProducer : Producer {
    override func subscribe(_ observer: O) -> Disposable where O.E == Element {
        observer.on(.completed)
        return Disposables.create()
    }
}

EmptyProducerProducer 的子类,重写了 subscribe(:) 。在 subscribe 方法中,观察者订阅了一个完成信号。

当我们通过 empty() 创建了一个 Observable 后,然后会调用 subscribe(_:),打开 ObservableType+Extensions.swift 文件, 可以看到 subscribe 方法的实现:

public func subscribe(_ on: @escaping (Event) -> Void) -> Disposable {
    let observer = AnonymousObserver { e in
        on(e)
    }
    return self.subscribeSafe(observer)
}

subscribe 方法接受了闭包之后,先创建了一个匿名观察者,subscribe 的闭包参数作为构造器的参数传给了 observer。点击进去 AnonymousObserver源码:

final class AnonymousObserver : ObserverBase {
    typealias Element = ElementType
    
    typealias EventHandler = (Event) -> Void
    
    private let _eventHandler : EventHandler
    
    init(_ eventHandler: @escaping EventHandler) {
#if TRACE_RESOURCES
        let _ = Resources.incrementTotal()
#endif
        _eventHandler = eventHandler
    }

    override func onCore(_ event: Event) {
        return _eventHandler(event)
    }
    
#if TRACE_RESOURCES
    deinit {
        let _ = Resources.decrementTotal()
    }
#endif
}

AnonymousObserver 的构造器接受一个闭包,然后在 onCore 方法中, 私有的 _eventHandler 会被调用。到这里为止,我们还是不知道我们在调用 subscribe(_:) 时传入的闭包最终的调用时机。不过已经很清楚的知道了,这个闭包在 onCore(:) 中调用了,我们继续进入 AnonymousObserver 的父类 ObserverBase 中一探究竟:

class ObserverBase : Disposable, ObserverType {
    typealias E = ElementType

    private var _isStopped: AtomicInt = 0

    func on(_ event: Event) {
        switch event {
        case .next:
            if _isStopped == 0 {
                onCore(event)
            }
        case .error, .completed:
            if AtomicCompareAndSwap(0, 1, &_isStopped) {
                onCore(event)
            }
        }
    }

    func onCore(_ event: Event) {
        rxAbstractMethod()
    }

    func dispose() {
        _ = AtomicCompareAndSwap(0, 1, &_isStopped)
    }
}

这一下就很清楚了,onCore(:) 会被 on(:) 调用。让我们再次回到 ObservableType+Extensions.swift 文件中,匿名观察者(AnonymousObserver)创建完后,调用 subscribeSafe(:) 作为函数返回值。在文件的最下面可以看到 subscribeSafe(:) 的实现:

fileprivate func subscribeSafe(_ observer: O) -> Disposable where O.E == E {
    return self.asObservable().subscribe(observer)
}

这里会调用 subscribe(:) ,注意了,这里的 subscribe(:) 是 ObservableType 协议中定义的方法:

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

这里的参数是一个 ObserverType,也就是一个观察者,千万要与 func subscribe(_ on: @escaping (Event) -> Void) -> Disposable 做好区分。

好了, subscribe 方法将创建的匿名观察者作为参数,而在 EmptyProducer 中的 subscribe 的实现我们已经看过了:

override func subscribe(_ observer: O) -> Disposable where O.E == Element {
    observer.on(.completed)
    return Disposables.create()
}

这里刚好调用了观察者的 on(:), 在 ObserverBase 中 on 方法会调用 onCore(:), onCore 方法调用了 subscribe(_ on: @escaping (Event) -> Void) -> Disposable 参数中的闭包。由于 subscribe(_ observer: O) 中观察者只订阅了 "completed" 信号,所有闭包不会执行。

至此从创建一个 observable, 到调用 subscribe(_:) 整个过程我们已经很清楚了。现在也就能明白为什么只是调用一个返回一个 Observable 的方法,生成序列不会被执行了。

小结

最后总结一下调用 subscribe(_:) 后的整个过程:用 subscribe 中的闭包创建一个匿名观察者(观察者私有的 _eventHandler 会将闭包保存起来),然后将创建的匿名观察者作为参数传给 subscribeSafe(:) , subscribeSafe(:) 会调用 subscribe(:), 并将匿名观察者作为参数。subscribe(:) 会调用 observer 的 on(:), 当 observer 的 on 方法被调用后,最终会调用开始时传入的闭包。

以上只是分析了一下 empty 的实现,像 of, just, create 等的实现在细节上有一些区别,总的思路是一样的。在查看源码时可能会有一点绕,主要是因为继承太多,很多方法都要到父类中去找,而且 ObservableType 和 ObserverType 的 Extension 太多,代码分散到各个文件中。

RxSwift 的代码只看了个开头,还有很多地方没有完全弄明白。在使用 RxSwift 的过程中你能体会到 "响应式" 和 "函数式" 给我们的开发带来的便利性。

返回顶部