Combine中錯(cuò)誤處理和Scheduler使用詳解

更新時(shí)間：2022年12月26日 10:19:54 作者：Layer

這篇文章主要為大家介紹了Combine中錯(cuò)誤處理和Scheduler使用示例詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

錯(cuò)誤處理

到目前為止，在我們編寫(xiě)的大部分代碼中，我們沒(méi)有處理錯(cuò)誤，而處理的都是“happy path”。在前面的文章中，我們了解到，Combine Publisher 聲明了兩個(gè)約束：

Output定義 Publisher 發(fā)出的值的類(lèi)型;
Failure 定義 Publisher 發(fā)出的失敗的類(lèi)型。

現(xiàn)在，我們將深入了解 Failure 在 Publisher 中的作用。

Never

失敗類(lèi)型為 Never 的 Publisher 表示永遠(yuǎn)不會(huì)發(fā)出失敗。它為這些 Publisher 提供了強(qiáng)大的保證。這類(lèi) Publisher 可讓我們專(zhuān)注于使用值，同時(shí)絕對(duì)確保 Publisher 只有成功完成的事件。

在新的 Playground 頁(yè)面添加以下代碼：

import Combine
import Foundation
import PlaygroundSupport
PlaygroundPage.current.needsIndefiniteExecution = true
func example(_ desc: String, _ action:() -> Void) {
    print("--- (desc) ---")
    action()
}
var subscriptions = Set<AnyCancellable>()
example("Just") {
  Just("Hello")
}

我們創(chuàng)建了一個(gè)帶有 Hello 字符串值的 Just。 Just 是不會(huì)發(fā)出失敗的。請(qǐng)按住 Command 并單擊 Just 初始化程序并選擇 Jump to Definition，查看定義：

In contrast with Result.Publisher, a Just publisher can’t fail with an error. And unlike Optional.Publisher, a Just publisher always produces a value.

Combine 對(duì) Never 的障保證不僅是理論上的，而是深深植根于框架及其各種 API 中。Combine 提供了幾個(gè) Operator，這些 Operator 僅在保證 Publisher 永遠(yuǎn)不會(huì)發(fā)出失敗事件時(shí)才可用。第一個(gè)是 sink 的變體，只處理值：

example("Just") {
  Just("Hello")
    .sink(receiveValue: { print($0) })
    .store(in: &subscriptions)
}

在上面的示例中，我們使用 sink(receiveValue:) ，這種特定的重載使我們可以忽略 Publisher 的完成事件，而只處理其發(fā)出的值。

此重載僅適用于這類(lèi)“可靠”的 Publisher。在錯(cuò)誤處理方面，Combine 是智能且安全的，如果可能拋出錯(cuò)誤，它會(huì)強(qiáng)制我們處理完成事件。要看到這一點(diǎn)，我們需要將 Never 的 Publisher 變成可能發(fā)出失敗事件的 Publisher。

setFailureType(to:)

func setFailureType<E>(to failureType: E.Type) -> Publishers.SetFailureType<Self, E> where E : Error

將 Never Publisher 轉(zhuǎn)變?yōu)榭赡馨l(fā)出失敗事件的 Publisher 的第一種方法是使用 setFailureType。這是另一個(gè)僅適用于失敗類(lèi)型為 Never 的 Publisher 的 Operator：

example("setFailureType") {
 &nbsp;Just("Hello")
 &nbsp; &nbsp;.setFailureType(to: MyError.self)
}

可以使用 .eraseToAnyPublisher()，來(lái)確認(rèn)已改變的 Publisher 類(lèi)型：

繼續(xù)修改上述代碼：

enum MyError: Error {
    case ohNo
}
example("setFailureType") {
    Just("Hello")
        .setFailureType(to: MyError.self)
        .sink(
            receiveCompletion: { completion in
                switch completion {
                case .failure(.ohNo):
                    print("Finished with OhNo!")
                case .finished:
                    print("Finished successfully!")
                }
            },
            receiveValue: { value in
                print("Got value: (value)")
            }
        )
        .store(in: &subscriptions)
}

現(xiàn)在我們只能使用 sink(receiveCompletion:receiveValue:)。 sink(receiveValue:) 重載不再可用，因?yàn)榇?Publisher 可能會(huì)發(fā)出失敗事件?？梢試L試注釋掉 receiveCompletion查看編譯錯(cuò)誤。

此外，失敗類(lèi)型為為 MyError，這使我們可以針對(duì).failure(.ohNo) 情況而無(wú)需進(jìn)行不必要的強(qiáng)制轉(zhuǎn)換來(lái)處理該錯(cuò)誤。

當(dāng)然，setFailureType 的作用只是類(lèi)型定義。由于原始 Publisher 是 Just，因此實(shí)際上也不會(huì)引發(fā)任何錯(cuò)誤。

assign(to:on:)

assign Operator 僅適用于不會(huì)發(fā)出失敗事件的 Publisher，與 setFailureType 相同。向提供的 keypath 發(fā)送錯(cuò)誤會(huì)導(dǎo)致未定義的行為。添加以下示例進(jìn)行測(cè)試：

example("assign(to:on:)") {
    class Person {
        var name = "Unknown"
    }
    let person = Person()
    print(person.name)
    Just("Layer")
        .handleEvents(
            receiveCompletion: { _ in 
                print(person.name) 
            }
        )
        .assign(to: .name, on: person)
        .store(in: &subscriptions)
}

