PromiseKit 源码阅读

在 PromiseKit 的实现当中，有两个核心的类 Promise 和 Guarantee，这两个类的差别为：

  • Promise 表示任务的执行结果有两种：一种是成功，一种是失败（结果值为 error)，
  • Guarantee 表示任务的执行结果只能为成功。

Guarantee 实现了 Thenable 协议，Promise 实现了 Thenable, CatchMixin 协议

下面我们先来分析 Guarantee 的实现。

Guarantee

简单的说，Guarantee 负责管理一组待处理的事件，当收到结果时，用这个结果去调用这一组待处理的任务。

Guarantee 中待处理的事件用 Handlers 类管理，Handlers 的定义如下：

final class Handlers<R> {
    var bodies: [(R) -> Void] = []
    func append(_ item: @escaping(R) -> Void) { bodies.append(item) }
}

可以看出 Handlers 内部就是一个包含待处理事件的数组。

Guarantee 有两种状态：

  1. 待处理状态（pending）：此时其中包含着一组待处理的任务
  2. 已处理状态（resolved）：此时其中包含着一个用于处理任务的结果值

这两种状态用 Sealant 来表示，定义如下：

enum Sealant<R> {
    case pending(Handlers<R>)
    case resolved(R)
}

PromiseKit 是线程安全的。在修改 Guarantee 的状态或者添加待处理的事件时，都需要考虑到线程安全的问题。所以在获取或修改 Guarantee 的 Sealant 状态时，需要保证线程安全，为了解决这个问题 Sealant 的获取和设置过程使用了 Box 进行了封装：

class Box<T> {
    func inspect() -> Sealant<T> { fatalError() } // 获取当前的状态
    func inspect(_: (Sealant<T>) -> Void) { fatalError() } // 获取并修改 Sealant
    func seal(_: T) {} // 设置为指定的值，并进行处理
}

在 PromiseKit 中 Box 有两个子类：

  • SealedBox：初始化时已经有了结果值，即只会返回 resolved 状态
  • EmptyBox ：初始化时还没有结果值
final class SealedBox<T>: Box<T> {
    let value: T

    init(value: T) {
        self.value = value
    }

    // SealedBox 没有重写 inspect(_: (Sealant<T>) -> Void) 的方法
    // 因为 SealedBox 不应该调用这个方法，因为只有在 Sealant 为 padding 时，才需要调用
    override func inspect() -> Sealant<T> {
        return .resolved(value)
    }
}

class EmptyBox<T>: Box<T> {
    private var sealant = Sealant<T>.pending(.init())
    private let barrier = DispatchQueue(label: "org.promisekit.barrier", attributes: .concurrent)

    // 设置结果值
    // 1. 修改状态 .pending -> .resolved
    // 2. 处理待处理的任务
    override func seal(_ value: T) {
        var handlers: Handlers<T>!
        barrier.sync(flags: .barrier) {
            guard case .pending(let _handlers) = self.sealant else {
                return  // already fulfilled!
            }
            handlers = _handlers
            self.sealant = .resolved(value)
        }

        //FIXME we are resolved so should `pipe(to:)` be called at this instant, “thens are called in order” would be invalid
        //NOTE we don’t do this in the above `sync` because that could potentially deadlock
        //THOUGH since `then` etc. typically invoke after a run-loop cycle, this issue is somewhat less severe

        if let handlers = handlers {
            handlers.bodies.forEach{ $0(value) }
        }

        //TODO solution is an unfortunate third state “sealed” where then's get added
        // to a separate handler pool for that state
        // any other solution has potential races
    }

    override func inspect() -> Sealant<T> {
        var rv: Sealant<T>!
        barrier.sync {
            rv = self.sealant
        }
        return rv
    }

    override func inspect(_ body: (Sealant<T>) -> Void) {
        var sealed = false
        barrier.sync(flags: .barrier) {
            switch sealant {
            case .pending:
                // body will append to handlers, so we must stay barrier’d
                body(sealant)
            case .resolved:
                sealed = true
            }
        }
        if sealed {
            // we do this outside the barrier to prevent potential deadlocks
            // it's safe because we never transition away from this state
            // 我们在 barrier 之外执行这个操作，来避免潜在的死锁
            // 这是安全的，因为我们不会将这个状态进行转换
            body(sealant)
        }
    }
}

