RxSwift源码分析(六)-销毁者Disposable

作者: 盾子 | 来源:发表于2019-08-08 13:06 被阅读0次

销毁者简介

可被清除的资源Disposable是RxSwift的核心成员之一，它主要是用来清除不再需要的资源。那么下面来探索一下RxSwift是怎样管理这些资源的生命周期呢。

通常来说，一个序列如果发出了 error 或者 completed 事件，那么所有内部资源都会被释放，不需要我们手动释放。
但是如果你需要提前释放这些资源或取消订阅的话，那么你可以对返回的可被清除的资源（Disposable）调用 dispose 方法。
不过官方推荐使用清除包（DisposeBag）来管理订阅的生命周期，一般是把资源加入到一个全局的DisposeBag里面，它跟随着页面的生命周期，当页面销毁时DisposeBag也会随之销毁，同时DisposeBag里面的资源也会被一一释放。

var disposeBag = DisposeBag() // 来自父类 ViewController

override func viewDidLoad() {
    super.viewDidLoad()

    ...

    usernameValid
        .bind(to: passwordOutlet.rx.isEnabled)
        .disposed(by: disposeBag)

    usernameValid
        .bind(to: usernameValidOutlet.rx.isHidden)
        .disposed(by: disposeBag)
}

销毁者的实现探索

下面这一段代码是常规的创建一个序列，然后订阅，最后手动销毁的流程。

let observable = Observable<Any>.create { (observer) -> Disposable in
    observer.onNext("七夕快乐")
    return Disposables.create {
        print("销毁释放了")
    }
}

let dispose = observable.subscribe(onNext: { (message) in
    print("有一条新消息:\(message)")
}, onError: { (error) in
    print("错误")
}, onCompleted: {
    print("完成")
}) {
    print("销毁回调")
}

print("开始调用dispose")
dispose.dispose()

执行结果：
有一条新消息:七夕快乐
开始调用dispose
销毁释放了
销毁回调

首先可以看到，在创建序列Observable<Any>.create方法有一个尾随闭包，需要返回一个实现了Disposable协议的实例。
进入到Disposables.create方法里面看看

extension Disposables {
    public static func create(with dispose: @escaping () -> Void) -> Cancelable {
        return AnonymousDisposable(disposeAction: dispose)
    }
}

创建了一个AnonymousDisposable对象并返回，很明显，这是一个匿名销毁者，跟创建序列的时候会创建一个匿名序列实现方式非常相似。

fileprivate final class AnonymousDisposable : DisposeBase, Cancelable {
    public typealias DisposeAction = () -> Void

    private let _isDisposed = AtomicInt(0)
    private var _disposeAction: DisposeAction?

    public var isDisposed: Bool {
        return isFlagSet(self._isDisposed, 1)
    }

    fileprivate init(_ disposeAction: @escaping DisposeAction) {
        self._disposeAction = disposeAction
        super.init()
    }

    fileprivate init(disposeAction: @escaping DisposeAction) {
        self._disposeAction = disposeAction
        super.init()
    }

    fileprivate func dispose() {
        if fetchOr(self._isDisposed, 1) == 0 {
            if let action = self._disposeAction {
                self._disposeAction = nil
                action()
            }
        }
    }
}

初始化的时候把外界传过来的闭包进行保存
然后看到有一个dispose方法，fetchOr(self._isDisposed, 1) == 0这行代码是控制if语句里面只会进去一次。
fetchOr方法的具体实现：AtomicInt是继承NSLock，在更改value值的时候加了一把锁，保证线程安全，然后运用了或运算并保存结果。位运算更加的高效。
最后先把self._disposeAction赋值给临时变量action，然后置空self._disposeAction，再执行action()。这样操作的原因是如果_disposeAction闭包是一个耗时操作，也能够保证_disposeAction能够立即释放。

func fetchOr(_ this: AtomicInt, _ mask: Int32) -> Int32 {
    this.lock()
    let oldValue = this.value
    this.value |= mask
    this.unlock()
    return oldValue
}

final class AtomicInt: NSLock {
    fileprivate var value: Int32
    public init(_ value: Int32 = 0) {
        self.value = value
    }
}

上面的流程，我们是在序列的回调闭包：_subscriberHandle里面，其实这个流程之前还有一个非常重要的流程：订阅，进入到observable.subscribe方法

public func subscribe(onNext: ((Element) -> Void)? = nil, onError: ((Swift.Error) -> Void)? = nil, onCompleted: (() -> Void)? = nil, onDisposed: (() -> Void)? = nil)
    -> Disposable {
    let disposable: Disposable
    
    if let disposed = onDisposed {
        disposable = Disposables.create(with: disposed)
    }
    else {
        disposable = Disposables.create()
    }
    
    let observer = AnonymousObserver<Element> { event in
        switch event {
        case .next(let value):
            onNext?(value)
        case .error(let error):
            if let onError = onError {
                onError(error)
            }
            else {
                Hooks.defaultErrorHandler(callStack, error)
            }
            disposable.dispose()
        case .completed:
            onCompleted?()
            disposable.dispose()
        }
    }
    
    return Disposables.create(
        self.asObservable().subscribe(observer),
        disposable
    )
}

