RxSwift 雜記(1)

協定 protocol 背後的 class

在 RxSwift 中有時會在協定 protocol 後面加上 class 關鍵字，比如下面的協定：

protocol SynchronizedDisposeType : class, Disposable, Lock {
    func _synchronized_dispose()
}

extension SynchronizedDisposeType {
    func synchronizedDispose() {
        lock(); defer { unlock() }
        _synchronized_dispose()
    }
}
複制代碼
           

通過在 protocol 後面加上 class 可以限定該協定隻能由類來繼承。因為在 swift 中 protocol 可以被除了類以外的其他類型所遵守，比如 struct、enum 這樣的類型。 那什麼時候需要需要在 protocol 後面加上 class 呢？ 當一個類的代理需要遵循某個協定，為了避免循環引用，通常會将類的代理聲明為 weak，例如下面的代碼：

protocol OneClassDelegate {
    func method()
}

class OneClass {
    weak var delegate: OneClassDelegate?
}
複制代碼
           

但是這樣的代碼是編譯不通過的，因為 struct、enum 這樣的類型可以遵循協定，但是不可以用 weak 這種 ARC 的記憶體管理方式修飾。這是就可以在 protocol 後面用 class 修飾。更詳細的可以檢視王巍的文章 DELEGATE。

協定和擴充協定

為了說明友善，下面一段代碼對 RxSwift 的源碼進行了變形：

protocol SynchronizedDisposeType : class {
    func lock()
    func unlock()
    func _synchronized_dispose()
}

extension SynchronizedDisposeType {
    func synchronizedDispose() {
        lock(); defer { unlock() }
        _synchronized_dispose()
    }
}
複制代碼
           

在

SynchronizedDisposeType

協定中，需要通過鎖實作一個同步的方法，它不是簡單的在 protocol 中定義一個方法，而是在協定中定義了一個在每個單詞前面加下劃線的方法

_synchronized_dispose

，并且擴充了一個方法

synchronizedDispose

，在内部将鎖和真正的方法組合了起來。在協定的實作上可以将加鎖的過程和真正要實作的方法進行了分離，保證了代碼的正确實作；如果有多個類要實作該協定時，減少了在方法前後加鎖的重複代碼。

内聯函數

通過 @inline 這個關鍵字可以聲明内聯函數，下面是 RxSwift 中一段代碼：

extension RecursiveLock : Lock {
    @inline(__always)
    final func performLocked(_ action: () -> Void) {
        lock(); defer { unlock() }
        action()
    }
}
複制代碼
           

@inline 的關鍵字有以下兩種用法：

// 聲明不要将函數編譯成内聯方式
@inline(never)

// 聲明要将函數編譯成内聯方式
@inline(__always) 
複制代碼
           

方法被串行執行的回報

在多線程的開發中，經常會遇到默寫方法必須保證串行執行，可以通過【方法或方法】來達到目的，但是在真正的實作過程中是否真的達到了目的呢？我們需要一種回報機制，可以確定代碼的運作方式和我們的預想一樣。

怎樣確定一個指定的方法在同一個時間隻被調用一次呢？也就是說我們需要確定：

1. 指定的方法在同一個時間内隻有一個線程調用，
2. 指定的方法在同一個時間同一個線程上隻被調用一次。

在 RxSwift 中實作的思路是這樣的，一個類中如果有方法需要被串行執行，則在一個遞歸鎖内用一個 Dictionary 來記錄方法目前被哪個線程執行了多少次，Dictionary 中線程為 key，執行次數為 value。如果被一個以上的線程執行，或者在一個線程中被執行的次數多于一次，則抛出異常。

在代碼具體實作中，将上述的記錄和判斷過程封裝到

SynchronizationTracker

類中，具體代碼如下：

final class SynchronizationTracker {
    // 遞歸鎖
    private let _lock = RecursiveLock()