Guarantee 中就管理着一个 Box 实例：box，通过 pipe(to: @escaping(Result<T>) -> Void) 向其中添加任务。通过调用 box.seal(:T) 的方法修改 Guarantee 的状态，并处理待执行的任务。

简单来描述 Guarantee 的实现原理就是：在一个数组中存放待执行的任务，等到获取到结果后执行待处理的任务，并清除待执行的任务列表，保存获取到的结果值。

调用链

为了让 Guarantee 的结果值进行传递，对 Guarantee 扩展了很多方法来方便链式调用，比如：then，map 等。

map

下面是 map 扩展方法的实现：

func map<U>(on: DispatchQueue? = conf.Q.map,
            flags: DispatchWorkItemFlags? = nil,
            _ body: @escaping(T) -> U) -> Guarantee<U> {
    let rg = Guarantee<U>(.pending)
    pipe { value in
          on.async(flags: flags) {
              rg.box.seal(body(value))
          }
    }
    return rg
}

在 map 中，给当前的 Guarantee 添加了一个新的任务，即当获取值时，将值通过 body 处理之后设置为新创建 Guarantee 实例 rg 的结果值。

then

下面是 then 方法的实现方法，定义如下：

@discardableResult
func then<U>(on: DispatchQueue? = conf.Q.map, 
             flags: DispatchWorkItemFlags? = nil, 
             _ body: @escaping(T) -> Guarantee<U>) -> Guarantee<U> {
    let rg = Guarantee<U>(.pending)
    pipe { value in
          on.async(flags: flags) {
              body(value).pipe(to: rg.box.seal)
          }
    }
    return rg
}

在这个扩展的方法中，一共牵涉到了三个 Guarantee

  • 调用 then 方法的当前 Guarantee
  • 在 then 方法中创建并返回的 Guarantee
  • 由参数 body 闭包返回的 Guarantee

这个三个 Guarantee 的关系为：通过调用 then 方法，给当前的 Guarantee 添加了一个待执行的任务，即调用 body 获取一个 Guarantee 并向其添加了一个完成 then 方法返回的 Guarantee 的任务。

所以，then 方法中返回的 Guarantee 需要等到从 body 中返回的 Guarantee 处理完成时才处理，而从 body 中返回的 Guarantee 需要等到当前的 Guarantee 处理完之后才处理。

为了应对不同的处理方式，PromiseKit 创建了多种方法。通过不同的方法名，可以让程序的可读性在代码层面得到提高。

Promise

对比 Guarantee，Promise 的执行结果可能是成功（获取到一个值），可能是发生异常（获取到一个 Error），所以在定义上， Promise 和 Guarantee 主要有两点不同：

  • Promise 的结果有两种，成功或失败。
  • Promise 不仅实现了 Thenable 协议，还实现了 CatchMixin 协议，用于处理异常的情况，

Promise 中的 Box 中的泛型就是 Result。 Result 定义如下：

public enum Result<T> {
    case fulfilled(T)
    case rejected(Error)
}

CatchMixin 异常处理

通过对 CatchMixin 进行扩展，提供了 catch 方法，定义入下：

public extension CatchMixin {

    @discardableResult
    func `catch`(on: DispatchQueue? = conf.Q.return, flags: DispatchWorkItemFlags? = nil, policy: CatchPolicy = conf.catchPolicy, _ body: @escaping(Error) -> Void) -> PMKFinalizer {
        let finalizer = PMKFinalizer()
        pipe {
            switch $0 {
            case .rejected(let error):
                guard policy == .allErrors || !error.isCancelled else {
                    fallthrough
                }
                on.async(flags: flags) {
                    body(error)
                    finalizer.pending.resolve(())
                }
            case .fulfilled:
                finalizer.pending.resolve(())
            }
        }
        return finalizer
    }
}

public class PMKFinalizer {
    let pending = Guarantee<Void>.pending()