我們定義一個(gè)具有 name 屬性的 Person 類(lèi)。創(chuàng)建一個(gè) Person 實(shí)例并立即打印其 name。一旦 Publisher 發(fā)送完成事件，使用 handleEvents 再次打印此 name。最后，使用 assign 將 name 設(shè)置為 Publisher 發(fā)出的值：

--- assign(to:on:) ---
Unknown
Layer

在 Just("Layer") 正下方添加以下行：

.setFailureType(to: Error.self)

這意味著它不再是 Publisher<String, Never>，而是現(xiàn)在的 Publisher<String, Error>。運(yùn)行 Playground，我們將進(jìn)行驗(yàn)證：

Referencing instance method 'assign(to:on:)' on 'Publisher' requires the types 'any Error' and 'Never' be equivalent

assign(to:)

assign(to:on:) 有一個(gè)棘手的部分——它會(huì) strong 捕獲提供給 on 參數(shù)的對(duì)象。在上一個(gè)示例之后添加以下代碼：

example("assign(to:)") {
  class MyViewModel: ObservableObject {
    @Published var currentDate = Date()
    init() {
      Timer.publish(every: 1, on: .main, in: .common)
        .autoconnect() 
        .prefix(3)
        .assign(to: .currentDate, on: self)
        .store(in: &subscriptions)
    }
  }
  let vm = MyViewModel()
  vm.$currentDate
    .sink(receiveValue: { print($0) })
    .store(in: &subscriptions)
}

我們 MyViewModel 中定義一個(gè) @Published 屬性。它的初始值為當(dāng)前日期。在 init 中創(chuàng)建一個(gè) Timer Publisher，它每秒發(fā)出當(dāng)前日期。使用 prefix Operator 只接受 3 個(gè)更新。使用 assign(to:on:) 將每個(gè)日期更新給@Published 屬性。實(shí)例化 MyViewModel，sink vm.$currentDate，并打印出每個(gè)值：

--- assign(to:) ---
2022-12-24 07:32:33 +0000
2022-12-24 07:32:34 +0000
2022-12-24 07:32:35 +0000
2022-12-24 07:32:36 +0000

看起來(lái)一切都很好。但是對(duì)assign(to:on:) 的調(diào)用創(chuàng)建了一個(gè) strong 持有 self 的 Subscription。導(dǎo)致 self 掛在Subscription 上，而 Subscription 掛在 self 上，創(chuàng)建了一個(gè)導(dǎo)致內(nèi)存泄漏的引用循環(huán)。

因此引入了該 Operator 的另一個(gè)重載 assign(to:)。該 Operator 通過(guò)對(duì) Publisher 的 inout 引用來(lái)將值分配給 @Published 屬性。因此以下兩行：

.assign(to: .currentDate, on: self)
.store(in: &subscriptions)

可以被替換為：

.assign(to: &$currentDate)

使用 assign(to:) Operator 將 inout 引用 Publisher 會(huì)打破引用循環(huán)。此外，它會(huì)在內(nèi)部自動(dòng)處理 Subscription 的內(nèi)存管理，這樣我們就可以省略 store(in: &subscriptions)。

assertNoFailure(_:file:line:)

當(dāng)我們?cè)陂_(kāi)發(fā)過(guò)程確認(rèn) Publisher 以失敗事件完成時(shí)，assertNoFailure Operator 非常有用。它不會(huì)阻止上游發(fā)出失敗事件。但是，如果它檢測(cè)到錯(cuò)誤，它會(huì)因錯(cuò)誤而崩潰：

example("assertNoFailure") {
  Just("Hello")
    .setFailureType(to: MyError.self)
    .assertNoFailure()
    .sink(receiveValue: { print("Got value: ($0) ")}) 
    .store(in: &subscriptions)
}

我們使用 Just 創(chuàng)建一個(gè)“可靠”的 Publisher 并將其錯(cuò)誤類(lèi)型設(shè)置為 MyError。如果 Publisher 以錯(cuò)誤事件完成，則使用 assertNoFailure 以崩潰。這會(huì)將 Publisher 的失敗類(lèi)型轉(zhuǎn)回 Never。使用 sink 打印出任何接收到的值。請(qǐng)注意，由于 assertNoFailure 將失敗類(lèi)型設(shè)置回 Never，因此 sink(receiveValue:) 重載可以直接使用。

運(yùn)行 Playground，它可以正常工作：

--- assertNoFailure ---
Got value: Hello

在 setFailureType 之后，添加以下行：

.tryMap { _ in throw MyError.ohNo }

一旦 Hello 被推送到下游，使用 tryMap 拋出錯(cuò)誤。再次運(yùn)行 Playground：

Playground execution failed:
error: Execution was interrupted, reason: EXC_BAD_INSTRUCTION (code=EXC_I386_INVOP, subcode=0x0).
...
frame #0: 0x00007fff232fbbf2 Combine`Combine.Publishers.AssertNoFailure...

由于 Publisher 發(fā)出失敗事件，playground 會(huì) crash。在某種程度上，我們可以將 assertNoFailure() 視為代碼的保護(hù)機(jī)制。雖然我們不應(yīng)該在生產(chǎn)環(huán)境中使用它，但在開(kāi)發(fā)過(guò)程中提前發(fā)現(xiàn)問(wèn)題非常有用。