首先创建了一个Disposable对象，并保存了销毁回调闭包，当执行销毁时，会把消息回调出去
在发出错误和完成事件之后也会执行disposable.dispose()，这就证实了前面说的：一个序列如果发出了 error 或者 completed 事件，那么所有内部资源都会被释放，不需要我们手动释放。
看最后一行代码return Disposables.create( self.asObservable().subscribe(observer), disposable )，这里返回的Disposable对象就是我们外面手动调用dispose.dispose()方法的dispose对象，或者说是加入到全局的DisposeBag的销毁者。
跟踪进入查看代码

public static func create(_ disposable1: Disposable, _ disposable2: Disposable) -> Cancelable {
    return BinaryDisposable(disposable1, disposable2)
}

创建了一个二元销毁者

func dispose() {
    if fetchOr(self._isDisposed, 1) == 0 {
        self._disposable1?.dispose()
        self._disposable2?.dispose()
        self._disposable1 = nil
        self._disposable2 = nil
    }
}

当执行dispose()时会把2个销毁者分别销毁
然后再来看看这个二元销毁者创建时的第一个参数：self.asObservable().subscribe(observer)的返回值是什么。来到Producer类的subscribe方法

let disposer = SinkDisposer()
let sinkAndSubscription = self.run(observer, cancel: disposer)
disposer.setSinkAndSubscription(sink: sinkAndSubscription.sink, subscription: sinkAndSubscription.subscription)
return disposer

创建了一个sink的销毁者SinkDisposer对象并返回，所以前面创建二元销毁者的第二个参数就是它。经过之前对RxSwift核心逻辑的分析文章，我们知道sink是连接序列和观察者的桥梁，当sink销毁后，序列和观察者之间就无法通讯了。
进入到self.run(observer, cancel: disposer)

override func run<Observer: ObserverType>(_ observer: Observer, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where Observer.Element == Element {
    let sink = AnonymousObservableSink(observer: observer, cancel: cancel)
    let subscription = sink.run(self)
    return (sink: sink, subscription: subscription)
}

创建了一个AnonymousObservableSink对象，并保存了上一步创建的SinkDisposer对象。在AnonymousObservableSink的源码里面发现on方法中，当发出完成和错误信号后，会立即执行dispose进行销毁，所以一旦我们的序列发出完成或者错误信号后就无法再次响应了！
执行sink.run(self)，方法里面执行的是parent._subscribeHandler(AnyObserver(self))，_subscribeHandler闭包就是外面创建序列Observable<Any>.create的尾随闭包，所以返回值就是Disposables.create {print("销毁释放了")}。
进入到setSinkAndSubscription方法

func setSinkAndSubscription(sink: Disposable, subscription: Disposable) {
    self._sink = sink
    self._subscription = subscription

    let previousState = fetchOr(self._state, DisposeState.sinkAndSubscriptionSet.rawValue)
    if (previousState & DisposeState.sinkAndSubscriptionSet.rawValue) != 0 {
        rxFatalError("Sink and subscription were already set")
    }

    if (previousState & DisposeState.disposed.rawValue) != 0 {
        sink.dispose()
        subscription.dispose()
        self._sink = nil
        self._subscription = nil
    }
}

保存了两个属性 : sink 和 subscription，就是上一步返回的销毁者和AnonymousObservableSink对象，AnonymousObservableSink里面保存了sink的销毁者SinkDisposer
根据记录的一个状态去判断刚刚保存的这两个属性是否需要销毁，需要的话就执行 dispose() 然后置空 nil
那么，当执行dispose.dispose()销毁时销毁的到底是什么呢

func dispose() {
    let previousState = fetchOr(self._state, DisposeState.disposed.rawValue)

    if (previousState & DisposeState.disposed.rawValue) != 0 {
        return
    }
    if (previousState & DisposeState.sinkAndSubscriptionSet.rawValue) != 0 {

        sink.dispose()
        subscription.dispose()

        self._sink = nil
        self._subscription = nil
    }
}

不管是系统销毁还是我们手动销毁都会执行dispose(),我们查看 dispose() 得出: 就是把初始化的时候保存的两个属性进行销毁然后置空。
在RxSwift中，sink存放了序列和观察者，来建立它们之间的响应关系，当把序列和观察者之间的桥梁sink销毁了，也就断开了它们之间的响应关系，从而无法再接收到消息。
附上一张图

总结

一个序列如果发出了 error 或者 completed 事件，那么所有内部资源都会被释放，不需要我们手动释放。
当执行销毁时，销毁的是序列和观察者之间的响应关系，不是序列和观察者对象本身
如果是加入到disposeBag，是在disposeBag对象销毁时，依次销毁里面存储的东西

RxSwift源码分析(六)-销毁者Disposable

销毁者简介

销毁者的实现探索

总结

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读