    /// `finally` is the same as `ensure`, but it is not chainable
    public func finally(on: DispatchQueue? = conf.Q.return, flags: DispatchWorkItemFlags? = nil, _ body: @escaping () -> Void) {
        
        pending.guarantee.done(on: on, flags: flags) {
            body()
        }
        
    }
}

在 catch 方法中，提供 body 参数传入一个处理 error 的处理程序，并返回一个 PMKFinalizer 实例。PMKFinalizer 类的作用有两个：

  1. 终结调用链
  2. 传入一个闭包，这个闭包不管 Promise 执行的结果是成功还是失败都会调用。

当发生异常的时候，还可以通过 error 尝试进行恢复：

func recover<U: Thenable>(on: DispatchQueue? = conf.Q.map, flags: DispatchWorkItemFlags? = nil, policy: CatchPolicy = conf.catchPolicy, _ body: @escaping(Error) throws -> U) -> Promise<T> where U.T == T {
    let rp = Promise<U.T>(.pending)
    pipe {
        switch $0 {
        case .fulfilled(let value):
            rp.box.seal(.fulfilled(value))
        case .rejected(let error):
            if policy == .allErrors || !error.isCancelled {
                on.async(flags: flags) {
                    do {
                        let rv = try body(error)
                        guard rv !== rp else { throw PMKError.returnedSelf }
                        rv.pipe(to: rp.box.seal)
                    } catch {
                        rp.box.seal(.rejected(error))
                    }
                }
            } else {
                rp.box.seal(.rejected(error))
            }
        }
    }
    return rp
}

特性方法

wait、hang —— 异步转同步

PromiseKit 是用来应对异步程序的，但是有时候为了方便测试，希望将异步转化为同步，方便的获取到结果值，所以给 Promise 扩展了 wait 方法。下面是 wait 方法的实现：

/**
Blocks this thread, so—you know—don’t call this on a serial thread that
any part of your chain may use. Like the main thread for example.
*/
func wait() throws -> T {

    if Thread.isMainThread {
        conf.logHandler(LogEvent.waitOnMainThread)
    }

    var result = self.result

    if result == nil {
        let group = DispatchGroup()
        group.enter()
        pipe { result = $0; group.leave() }
        group.wait()
    }

    switch result! {
        case .rejected(let error):
        throw error
        case .fulfilled(let value):
        return value
    }
}

wait 方法通过 DispatchGroup 实现了将异步转化为同步，实现思路为：

  1. 创建一个 grounp
  2. 调用 grounp 的 enter 方法
  3. 通过 pipe 方法给当前的 promise 添加一个待处理的任务：调用 grounp 的 leave 方法
  4. 调用 grounp 的 wait 方法

在 PromiseKit 中，同时给 Promise 和 Guarantee 都提供了 wait 方法。区别是 Promise 的 wait 方法会抛出异常，Guarantee 的 wait 不会抛出异常。

除了 wait 方法外，还提供了一个全局方法 hang，定义如下：

public func hang<T>(_ promise: Promise<T>) throws -> T {
    #if os(Linux) || os(Android)
    
    #if swift(>=4.2)
    let runLoopMode: CFRunLoopMode = kCFRunLoopDefaultMode
    #else
    // isMainThread is not yet implemented on Linux.
    let runLoopModeRaw = RunLoopMode.defaultRunLoopMode.rawValue._bridgeToObjectiveC()
    let runLoopMode: CFString = unsafeBitCast(runLoopModeRaw, to: CFString.self)
    #endif
    
    #else
    guard Thread.isMainThread else {
        // hang doesn't make sense on threads that aren't the main thread.
        // use `.wait()` on those threads.
        fatalError("Only call hang() on the main thread.")
    }
    let runLoopMode: CFRunLoopMode = CFRunLoopMode.defaultMode
    #endif
    
    if promise.isPending {
        var context = CFRunLoopSourceContext()
        let runLoop = CFRunLoopGetCurrent()
        let runLoopSource = CFRunLoopSourceCreate(nil, 0, &context)
        CFRunLoopAddSource(runLoop, runLoopSource, runLoopMode)
        
        _ = promise.ensure {
            CFRunLoopStop(runLoop)
        }
        
        while promise.isPending {
            CFRunLoopRun()
        }
        CFRunLoopRemoveSource(runLoop, runLoopSource, runLoopMode)
    }
    
    switch promise.result! {
    case .rejected(let error):
        throw error
    case .fulfilled(let value):
        return value
    }
}