處理錯(cuò)誤

try* Operator

Combine 提供了一個(gè)區(qū)分可能引發(fā)錯(cuò)誤和可能不會(huì)引發(fā)錯(cuò)誤的 Operator 的方法：try 前綴。

注意：Combine 中所有以 try 為前綴的 Operator 在遇到錯(cuò)誤時(shí)的行為相同。我們將只在本章中嘗試使用 tryMap Operator。

example("tryMap") {
    enum NameError: Error {
        case tooShort(String)
        case unknown
    }
    
    ["Aaaa", "Bbbbb", "Cccccc"]
        .publisher
        .map { value in
            return value.count
        }
        .sink(
            receiveCompletion: { print("Completed with ($0)") },
            receiveValue: { print("Got value: ($0)") }
        )
}

在上面的示例中，我們定義一個(gè) NameError 錯(cuò)誤枚舉。創(chuàng)建發(fā)布三個(gè)字符串的 Publisher。將每個(gè)字符串映射到它的長(zhǎng)度。運(yùn)行示例并查看控制臺(tái)輸出：

--- tryMap ---
Got value: 4
Got value: 5
Got value: 6
Completed with finished

將上面示例中的 map 替換為以下內(nèi)容：

.tryMap { value -> Int in
    let length = value.count
    guard length >= 5 else {
        throw NameError.tooShort(value)
    }
    return value.count
}

我們檢查字符串的長(zhǎng)度是否大于等于 5。否則，我們會(huì)拋出錯(cuò)誤：

--- tryMap ---
Completed with failure(Page_Contents.(unknown context at $10e3cb984).(unknown context at $10e3cba6c).(unknown context at $10e3cbaa8).NameError.tooShort("Aaaa"))

映射錯(cuò)誤

map 和 tryMap 之間的區(qū)別不僅僅是后者允許拋出錯(cuò)誤。 map 繼承了現(xiàn)有的失敗類(lèi)型并且只操作 Publisher 的值，但 tryMap 沒(méi)有——它實(shí)際上將錯(cuò)誤類(lèi)型擦除為普通的 Swift 錯(cuò)誤。與帶有 try 前綴的所有 Operator 都是如此。

example("map vs tryMap") {
  enum NameError: Error {
    case tooShort(String)
    case unknown
  }
  Just("Hello")
    .setFailureType(to: NameError.self)
    .map { $0 + " World!" }
    .sink(
      receiveCompletion: { completion in
        switch completion {
        case .finished:
          print("Done!")
        case .failure(.tooShort(let name)):
          print("(name) is too short!")
        case .failure(.unknown):
          print("An unknown name error occurred")
        }
      },
      receiveValue: { print("Got value ($0)") }
    )
    .store(in: &subscriptions)
}

我們定義一個(gè)用于此示例的 NameError。創(chuàng)建一個(gè)只發(fā)出字符串 Hello 的 Just。使用 setFailureType 設(shè)置失敗類(lèi)型為 NameError。使用 map 將另一個(gè)字符串附加。最后，使用 sink 的 receiveCompletion 為 NameError 的每個(gè)情況打印出適當(dāng)?shù)南?。運(yùn)行 Playground：

--- map vs tryMap ---
Got value Hello World!
Done!

Completion 的失敗類(lèi)型是 NameError，這正是我們想要的。 setFailureType 允許我們專(zhuān)門(mén)針對(duì) NameError 進(jìn)行處理，例如 failure(.tooShort(let name))。

將 map 更改為 tryMap。

.tryMap { throw NameError.tooShort($0) }

我們會(huì)立即注意到 Playground 不再編譯。再次點(diǎn)擊 completion：

tryMap 刪除了我們的類(lèi)型錯(cuò)誤并將其替換為通用 Swift.Error 類(lèi)型。即使我們實(shí)際上并沒(méi)有從 tryMap 中拋出錯(cuò)誤，也會(huì)發(fā)生這種情況。

原因很簡(jiǎn)單：Swift 還不支持類(lèi)型化 throws，盡管自 2015 年以來(lái) Swift Evolution 中一直在討論這個(gè)主題。這意味著當(dāng)我們使用帶有 try 前綴的 Operator 時(shí)，我們的錯(cuò)誤類(lèi)型將總是被抹去到最常見(jiàn)的父類(lèi)：Swift.Error。

一種方法是將通用錯(cuò)誤手動(dòng)轉(zhuǎn)換為特定的錯(cuò)誤類(lèi)型，但這不是最理想的。它打破了嚴(yán)格類(lèi)型錯(cuò)誤的整個(gè)目的。幸運(yùn)的是，Combine 為這個(gè)問(wèn)題提供了一個(gè)很好的解決方案，稱(chēng)為 mapError。

在調(diào)用 tryMap 之后，添加以下行：

.mapError { $0 as? NameError ?? .unknown }

mapError 接收上游 Publisher 拋出的任何錯(cuò)誤，并將其映射到我們想要的任何錯(cuò)誤。在這種情況下，我們可以利用它將錯(cuò)誤轉(zhuǎn)換回 NameError。這會(huì)將 Failure 恢復(fù)為所需要的類(lèi)型，并將我們的 Publisher 轉(zhuǎn)回 Publisher<String, NameError>。構(gòu)建并運(yùn)行 Playground，最終可以按預(yù)期編譯和工作：

