RXSwift之Dispose销毁者解析

销毁同样是 RxSwift 中和序列，观察，调度并列，最重要的四个元素之一。
销毁一般有两种方法：

订阅产生的可 Disposable 清除资源对象，单独调用 dispose 方法进行销毁
通过将产生的清除对象放到 DisposeBag 中，在作用域结束后被释放，也可以在需要的时候置空释放。

初探

那么首先从基本序列创建和订阅开始分析

例子：

    let ob = Observable<Any>.create { (observer) -> Disposable in
        observer.onNext("lb")
        return Disposables.create {print("销毁释放了")}
    }
    
    let dispose = ob.subscribe(onNext: { (element) in
        print("订阅到了：\(element)")
    }, onError: { (error) in
        print("订阅到了：\(error)")
    }, onCompleted: {
        print("完成")
    }) {
        print("销毁回调")
    }
    
    print("执行完毕")
    dispose.dispose()

打印结果：

订阅到了：lb
执行完毕
销毁释放了
销毁回调

分析：
点进去 Disposables.create 点不进去的按老方法，直接搜索找，swift 会记录下一次就点的进了。

extension Disposables {

    /// - parameter dispose: Disposal action which will be run upon calling `dispose`.
    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?

    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()
            }
        }
    }
}

跟订阅方法类似，同样是产生了一个匿名中间类 AnonymousDisposable 。使用属性保存了销毁闭包参数。
这个类继承与 DisposeBase，遵循 Cancelable 的 protocol，而且自己实现了私有方法 dispose()
这个方法中有一个 fetchOr 方法值得一提：

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

首先其实现目的是为了保证满足 if 条件里的代码只执行一次。采用与1进行位运算，并且

private let _isDisposed = AtomicInt(0)

可以看出 self._isDisposed 的初始值是0，结合 fetchOr 具体方法实现，达到只会执行一次的效果。
执行 if 里的代码

if let action = self._disposeAction {
     self._disposeAction = nil
     action()
}

先赋值到一个临时变量 action 同时判断 _disposeAction 为不为空，将自己的 _disposeAction 属性置空，执行销毁回调。也就是我们写的 print("销毁释放了")

那么这个 dispose() 方法干了啥我们搞清楚了，问题又来了它是什么时候被调用的呢？

不着急，我们接着往下走，来到我们写的代码中，走 ob.subscribe(onNext ... ，这个方法很熟悉了。直接进来

public func subscribe(onNext: ((E) -> 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<E> { event in
            switch event {
            case .next(let value):
                onNext?(value)
            case .error(let error):
                if let onError = onError {
                    onError(error)
                }
                else { ... }
                disposable.dispose()
            case .completed:
                onCompleted?()
                disposable.dispose()
            }
        }
        return Disposables.create(
            self.asObservable().subscribe(observer),
            disposable
        )
}

这个方法中首先 Disposables.create 创建 Disposables 并保存了外界传递的销毁回调也就是我们写的

print("销毁回调")

接下来来到重点:

return Disposables.create(
   self.asObservable().subscribe(observer),
   disposable
)

这里可以看到, 它就是我们外界订阅调用 subscribe 方法得到的返回,并且外界可以随时调用这个返回的对象的 .dispose() 方法进行销毁.

直接点进去 .create方法

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

继续进去该类找 dispose 方法

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

这个二元销毁者的 dispose 方法也是常规操作，分别销毁. 值得一提的是它是一个开放出来的方法,因此外界可以随时调用该对象的销毁方法.

接下来回到 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

run方法由子类自己实现 ,从而来到 AnonymousObservable 的 run 方法中

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

众所周知 sink.run 就是调用
这个 subscription 就是 AnonymousDisposable。(原因是 sink.run 就是调用 AnonymousObservable 中保存的 _subscribeHandler 回调也就是用户创建 create 时初始化传进来的). 而且我们 create 闭包中返回的是
AnonymousDisposable
也就是这句我们自己写的代码

let ob = Observable<Any>.create { (observer) -> Disposable in
   observer.onNext("lb")
   return Disposables.create {print("销毁释放了")}
}

然后就把 AnonymousObservableSink 和 AnonymousDisposable 打包进元组给返回了。

再继续看 SinkDisposer 的 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 (就是外界创建序列的闭包的返回销毁者)
取了某一个状态：previousState ,判断状态的条件，然后执行这两个保存属性的销毁和置空释放销毁

最后回来到外部我们自己敲的代码中
来到

dispose.dispose()

也就是调用 BinaryDisposable 二元的销毁者的 .dispose() 方法一一销毁. 方法实现前面提到.

至此整个流程走完. 最值得一提的是

我们在 RxSwift 的世界里最重要的东西 -- 管道 sink，我们就是通过销毁 sink中保存的属性以及关联来实现销毁响应流程.