hang 通过 RunLoop 原理实现了异步转同步。

实际上 RunLoop 就是这样一个函数，其内部是一个 do-while 循环。当你调用 CFRunLoopRun() 时，线程就会一直停留在这个循环里；直到超时或被手动停止，该函数才会返回。

实现过程为：

  1. 在当前的 RunLoop 的 defaultMode 中添加一个 Source，防止 RunLoop 开始之后退出
  2. 在当前的 promise 中添加一个待处理事件，即调用 CFRunLoopStop 来退出 RunLoop
  3. 在当前的 promise 为 pending 状态时，调用 CFRunLoopRun() 进入 RunLoop 循环
  4. 退出 RunLoop 循环之后，移除之前添加的 Source

关于 RunLoop 更多的内容可以查看下面的文章：

https://blog.ibireme.com/2015/05/18/runloop/
https://blog.devtang.com/2012/06/24/enhance-uiactionsheet/

after 等待指定的时间

有时需要等待一段指定的时间再去执行，after 就可以快速的返回这样一个 Guarantee。

/**
     after(seconds: 1.5).then {
         //…
     }

- Returns: A guarantee that resolves after the specified duration.
*/
public func after(seconds: TimeInterval) -> Guarantee<Void> {
    let (rg, seal) = Guarantee<Void>.pending()
    let when = DispatchTime.now() + seconds
#if swift(>=4.0)
    q.asyncAfter(deadline: when) { seal(()) }
#else
    q.asyncAfter(deadline: when, execute: seal)
#endif
    return rg
}

/**
     after(.seconds(2)).then {
         //…
     }

 - Returns: A guarantee that resolves after the specified duration.
*/
public func after(_ interval: DispatchTimeInterval) -> Guarantee<Void> {
    let (rg, seal) = Guarantee<Void>.pending()
    let when = DispatchTime.now() + interval
#if swift(>=4.0)
    q.asyncAfter(deadline: when) { seal(()) }
#else
    q.asyncAfter(deadline: when, execute: seal)
#endif
    return rg
}

在 after 方法内部就是通过 DispatchQueue 的 asyncAfter 方法来实现的。

firstly

为了增强代码的可读性，PromiseKit 提供了一个 firstly 方法，先来看一下不使用 firstly 和使用 firstly 的代码对比：

// 不使用的情况
URLSession.shared.dataTask(url: url1).then {
    URLSession.shared.dataTask(url: url2)
}.then {
    URLSession.shared.dataTask(url: url3)
}

// 使用 firstly 的情况
firstly {
    URLSession.shared.dataTask(url: url1)
}.then {
    URLSession.shared.dataTask(url: url2)
}.then {
    URLSession.shared.dataTask(url: url3)
}

使用 first 之后，确实可以提高代码的可读性。firstly 的实现方式如下：

public func firstly<U: Thenable>(execute body: () throws -> U) -> Promise<U.T> {
    do {
        let rp = Promise<U.T>(.pending)
        try body().pipe(to: rp.box.seal)
        return rp
    } catch {
        return Promise(error: error)
    }
}

/// - See: firstly()
public func firstly<T>(execute body: () -> Guarantee<T>) -> Guarantee<T> {
    return body()
}

race

在有时候我们需要在多个异步程序中，获取到最先执行结束的结果值，这是就可以使用 race 方法。race 方法有几个变体，我们来看其中的一个：

public func race<T>(_ guarantees: Guarantee<T>...) -> Guarantee<T> {
    let rg = Guarantee<T>(.pending)
    for guarantee in guarantees {
        guarantee.pipe(to: rg.box.seal)
    }
    return rg
}

在 race 中创建了一个新的 Guarantee，在所有参与竞争的 Guarantee 中添加一个待处理程序，即设置这个新创建 Guarantee 的结果。由于 Guarantee 只会保存第一次设置的结果值，而忽略掉以后再设置的结果值，所以新创建的 Guarantee 中就保存了最先执行完的结果值。

when