--- map vs tryMap ---
Hello is too short!

捕獲錯(cuò)誤并重試

很多時(shí)候，當(dāng)我們請(qǐng)求資源或執(zhí)行某些計(jì)算時(shí)，失敗可能是由于網(wǎng)絡(luò)不穩(wěn)定或其他資源不可用而導(dǎo)致的一次性事件。

在這些情況下，我們通常會(huì)編寫(xiě)一個(gè)機(jī)制來(lái)重試不同的工作，跟蹤嘗試次數(shù)，并處理如果所有嘗試都失敗的情況。Combine 讓這一切變得非常簡(jiǎn)單。

retry Operator 接受一個(gè)數(shù)字。如果 Publisher 失敗，它將重新訂閱上游并重試至我們指定的次數(shù)。如果所有重試都失敗，它將錯(cuò)誤推送到下游，就像沒(méi)有 retry Operator 一樣：

example("Catching and retrying") {
    enum MyError: Error {
        case network
    }
    var service1 = PassthroughSubject<Int, MyError>()    service1.send(completion: .failure(.network))
  
    service1
        .handleEvents(
            receiveSubscription: { _ in print("Trying ...") },
            receiveCompletion: {
                guard case .failure(let error) = $0 else { return }
                print("Got error: (error)")
            }
        )
        .retry(3)
        .sink(
            receiveCompletion: { print("($0)") },
            receiveValue: { number in
                print("Got Number: (number)")
            }
        )
        .store(in: &subscriptions)
}

我們有一個(gè) service1，它發(fā)出了失敗事件。因此，訂閱 service1 肯定會(huì)獲得失敗事件。我們嘗試三次，并通過(guò) handleEvents 打印訂閱和完成：

--- Catching and retrying ---
Trying ...
Got error: network
Trying ...
Got error: network
Trying ...
Got error: network
Trying ...
Got error: network
failure(Page_Contents.(unknown context at $10fc7b584).(unknown context at $10fc7b77c).(unknown context at $10fc7b7b8).MyError.network)

運(yùn)行 Playerground，我們會(huì)看到有四次 Trying。初始 Trying，加上由 retry Operator 觸發(fā)的三次重試。由于 service1 不斷失敗，因此 Operator 會(huì)耗盡所有重試嘗試并將錯(cuò)誤推送到 sink。

調(diào)整代碼：

example("Catching and retrying") {
    enum MyError: Error {
        case network
    }
    var service1 = PassthroughSubject<Int, MyError>()
    service1.send(completion: .failure(.network))
    
    service1
        .handleEvents(
            receiveSubscription: { _ in print("Trying ...") },
            receiveCompletion: {
                guard case .failure(let error) = $0 else { return }
                print("Got error: (error)")
            }
        )
        .retry(3)
        .replaceError(with: 1)
        .sink(
            receiveCompletion: { print("($0)") },
            receiveValue: { number in
                print("Got Number: (number)")
            }
        )
        .store(in: &subscriptions)
}

在 service1 重試后，若還是失敗，我們將通過(guò) replaceError 將失敗替換為 1：

--- Catching and retrying ---
Trying ...
Got error: network
Trying ...
Got error: network
Trying ...
Got error: network
Trying ...
Got error: network
Got Number: 1
finished

或者，我們可以使用 catch 捕獲 service1 的失敗，并為下游提供另一個(gè) Publisher：

example("Catching and retrying") {
    enum MyError: Error {
        case network
    }
    var service1 = PassthroughSubject<Int, MyError>()
    service1.send(completion: .failure(.network))
    var service2 = PassthroughSubject<Int, MyError>()
    
    service1
        .handleEvents(
            receiveSubscription: { _ in print("Trying ...") },
            receiveCompletion: {
                guard case .failure(let error) = $0 else { return }
                print("Got error: (error)")
            }
        )
        .retry(3)
        .catch { error in
            return service2
        }
        .sink(
            receiveCompletion: { print("($0)") },
            receiveValue: { number in
                print("Got Number: (number)")
            }
        )
        .store(in: &subscriptions)
    
    service2.send(2)
    service2.send(completion: .finished)
}

此時(shí)，下游將獲得到 service2 發(fā)出的值 2 和完成事件：

--- Catching and retrying ---
Trying ...
Got error: network
Trying ...
Got error: network
Trying ...
Got error: network
Trying ...
Got error: network
Got Number: 2
finished

cheduler

我們已經(jīng)遇到了一些將 Scheduler 作為參數(shù)的 Operator。大多數(shù)情況下，我們會(huì)簡(jiǎn)單地使用 DispatchQueue.main，因?yàn)樗奖?、易于理解。除?DispatchQueue.main，我們肯定已經(jīng)使用了全局并發(fā)隊(duì)列，或創(chuàng)建一個(gè)串行調(diào)度隊(duì)列來(lái)運(yùn)行操作。

但是為什么 Combine 需要一個(gè)新的類(lèi)似概念呢？我們接著將了解為什么會(huì)出現(xiàn) Scheduler 的概念，將探索 Combine 如何使異步事件和操作更易于使用，當(dāng)然，我們還會(huì)試使用 Combine 提供的所有 Scheduler。

Scheduler 簡(jiǎn)介