    // 同步出錯時的錯誤類型
    public enum SychronizationErrorMessages: String {
        case variable = "..." // 具體資訊被删減
        case `default` = "..." // 具體資訊被删減
    }
    
    // 用于儲存執行情況的 Dictionary
    private var _threads = Dictionary<UnsafeMutableRawPointer, Int>()

    // 同步錯誤之後的處理
    private func synchronizationError(_ message: String) {
        #if FATAL_SYNCHRONIZATION
            rxFatalError(message)
        #else
            print(message)
        #endif
    }

    // 在同步方法内首先調用該方法用于記錄和判斷
    func register(synchronizationErrorMessage: SychronizationErrorMessages) {

        _lock.lock(); defer { _lock.unlock() }

        // 擷取目前線程的指針
        let pointer = Unmanaged.passUnretained(Thread.current).toOpaque()
        // 目前線程中調用（注冊）的次數
        let count = (_threads[pointer] ?? 0) + 1
        
        // 檢測是否同時被多個線程執行
        if count > 1 {
            synchronizationError("error")
        }
        
        _threads[pointer] = count

        if _threads.count > 1 {
            synchronizationError("error")
        }
    }

    // 在同步方法内結束執行前清除記錄（清除注冊資訊）
    func unregister() {
        _lock.lock(); defer { _lock.unlock() }
        let pointer = Unmanaged.passUnretained(Thread.current).toOpaque()
        _threads[pointer] = (_threads[pointer] ?? 1) - 1
        if _threads[pointer] == 0 {
            _threads[pointer] = nil
        }
    }
}
複制代碼
           

多線程下的 BOOL 判斷

我們經常用 BOOL 來表示 true 和 false，在多線程環境下又需要考慮并發的問題，是以在設定和讀取 BOOL 值的時候就需要考慮用鎖了，但是同時又帶來了性能的損失。有沒有代替方案呢？ RxSwift 使用了一個 Int32 值，并以原子操作的方式對 Int32 值進行修改，確定了線程安全，又不損失性能。

原子操作 Int32 類型的主要方法有：

1. 對比并交換 Int32 值。如果 __theValue 指向的記憶體值和 __oldValue 相等，則修改為 __newValue 值。

public func OSAtomicCompareAndSwap32Barrier(_ __oldValue: Int32, _ __newValue: Int32, _ __theValue: UnsafeMutablePointer<Int32>!) -> Bool
複制代碼
           

1. 将 __theValue 指向的記憶體值進行加1操作

public func OSAtomicIncrement32Barrier(_ __theValue: UnsafeMutablePointer<Int32>!) -> Int32
複制代碼
           

1. 将 __theValue 指向的記憶體值進行減1操作

public func OSAtomicDecrement32Barrier(_ __theValue: UnsafeMutablePointer<Int32>!) -> Int32
複制代碼
           

1. 或操作

public func OSAtomicOr32OrigBarrier(_ __theMask: UInt32, _ __theValue: UnsafeMutablePointer<UInt32>!) -> Int32
複制代碼
           

1. 異或操作

public func OSAtomicXor32OrigBarrier(_ __theMask: UInt32, _ __theValue: UnsafeMutablePointer<UInt32>!) -> Int32
複制代碼
           

下面 RefCountInnerDisposable 類通過 OSAtomicCompareAndSwap32Barrier 将 _isDisposed 和0做比較來判斷 dispose 方法是否調用過。

internal final class RefCountInnerDisposable: DisposeBase, Disposable
{
    private let _parent: RefCountDisposable
    // AtomicInt 為 Int32 的别名
    private var _isDisposed: AtomicInt = 0

    init(_ parent: RefCountDisposable)
    {
        _parent = parent
        super.init()
    }

    internal func dispose()
    {
        // 和 0 做對比，來判斷是否調用過
        if AtomicCompareAndSwap(0, 1, &_isDisposed) {
            _parent.release()
        }
    }
}
複制代碼