有时需要同时获取多个异步处理程序的执行结果，所有需要等到所有的异步程序都完成，这种场景就可以通过 when 来实现。同样 when 的方法也有多个变型，下面是其中一个：

private func _when<U: Thenable>(_ thenables: [U]) -> Promise<Void> {
    var countdown = thenables.count
    guard countdown > 0 else {
        return .value(Void())
    }

    let rp = Promise<Void>(.pending)

#if PMKDisableProgress || os(Linux)
    var progress: (completedUnitCount: Int, totalUnitCount: Int) = (0, 0)
#else
    let progress = Progress(totalUnitCount: Int64(thenables.count))
    progress.isCancellable = false
    progress.isPausable = false
#endif

    let barrier = DispatchQueue(label: "org.promisekit.barrier.when", attributes: .concurrent)

    for promise in thenables {
        promise.pipe { result in
            barrier.sync(flags: .barrier) {
                switch result {
                case .rejected(let error):
                    if rp.isPending {
                        progress.completedUnitCount = progress.totalUnitCount
                        rp.box.seal(.rejected(error))
                    }
                case .fulfilled:
                    guard rp.isPending else { return }
                    progress.completedUnitCount += 1
                    countdown -= 1
                    if countdown == 0 {
                        rp.box.seal(.fulfilled(()))
                    }
                }
            }
        }
    }

    return rp
}

public func when<U: Thenable>(fulfilled thenables: [U]) -> Promise<[U.T]> {
    return _when(thenables).map(on: nil) { thenables.map{ $0.value! } }
}

其中核心方法是 _when，设计思路为：新创建一个 Promise，每当一个需要等待的 Promise 执行完之后，查看是否还有未执行结束的 Promise，如果全部执行完，则设置新创建的 Promise 的结果值。

技巧

Resolver 在必要的时候输出警告

Promise 有个初始化方法：

/// - Returns: a tuple of a new pending promise and its `Resolver`.
public class func pending() -> (promise: Promise<T>, resolver: Resolver<T>) {
    return { ($0, Resolver($0.box)) }(Promise<T>(.pending))
}

Resolver 的定义如下：

/// An object for resolving promises
public final class Resolver<T> {
    let box: Box<Result<T>>

    init(_ box: Box<Result<T>>) {
        self.box = box
    }

    deinit {
        if case .pending = box.inspect() {
            conf.logHandler(.pendingPromiseDeallocated)
        }
    }
}

看起来 Resolver 仅仅是将 Box 包装了一层，为什么要这样做呢？

在 PromiseKit 中，在 Promise 释放时，如果 Promise 的状态为未处理（pending），则会在控制台输出一条警告信息，来提示用户可能存在问题。

我想这样设计的主要目的是来满足这个需求的。

为什么不用子类呢？为什么不把这个判断放到 Box 类中呢？

扩展系统 API 的注意事项

为了方便的在指定的线程上创建一个 Guarantee，所以作者对 DispatchQueue 进行了扩展，但是在扩展的时候，可能会和其他库中的扩展重名，也可能在将来会和苹果的 API 重名，这样就会导致难以排查的 bug。PromiseKit 的做法是 在方法中添加一个独有类型的参数。代码如下：

public extension DispatchQueue {
    /**
     Asynchronously executes the provided closure on a dispatch queue.

         DispatchQueue.global().async(.promise) {
             md5(input)
         }.done { md5 in
             //…
         }

     - Parameter body: The closure that resolves this promise.
     - Returns: A new `Guarantee` resolved by the result of the provided closure.
     - Note: There is no Promise/Thenable version of this due to Swift compiler ambiguity issues.
     */
    @available(macOS 10.10, iOS 2.0, tvOS 10.0, watchOS 2.0, *)
    final func async<T>(_: PMKNamespacer, group: DispatchGroup? = nil, qos: DispatchQoS = .default, flags: DispatchWorkItemFlags = [], execute body: @escaping () -> T) -> Guarantee<T> {
        let rg = Guarantee<T>(.pending)
        async(group: group, qos: qos, flags: flags) {
            rg.box.seal(body())
        }
        return rg
    }
}

其中 PMKNamespacer 的定义如下：

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