根據(jù) Apple 的文檔，Scheduler 是一種定義何時(shí)及如何執(zhí)行閉包的協(xié)議。Scheduler 提供上下文以盡快或在將來(lái)的某個(gè)事件執(zhí)行未來(lái)的操作。該操作就是協(xié)議本身中定義的閉包。閉包也可以隱藏 Publisher 在特定 Scheduler 上執(zhí)行的某些值的傳遞。

我們會(huì)注意到此定義有意避免對(duì)線程的任何引用，這是因?yàn)榫唧w的實(shí)現(xiàn)是在 Scheduler 協(xié)議中，提供的“上下文”中的。因此，我們的代碼將在哪個(gè)線程上執(zhí)行取決于選擇的 Scheduler。

記住這個(gè)重要的概念：Scheduler 不等于線程。我們將在后面詳細(xì)了解這對(duì)每個(gè) Scheduler 意味著什么。讓我們從事件流的角度來(lái)看 Scheduler 的概念：

我們?cè)谏蠄D中看到的內(nèi)容：

在主 (UI) 線程上發(fā)生用戶(hù)操作，如按鈕按下；
它會(huì)觸發(fā)一些工作在 Background Scheduler 上進(jìn)行處理；
要顯示的最終數(shù)據(jù)在主線程上傳遞給 Subscriber，Subscriber 可以更新 UI。

我們可以看到 Scheduler 的概念深深植根于前臺(tái)/后臺(tái)執(zhí)行的概念。此外，根據(jù)我們選擇的實(shí)現(xiàn)，工作可以串行化或并行化。

因此，要全面了解 Scheduler，需要查看哪些類(lèi)符合 Scheduler 協(xié)議。首先，我們需要了解與 Scheduler 相關(guān)的兩個(gè)重要 Operator。

Scheduler Operator

Combine 提供了兩個(gè)基本的 Operator 來(lái)使用 Scheduler：

subscribe(on:) 和 subscribe(on:options:) 在指定的 Scheduler 上創(chuàng)建 Subscription（開(kāi)始工作）；

receive(on:) 和 receive(on:options:) 在指定的 Scheduler 上傳遞值。

此外，以下 Operator 將 Scheduler 和 Scheduler options 作為參數(shù)：

debounce(for:scheduler:options:)

delay(for:tolerance:scheduler:options:)

measureInterval(using:options:)

throttle(for:scheduler:latest:)

timeout(_:scheduler:options:customError:)

subscribe(on:) 和 receive(on:)

在我們訂閱它之前，Publisher 是一個(gè)無(wú)生命的實(shí)體。但是當(dāng)我們訂閱 Publisher 時(shí)會(huì)發(fā)生什么？有幾個(gè)步驟：

Publiser receive Subscriber 并創(chuàng)建 Subscription；
Subscriber receive Subscription 并從 Publiser 請(qǐng)求值（虛線）；
Publiser 開(kāi)始工作（通過(guò) Subscription）；
Publiser 發(fā)出值（通過(guò) Subscription）；
Operator 轉(zhuǎn)換值；
Subscriber 收到最終值。

當(dāng)代碼訂閱 Publiser 時(shí)，步驟一、二和三通常發(fā)生在當(dāng)前線程上。但是當(dāng)我們使用 subscribe(on:) Operator 時(shí)，所有這些操作都在我們指定的 Scheduler 上運(yùn)行。

我們可能希望 Publiser 在后臺(tái)執(zhí)行一些昂貴的計(jì)算以避免阻塞主線程。執(zhí)行此操作的簡(jiǎn)單方法是使用 subscribe(on:)。以下是偽代碼：

let queue = DispatchQueue(label: "serial queue")
let subscription = publisher
  .subscribe(on: queue)
  .sink { value in ...

如果我們收到值后，想更新一些 UI 怎么辦？我們可以在閉包中執(zhí)行類(lèi)似 DispatchQueue.main.async { ... } 的操作，從主線程執(zhí)行 UI 更新。有一種更有效的方法可以使用 Combine 的 receive(on:):

let subscription = publisher
  .subscribe(on: queue)
  .receive(on: DispatchQueue.main)
  .sink { value in ...

即使計(jì)算工作正常并從后臺(tái)線程發(fā)出結(jié)果，我們現(xiàn)在也可以保證始終在主隊(duì)列上接收值。這是安全地執(zhí)行 UI 更新所需要的。

Scheduler 實(shí)現(xiàn)

Apple 提供了幾種 Scheduler 協(xié)議的具體實(shí)現(xiàn)：

ImmediateScheduler：一個(gè)簡(jiǎn)單的 Scheduler，它立即在當(dāng)前線程上執(zhí)行代碼，這是默認(rèn)的執(zhí)行上下文，除非使用 subscribe(on:)、receive(on:) 或任何其他將 Scheduler 作為參數(shù)的 Operator 進(jìn)行修改。
RunLoop：綁定到 Foundation 的 Thread 對(duì)象。
DispatchQueue：可以是串行的或并發(fā)的。
OperationQueue：規(guī)范工作項(xiàng)執(zhí)行的隊(duì)列。

這里省略了 TestScheduler，是一個(gè)虛擬的、模擬的 Scheduler，它是任何響應(yīng)式編程框架測(cè)試時(shí)不可或缺的一部分。

`ImmediateScheduler`

在 Playground 中新增代碼：

example("ImmediateScheduler") { 
    let source = Timer
      .publish(every: 1.0, on: .main, in: .common)
      .autoconnect()
      .scan(0) { counter, _ in counter + 1 }
    let publisher = source
        .receive(on: ImmediateScheduler.shared)
        .eraseToAnyPublisher()
    publisher.sink(receiveValue: { _ in
        print(Thread.current)
    })
    .store(in: &amp;subscriptions)
}

運(yùn)行 Playground，我們會(huì)看到 Publisher 發(fā)出的每個(gè)值，都是在 MainThread 上：

--- ImmediateScheduler ---
<_NSMainThread: 0x129617390>{number = 1, name = main}
<_NSMainThread: 0x129617390>{number = 1, name = main}
<_NSMainThread: 0x129617390>{number = 1, name = main}
<_NSMainThread: 0x129617390>{number = 1, name = main}
<_NSMainThread: 0x129617390>{number = 1, name = main}

當(dāng)前線程是主線程， ImmediateScheduler 立即在當(dāng)前線程上調(diào)度。當(dāng)我們?cè)?.receive(on: ImmediateScheduler.shared) 前添加一行：

.receive(on: DispatchQueue.global())

執(zhí)行 Playground，我們將在不同的線程收到值：

--- ImmediateScheduler ---
<NSThread: 0x12e7286c0>{number = 4, name = (null)}
<NSThread: 0x12e7286c0>{number = 4, name = (null)}
<NSThread: 0x11f005310>{number = 2, name = (null)}
<NSThread: 0x11f005310>{number = 2, name = (null)}
<NSThread: 0x12e7286c0>{number = 4, name = (null)}

ImmediateScheduler options 由于大多數(shù) Operator 在其參數(shù)中接受 Scheduler，我們還可以找到一個(gè)接受 SchedulerOptions 值的參數(shù)。在 ImmediateScheduler 的情況下，此類(lèi)型被定義為 Never，因此在使用 ImmediateScheduler 時(shí)，我們永遠(yuǎn)不應(yīng)該為 Operator 的 options 參數(shù)傳遞值。

ImmediateScheduler 的陷阱 關(guān)于 ImmediateScheduler 的一件事是它是即時(shí)的。我們無(wú)法使用 Scheduler 協(xié)議的任何 schedule(after:) 變體，因?yàn)槲覀冃枰付ǖ?SchedulerTimeType 沒(méi)有初始化方法，對(duì)于 ImmediateScheduler 無(wú)意義。

RunLoop scheduler

RunLoop 早于 DispatchQueue，它是一種在線程級(jí)別管理輸入源的方法。主線程有一個(gè)關(guān)聯(lián)的 RunLoop，我們還可以通過(guò)從當(dāng)前線程調(diào)用 RunLoop.current 為任何線程獲取一個(gè) RunLoop。

在 Playground 中添加此代碼：

example("RunLoop") { 
    let source = Timer
      .publish(every: 1.0, on: .main, in: .common)
      .autoconnect()
      .scan(0) { counter, _ in counter + 1 }
    let publisher = source
        .receive(on: DispatchQueue.global())
        .handleEvents(receiveOutput: { _ in
            print("DispatchQueue.global: \(Thread.current)")
        })
        .receive(on: RunLoop.current)
        .handleEvents(receiveOutput: { _ in
            print("RunLoop.current: \(Thread.current)")
        })
        .eraseToAnyPublisher()
    publisher.sink(receiveValue: { _ in
    })
    .store(in: &amp;subscriptions)
}

當(dāng)前 RunLoop.current 就是主線程的 RunLoop。執(zhí)行 Playground：

--- RunLoop ---
DispatchQueue.global: &lt;NSThread: 0x12a71cd20&gt;{number = 3, name = (null)}
RunLoop.current: &lt;_NSMainThread: 0x12a705760&gt;{number = 1, name = main}
DispatchQueue.global: &lt;NSThread: 0x12a71cd20&gt;{number = 3, name = (null)}
RunLoop.current: &lt;_NSMainThread: 0x12a705760&gt;{number = 1, name = main}
DispatchQueue.global: &lt;NSThread: 0x12a71cd20&gt;{number = 3, name = (null)}
RunLoop.current: &lt;_NSMainThread: 0x12a705760&gt;{number = 1, name = main}

每發(fā)出一個(gè)值，都通過(guò)一個(gè)全局并發(fā)隊(duì)列的線程，然后在主線程上繼續(xù)。

RunLoop Options 與 ImmediateScheduler 一樣，RunLoop 不提供 SchedulerOptions 參數(shù)。

RunLoop 陷阱 RunLoop 的使用應(yīng)僅限于主線程的 RunLoop，以及我們?cè)谛枰獣r(shí)控制的 Foundation 線程中可用的 RunLoop。要避免的一個(gè)是在 DispatchQueue 上執(zhí)行的代碼中使用 RunLoop.current。這是因?yàn)?DispatchQueue 線程可能是短暫的，這使得它們幾乎不可能依賴(lài) RunLoop。

DispatchQueue Scheduler

DispatchQueue 符合 Scheduler 協(xié)議，并且完全可用于所有將 Scheduler 作為參數(shù)的 Operator。Dispatch 框架是 Foundation 的一個(gè)強(qiáng)大組件，它允許我們通過(guò)向系統(tǒng)管理的調(diào)度隊(duì)列提交工作來(lái)在多核硬件上同時(shí)執(zhí)行代碼。DispatchQueue 可以是串行的（默認(rèn)）或并發(fā)的。串行隊(duì)列按順序執(zhí)行你提供給它的所有工作項(xiàng)。并發(fā)隊(duì)列將并行啟動(dòng)多個(gè)工作項(xiàng)，以最大限度地提高 CPU 使用率：

串行隊(duì)列通常用于保證某些操作不重疊。因此，如果所有操作都發(fā)生在同一個(gè)隊(duì)列中，他們可以使用共享資源而無(wú)需加鎖。
并發(fā)隊(duì)列將同時(shí)執(zhí)行盡可能多的操作。因此，它更適合純計(jì)算。

我們一直使用的最熟悉的隊(duì)列是 DispatchQueue.main。它直接映射到主線程，在這個(gè)隊(duì)列上執(zhí)行的所有操作都可以自由地更新用戶(hù)界面。當(dāng)然，UI 更新只能在主線程進(jìn)行。所有其他隊(duì)列，無(wú)論是串行的還是并發(fā)的，都在系統(tǒng)管理的線程池中執(zhí)行它們的代碼。這意味著我們永遠(yuǎn)不應(yīng)該對(duì)隊(duì)列中運(yùn)行的代碼中的當(dāng)前線程做出任何假設(shè)。尤其不應(yīng)使用 RunLoop.current 來(lái)安排工作，因?yàn)?DispatchQueue 管理其線程的方式有不同。

所有調(diào)度隊(duì)列共享同一個(gè)線程池，執(zhí)行的串行隊(duì)列將使用該池中的任何可用線程。一個(gè)直接的結(jié)果是，來(lái)自同一隊(duì)列的兩個(gè)連續(xù)工作項(xiàng)可能使用不同的線程，但仍可以按順序執(zhí)行。這是一個(gè)重要的區(qū)別：當(dāng)使用 subscribe(on:)、receive(on:) 或任何其他有 Scheduler 參數(shù)的 Operator 時(shí)，我們永遠(yuǎn)不應(yīng)假設(shè)線程每次都是相同的。

在 Playground 中添加代碼：

example("DispatchQueue") { 
    let source = PassthroughSubject<Void, Never>()
    let sourceQueue = DispatchQueue.main
    let subscription = sourceQueue.schedule(after: sourceQueue.now,
                                            interval: .seconds(1)) {
        source.send()
    }
    .store(in: &subscriptions)    let serialQueue = DispatchQueue(label: "Serial queue")
    source
        .handleEvents(receiveOutput: { _ in
            print("\(Thread.current)")
        })
        .receive(on: serialQueue)
        .handleEvents(receiveOutput: { _ in
            print("\(Thread.current)")
        })
        .sink(receiveValue: { _ in
        })
        .store(in: &subscriptions)
}

Timer 在主隊(duì)列 sourceQueue 上觸發(fā)并通過(guò) source 發(fā)送 Void 值。接著在串行隊(duì)列 serialQueue 上接收值：

--- DispatchQueue ---
<_NSMainThread: 0x126f0a250>{number = 1, name = main}
<NSThread: 0x128025cd0>{number = 2, name = (null)}
<_NSMainThread: 0x126f0a250>{number = 1, name = main}
<NSThread: 0x1178243e0>{number = 6, name = (null)}
<_NSMainThread: 0x126f0a250>{number = 1, name = main}
<NSThread: 0x117904d90>{number = 5, name = (null)}
<_NSMainThread: 0x126f0a250>{number = 1, name = main}
<NSThread: 0x1178243e0>{number = 6, name = (null)}
<_NSMainThread: 0x126f0a250>{number = 1, name = main}
<NSThread: 0x1178243e0>{number = 6, name = (null)}

將 sourceQueue 也改為 DispatchQueue(label: "Serial queue")，也將在全局并發(fā)隊(duì)列上發(fā)出值：

--- DispatchQueue ---
<NSThread: 0x137e275b0>{number = 6, name = (null)}
<NSThread: 0x130905310>{number = 2, name = (null)}
<NSThread: 0x130905310>{number = 2, name = (null)}
<NSThread: 0x130905310>{number = 2, name = (null)}
<NSThread: 0x127e0f400>{number = 4, name = (null)}
<NSThread: 0x137e275b0>{number = 6, name = (null)}

DispatchQueue Options DispatchQueue 是唯一提供一組 Options 的 Scheduler，當(dāng) Operator 需要 SchedulerOptions 參數(shù)時(shí)，我們可以傳遞這些 Options。主要圍繞 QoS（服務(wù)質(zhì)量）值，獨(dú)立于 DispatchQueue 上已設(shè)置的值。例如：

.receive(
  on: serialQueue,
  options: DispatchQueue.SchedulerOptions(qos: .userInteractive)
)

我們將 DispatchQueue.SchedulerOptions 的實(shí)例傳遞.userInteractive。在實(shí)際開(kāi)發(fā)中使用這些 Options 有助于操作系統(tǒng)決定在同時(shí)有許多隊(duì)列忙碌的情況下首先安排哪個(gè)任務(wù)。

OperationQueue Scheduler

由于 OperationQueue 在內(nèi)部使用 Dispatch，因此在表面上幾乎沒(méi)有區(qū)別：

example("OperationQueue") { 
    let queue = OperationQueue()
    let subscription = (1...10).publisher
        .receive(on: queue)
        .print()
        .sink { value in
            print("Received \(value)")
        }
        .store(in: &amp;subscriptions)
}

創(chuàng)建一個(gè)簡(jiǎn)單的 Publisher 發(fā)出 1 到 10 之間的數(shù)字，然后打印該值，執(zhí)行 Playground：

--- OperationQueue ---
receive subscription: (ReceiveOn)
request unlimited
receive value: (1)
Received 1
receive value: (8)
Received 8
receive value: (9)
Received 9
receive value: (6)
Received 6
receive value: (3)
Received 3
receive value: (5)
Received 5
receive finished
receive value: (10)
receive value: (4)
receive value: (7)
receive value: (2)

按順序發(fā)出但無(wú)序到達(dá)！我們可以更改打印行以顯示當(dāng)前線程：

print("Received \(value) on thread \(Thread.current)")

再次執(zhí)行 Playground：

--- OperationQueue ---
receive subscription: (ReceiveOn)
request unlimited
receive value: (4)
Received 4 on thread <NSThread: 0x14d720980>{number = 2, name = (null)}
receive value: (10)
Received 10 on thread <NSThread: 0x14d720980>{number = 2, name = (null)}
receive value: (3)
Received 3 on thread <NSThread: 0x14e833620>{number = 6, name = (null)}
receive value: (5)
Received 5 on thread <NSThread: 0x14e80dfd0>{number = 4, name = (null)}
receive value: (1)
Received 1 on thread <NSThread: 0x14d70d840>{number = 5, name = (null)}
receive finished
receive value: (2)
receive value: (9)
receive value: (8)
receive value: (6)

每個(gè)值都是在不同的線程上接收的！如果我們查看有關(guān) OperationQueue 的文檔，有一條關(guān)于線程的說(shuō)明，OperationQueue 使用 Dispatch 框架（因此是 DispatchQueue）來(lái)執(zhí)行操作。這意味著它不保證它會(huì)為每個(gè)交付的值使用相同的底層線程。

此外，每個(gè) OperationQueue 中都有一個(gè)參數(shù)可以解釋一切：它是 maxConcurrentOperationCount。它默認(rèn)為系統(tǒng)定義的數(shù)字，允許操作隊(duì)列同時(shí)執(zhí)行大量操作。由于 Publisher 幾乎在同一時(shí)間發(fā)出所有值，它們被 Dispatch 的并發(fā)隊(duì)列分派到多個(gè)線程。

對(duì)代碼進(jìn)行一些修改：

queue.maxConcurrentOperationCount = 1

再次執(zhí)行 Playground：

--- OperationQueue ---
receive subscription: (ReceiveOn)
request unlimited
receive value: (1)
Received 1 on thread <NSThread: 0x117609390>{number = 4, name = (null)}
receive value: (2)
Received 2 on thread <NSThread: 0x117609390>{number = 4, name = (null)}
receive value: (3)
Received 3 on thread <NSThread: 0x117609390>{number = 4, name = (null)}
receive value: (4)
Received 4 on thread <NSThread: 0x117609390>{number = 4, name = (null)}
receive value: (5)
Received 5 on thread <NSThread: 0x117627160>{number = 6, name = (null)}
receive value: (6)
Received 6 on thread <NSThread: 0x117627160>{number = 6, name = (null)}
receive value: (7)
Received 7 on thread <NSThread: 0x117627160>{number = 6, name = (null)}
receive value: (8)
Received 8 on thread <NSThread: 0x117627160>{number = 6, name = (null)}
receive value: (9)
Received 9 on thread <NSThread: 0x117627160>{number = 6, name = (null)}
receive value: (10)
Received 10 on thread <NSThread: 0x117627160>{number = 6, name = (null)}
receive finished

這一次，我們將獲得真正的順序執(zhí)行——將 maxConcurrentOperationCount 設(shè)置為 1 相當(dāng)于使用串行隊(duì)列。

OperationQueue Options OperationQueue 沒(méi)有可用的 SchedulerOptions。它實(shí)際上是 RunLoop.SchedulerOptions 類(lèi)型，本身沒(méi)有提供任何 Options。

OperationQueue 陷阱 我們剛剛看到 OperationQueue 默認(rèn)并發(fā)執(zhí)行操作，我們需要非常清楚這一點(diǎn)，因?yàn)樗赡軙?huì)給我們帶來(lái)麻煩。當(dāng)我們的 Publisher 發(fā)出值時(shí)都有大量工作要執(zhí)行時(shí)，它可能是一個(gè)很好的工具。我們可以通過(guò)調(diào)整 maxConcurrentOperationCount 參數(shù)來(lái)控制負(fù)載。

內(nèi)容參考

Combine | Apple Developer Documentation；
來(lái)自 Kodeco 的書(shū)籍《Combine: Asynchronous Programming with Swift》；
對(duì)上述 Kodeco 書(shū)籍的漢語(yǔ)自譯版《Combine: Asynchronous Programming with Swift》整理與補(bǔ)充。

以上就是Combine中錯(cuò)誤處理和Scheduler使用詳解的詳細(xì)內(nèi)容，更多關(guān)于Combine錯(cuò)誤處理Scheduler的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: