引言

在之前的文章中，我們了解了 Combine 的基礎(chǔ)知識(shí)：了解了 Publisher、Subscriber 和 Subscription 如何工作以及這些部分之間的相互關(guān)系，以及如何使用 Operator 來(lái)操作 Publisher 及處理其事件。

本文將 Combine 用于用于特定用例，更貼近實(shí)際的應(yīng)用開(kāi)發(fā)。我們將了解如何利用 Combine 進(jìn)行網(wǎng)絡(luò)任務(wù)、如何調(diào)試 Combine Publisher、如何使用 Timer、觀察對(duì)象，以及了解 Combine 中的資源管理。

網(wǎng)絡(luò)

Combine 提供了 API 來(lái)幫助開(kāi)發(fā)者以聲明方式執(zhí)行常見(jiàn)任務(wù)。這些 API 圍繞兩個(gè)關(guān)鍵功能：

使用 URLSession 執(zhí)行網(wǎng)絡(luò)請(qǐng)求。

使用 Codable 協(xié)議對(duì) JSON 數(shù)據(jù)進(jìn)行編碼和解碼。

URLSession Extension

URLSession 是 Apple 平臺(tái)下執(zhí)行網(wǎng)絡(luò)相關(guān)任務(wù)的標(biāo)準(zhǔn)方式，可以幫助我們完成多種操作。例如：

• 用于檢索 URL 的內(nèi)容的數(shù)據(jù)傳輸任務(wù)；
• 用于獲取 URL 的內(nèi)容的數(shù)據(jù)下載任務(wù)；
• 用于向 URL 上傳數(shù)據(jù)或文件的上傳任務(wù)；
• 在兩方之間傳輸數(shù)據(jù)的流式傳輸任務(wù)；
• 連接到 Websocket 的 Websocket 任務(wù)。

其中，只有數(shù)據(jù)傳輸任務(wù)公開(kāi)了一個(gè) Combine Publisher。 Combine 使用具有兩個(gè)變體的單個(gè) API 處理這些任務(wù)。入?yún)?shù)為 URLRequest 或 URL：

func dataTaskPublisher(for url: URL) -> URLSession.DataTaskPublisher
func dataTaskPublisher(for request: URLRequest) -> URLSession.DataTaskPublisher

下面看看如何使用這個(gè) API：

import Combine
import Foundation
import PlaygroundSupport
PlaygroundPage.current.needsIndefiniteExecution = true
func example(_ desc: String, _ action:() -> Void) {
    print("--- \(desc) ---")
    action()
}
var subscriptions = Set<AnyCancellable>()
example("URLSession") { 
    guard let url = URL(string: "https://random-data-api.com/api/v2/appliances") else { 
        return 
    }
    URLSession.shared
        .dataTaskPublisher(for: url)
        .sink(receiveCompletion: { completion in
            if case .failure(let err) = completion {
                print("Retrieving data failed with error \(err)")
            }
        }, receiveValue: { data, response in
            print("Retrieved data of size \(data.count), response = \(response)")
        })
        .store(in: &subscriptions)
}

在一些基礎(chǔ)代碼后，進(jìn)入我們的 example 函數(shù)。我們使用 URL 作為參數(shù)的 dataTaskPublisher(for:) 。確保我們處理了錯(cuò)誤。請(qǐng)求結(jié)果是包含 Data 和 URLResponse 的元組。Combine 在 URLSession.dataTask 上提供了 Publisher 而不是閉包。最后保留 Subscription，否請(qǐng)求它會(huì)立即被取消，并且請(qǐng)求永遠(yuǎn)不會(huì)執(zhí)行。

Codable

Codable 協(xié)議是我們絕對(duì)應(yīng)該了解的 Swift 的編碼和解碼機(jī)制。Foundation 通過(guò) JSONEncoder 和 JSONDecoder 對(duì) JSON 進(jìn)行編碼和解碼。 我們也可以使用 PropertyListEncoder 和 PropertyListDecoder，但這些在網(wǎng)絡(luò)請(qǐng)求的上下文中用處不大。

在前面的示例中，我們獲取了一些 JSON。 我們可以使用 JSONDecoder 對(duì)其進(jìn)行解碼：

example("URLSession") { 
    guard let url = URL(string: "https://random-data-api.com/api/v2/appliances") else { 
        return 
    }
    URLSession.shared
        .dataTaskPublisher(for: url)
        .tryMap({ data, response in
            try JSONDecoder().decode([String:String].self, from: data)
        })
        .sink(receiveCompletion: { completion in
            if case .failure(let err) = completion {
                print("Retrieving data failed with error \(err)")
            }
        }, receiveValue: { data in
            print("Retrieved data: \(data)")
        })
        .store(in: &subscriptions)
}

我們?cè)?tryMap Operator 中解碼 JSON，該方法有效，但 Combine 提供了一個(gè)在該場(chǎng)景下更合適的 Operator 來(lái)幫助減少代碼：decode(type:decoder:)：

    URLSession.shared
        .dataTaskPublisher(for: url)
        .map(\.data)
        .decode(type: [String:String].self, decoder: JSONDecoder())
        .sink(receiveCompletion: { completion in
            if case .failure(let err) = completion {
                print("Retrieving data failed with error \(err)")
            }
        }, receiveValue: { data in
            print("Retrieved data: \(data)")
        })
        .store(in: &subscriptions)

但由于 dataTaskPublisher(for:) 發(fā)出一個(gè)元組，我們不能直接使用 decode(type:decoder:)， 需要使用 map(_:) 只處理部分?jǐn)?shù)據(jù)。其他的優(yōu)點(diǎn)包括我們只在設(shè)置 Publisher 時(shí)實(shí)例化 JSONDecoder 一次，而不是每次在 tryMap(_:) 閉包中創(chuàng)建它。

向多個(gè) Subscriber 發(fā)布網(wǎng)絡(luò)數(shù)據(jù)

每次訂閱 Publisher 時(shí)，它都會(huì)開(kāi)始工作。在網(wǎng)絡(luò)請(qǐng)求的情況下，如果多個(gè) Subscriber 需要結(jié)果，則多次發(fā)送相同的請(qǐng)求。

Combine 沒(méi)有像其他框架那樣容易實(shí)現(xiàn)這一點(diǎn)的 Operator。 我們可以使用 share() Operator，但這需要在結(jié)果返回之前設(shè)置所有的 Subscription。

還有另一種解決方案：使用 multicast() Operator，它返回一個(gè) ConnectablePublisher，該 Publisher 為每個(gè) Subscriber 創(chuàng)建一個(gè)單獨(dú)的 Subject。 它允許我們多次訂閱 Subject，然后在我們準(zhǔn)備好時(shí)，調(diào)用 Publisher 的 connect() 方法：

example("connect") { 
    guard let url = URL(string: "https://random-data-api.com/api/v2/appliances") else { 
        return 
    }
    let publisher = URLSession.shared
        .dataTaskPublisher(for: url)
        .map(\.data)
        .multicast { PassthroughSubject<Data, URLError>() }
    let subscription1 = publisher
        .sink(receiveCompletion: { completion in
            if case .failure(let err) = completion {
                print("Sink1 Retrieving data failed with error \(err)")
            }
        }, receiveValue: { object in
            print("Sink1 Retrieved object \(object)")
        })
        .store(in: &subscriptions)
    let subscription2 = publisher
        .sink(receiveCompletion: { completion in
            if case .failure(let err) = completion {
                print("Sink2 Retrieving data failed with error \(err)")
            }
        }, receiveValue: { object in
            print("Sink2 Retrieved object \(object)")
        })
        .store(in: &subscriptions)
    let subscription = publisher.connect()
}

在上述代碼中，創(chuàng)建 DataTaskPublisher 后，map data，然后使用 multicast。傳遞給 multicast 的閉包必須返回適當(dāng)類型的 Subject。 我們會(huì)在后文中了解有關(guān) multicast 的更多信息。首次訂閱 Publisher，由于它是一個(gè) ConnectablePublisher，它不會(huì)立即開(kāi)始工作。準(zhǔn)備好后使用 publisher.connect() 它將開(kāi)始工作并向所有 Subscriber 推送值。

通過(guò)上述代碼，我們可以一次請(qǐng)求并與兩個(gè) Subscriber 共享結(jié)果。這個(gè)過(guò)程仍然有點(diǎn)復(fù)雜，因?yàn)?Combine 不像其他響應(yīng)式框架那樣為這種場(chǎng)景提供 Operator。后續(xù)文章我們將探索如何設(shè)計(jì)一個(gè)更好的解決方案。

調(diào)試

理解異步代碼中的事件流一直是一個(gè)挑戰(zhàn)。在 Combine 的上下文中尤其如此，因?yàn)?Publisher 中的 Operator 鏈可能不會(huì)立即發(fā)出事件。 例如，像 throttle(for:scheduler:latest:) 這樣的 Operator 不會(huì)發(fā)出它們接收到的所有事件，所以我們需要了解發(fā)生了什么。 Combine 提供了一些 Operator 來(lái)幫助我們進(jìn)行調(diào)試。

打印事件

print(_:to:) Operator 是我們?cè)诓淮_定是否有任何內(nèi)容通過(guò)時(shí)，應(yīng)該使用的第一個(gè) Operator。它返回一個(gè)PassthroughPublisher ，可以打印很多關(guān)于正在發(fā)生的事情的信息：

即使是這樣的簡(jiǎn)單案例：

let subscription = (1...3).publisher
  .print("publisher")
  .sink { _ in }

輸出非常詳細(xì)：

publisher: receive subscription: (1...3)
publisher: request unlimited
publisher: receive value: (1)
publisher: receive value: (2)
publisher: receive value: (3)
publisher: receive finished

我們會(huì)看到 print(_:to:) Operator 顯示了很多信息：

• 在收到訂閱時(shí)打印并顯示其上游 Publisher 的描述；
• 打印 Subscriber 的 demand request，以便我們查看請(qǐng)求的值的數(shù)量。
• 打印上游 Publisher 發(fā)出的每個(gè)值。
• 最后，打印完成事件。

print 有一個(gè)額外的參數(shù)接受一個(gè) TextOutputStream 對(duì)象。 我們可以使用它來(lái)重定向字符串以打印到自定義的記錄器中。我們還可以在日志中添加額外信息，例如當(dāng)前日期和時(shí)間等。

我們可以創(chuàng)建一個(gè)簡(jiǎn)單的記錄器來(lái)顯示每個(gè)字符串之間的時(shí)間間隔，以便了解發(fā)布者發(fā)出值的速度：

example("print") { 
    class TimeLogger: TextOutputStream {
      private var previous = Date()
      private let formatter = NumberFormatter()
      init() {
        formatter.maximumFractionDigits = 5
        formatter.minimumFractionDigits = 5
      }
      func write(_ string: String) {
        let trimmed = string.trimmingCharacters(in: .whitespacesAndNewlines)
        guard !trimmed.isEmpty else { return }
        let now = Date()
        print("+\(formatter.string(for: now.timeIntervalSince(previous))!)s: \(string)")
        previous = now
      }
    }
    let subscription = (1...3).publisher
      .print("publisher", to: TimeLogger())
      .sink { _ in }
}

結(jié)果顯示每條打印行之間的時(shí)間：

--- print ---
+0.00064s: publisher: receive subscription: (1...3)
+0.00145s: publisher: request unlimited
+0.00035s: publisher: receive value: (1)
+0.00026s: publisher: receive value: (2)
+0.00028s: publisher: receive value: (3)
+0.00026s: publisher: receive finished

執(zhí)行副作用

除了打印信息外，對(duì)特定事件執(zhí)行操作通常很有用，我們將此稱為執(zhí)行副作用：額外的操作不會(huì)直接影響下游的其他發(fā) Publisher，但會(huì)產(chǎn)生類似于修改外部變量的效果。

handleEvents(receiveSubscription:receiveOutput:receiveCompletion:receiveCancel:receiveRequest:) 讓我們可以攔截 Publisher 生命周期中的所有事件，然后在每個(gè)步驟中進(jìn)行額外的操作。

考慮這段代碼：

example("handleEvents", {
    guard let url = URL(string: "https://random-data-api.com/api/v2/appliances") else { 
        return 
    }
    URLSession.shared
        .dataTaskPublisher(for: url)
        .map(\.data)
        .decode(type: [String:String].self, decoder: JSONDecoder())
        .sink(receiveCompletion: { completion in
            print("\(completion)")
        }, receiveValue: { data in
            print("\(data)")
        })
})

我們運(yùn)行它，從來(lái)沒(méi)有看到任何打印。我們使用 handleEvents 來(lái)跟蹤正在發(fā)生的事情。 你可以在 publisher 和 sink 之間插入此 Operator：

.handleEvents(receiveSubscription: { _ in
  print("Network request will start")
}, receiveOutput: { _ in
  print("Network request data received")
}, receiveCancel: {
  print("Network request cancelled")
})

再次運(yùn)行代碼，這次我們會(huì)看到一些調(diào)試輸出：

--- handleEvents ---
Network request will start
Network request cancelled

我們忘記保留 AnyCancellable。 因此 Subscription 開(kāi)始但立即被取消。

使用 Debugger Operator

Debugger 操作符是我們?cè)谌f(wàn)不得已的時(shí)候確實(shí)需要使用的 Operator。

第一個(gè)簡(jiǎn)單的 Operator 是 breakpointOnError()。 顧名思義，當(dāng)我們使用此 Operator 時(shí)，如果任何上游 Publisher 發(fā)出錯(cuò)誤，Xcode 將在調(diào)試器中中斷。

一個(gè)更完整的變體是 breakpoint（receiveSubscription:receiveOutput:receiveCompletion:)。 它允許你攔截所有事件并根據(jù)具體情況決定是否要暫停。

例如，只有當(dāng)某些值通過(guò)時(shí)才中斷：

.breakpoint(receiveOutput: { value in
  return value > 0 && value < 5
})

假設(shè)上游 Publisher 發(fā)出整數(shù)值，但值 1 到 5 永遠(yuǎn)不會(huì)被發(fā)出，我們可以將斷點(diǎn)配置為僅在這種情況下中斷。

Timer

Timer 在編碼時(shí)經(jīng)常用到，除了異步執(zhí)行代碼之外，可能還需要控制任務(wù)應(yīng)該重復(fù)的時(shí)間和頻率。

在 Dispatch 框架可用之前，開(kāi)發(fā)人員依靠 RunLoop 來(lái)異步執(zhí)行任務(wù)并實(shí)現(xiàn)并發(fā)。以上所有方法都能夠創(chuàng)建 Timer，但在 Combine 中并非所有 Timer 都相同。

使用 RunLoop

線程可以擁有自己的 RunLoop，只需從當(dāng)前線程調(diào)用 RunLoop.current。請(qǐng)注意，除非我們了解 RunLoop 是如何運(yùn)行的——特別是真的需要一個(gè) RunLoop —— 否則最好只使用主線程的 RunLoop。

注意：Apple 文檔中的一個(gè)重要說(shuō)明和警告是 RunLoop 類不是線程安全的。 我們應(yīng)該只為當(dāng)前線程的 RunLoop 用 RunLoop 方法。

RunLoop 實(shí)現(xiàn)了我們將后續(xù)文章中了解的 Scheduler 協(xié)議。它定義了幾種相對(duì)低級(jí)別的方法，并且是唯一一種可以讓你創(chuàng)建可取消 Timer 的方法：

example("Timer RunLoop") { 
    let runLoop = RunLoop.main
    let subscription = runLoop.schedule(
      after: runLoop.now,
      interval: .seconds(1),
      tolerance: .milliseconds(100)
    ) {
      print("Timer fired")
    }
    .store(in: &subscriptions)
}

此 Timer 不傳遞任何值，也不創(chuàng)建 Publisher。 它從 after: 參數(shù)中指定的 date 開(kāi)始，具有指定的間隔 interval 和容差 tolerance。 它與 Combine 相關(guān)的唯一用處是它返回的 Cancelable 可讓我們?cè)谝欢螘r(shí)間后將其停止：

example("Timer RunLoop") { 
    let runLoop = RunLoop.main
    let subscription = runLoop.schedule(
      after: runLoop.now,
      interval: .seconds(1),
      tolerance: .milliseconds(100)
    ) {
      print("Timer fired")
    }
    runLoop.schedule(after: .init(Date(timeIntervalSinceNow: 3.0))) {
      subscription.cancel()
    }
}

考慮到所有因素，RunLoop 并不是創(chuàng)建 Timer 的最佳方式，使用 Timer 類會(huì)更好。

使用 Timer 類

Timer 是 Mac OS X 中可用的最古老的計(jì)時(shí)器。 由于它的委托模式和與 RunLoop 的緊密關(guān)系，它一直很難使用。 Combine 帶來(lái)了一個(gè)現(xiàn)代變體，我們可以直接用作 Publisher：

let publisher = Timer.publish(every: 1.0, on: .main, in: .common)

on 和 in 兩個(gè)參數(shù)確定：

• Timer 附加到哪個(gè) RunLoop，這里是主線程的 RunLoop。
• Timer 在哪個(gè) RunLoop 模式下運(yùn)行，這里是默認(rèn)的 RunLoop 模式。

RunLoop 是 macOS 中異步事件處理的基本機(jī)制，但它們的 API 有點(diǎn)繁瑣。我們可以通過(guò)調(diào)用 RunLoop.current 為我們自己創(chuàng)建或從 Foundation 獲取的任何線程獲取 RunLoop，因此我們也可以編寫(xiě)以下代碼：

let publisher = Timer.publish(every: 1.0, on: .current, in: .common)

注意：在 DispatchQueue.main 以外的 Dispatch 隊(duì)列上運(yùn)行此代碼可能會(huì)導(dǎo)致不可預(yù)知的結(jié)果。 Dispatch 框架不使用 RunLoop 來(lái)管理其線程。 由于 RunLoop 需要調(diào)用其方法來(lái)處理事件，因此我們永遠(yuǎn)不會(huì)看到 Timer 在除主隊(duì)列之外的任何隊(duì)列上觸發(fā)。 為 Timer 設(shè)置為 RunLoop.main 是最簡(jiǎn)單安全的選擇。

計(jì)時(shí)器返回的發(fā)布者是 ConnectablePublisher，在我們顯式調(diào)用它的 connect() 方法之前，它不會(huì)在 Subscription 時(shí)開(kāi)始觸發(fā)。我們還可以使用 autoconnect() Operator，它會(huì)在第一個(gè) Subscriber 訂閱時(shí)自動(dòng)連接。

因此，創(chuàng)建將在訂閱時(shí)啟動(dòng) Timer 的 Publisher 的最佳方法是編寫(xiě)：

let publisher = Timer
  .publish(every: 1.0, on: .main, in: .common)
  .autoconnect()

Timer Publisher 發(fā)出當(dāng)前日期，其 Publisher.Output 類型為 Date。 我們可以使用 scan 制作一個(gè)發(fā)出遞增值的計(jì)時(shí)器：

example("Timer Timer") { 
    let subscription = Timer
        .publish(every: 1.0, on: .main, in: .common)
        .autoconnect()
        .scan(0) { counter, _ in counter + 1 }
        .sink { counter in
            print("Counter is \(counter)")
        }
        .store(in: &subscriptions)
}

還有一個(gè)我們?cè)谶@里沒(méi)有使用的 Timer.publish() 參數(shù)：容差(Tolerance)。 它以 TimeInterval 形式指定可接受的偏差。但請(qǐng)注意，使用低于 RunLoop 的 minimumTolerance 值的值可能會(huì)產(chǎn)生不符合預(yù)期的結(jié)果。

使用 DispatchQueue

我們可以使用 DispatchQueue 來(lái)生成 Timer。雖然 Dispatch 框架有一個(gè) DispatchTimerSource，但 Combine 沒(méi)有為其提供 Timer 接口。 相反，我們將使用另一種方法生成 Timer 事件：

example("Timer DispatchQueue") { 
    let queue = DispatchQueue.main
    let source = PassthroughSubject<Int, Never>()
    var counter = 0
    let cancellable = queue.schedule(
      after: queue.now,
      interval: .seconds(1)
    ) {
      source.send(counter)
      counter += 1
    }
    .store(in: &subscriptions)
    let subscription = source.sink {
      print("Timer emitted \($0)")
    }
    .store(in: &subscriptions)
}

這代碼并不漂亮。我們創(chuàng)建一個(gè) subject source，我們將向其發(fā)送 counter 值。每次計(jì)時(shí)觸發(fā)時(shí)，counter 都會(huì)增加它。每秒在所選隊(duì)列上安排一個(gè)重復(fù)操作，這將立即開(kāi)始。訂閱 source 獲取 counter 值。

KVO

處理變化是 Combine 的核心。Publisher 讓我們訂閱它們以處理異步事件。我們了解了 assign(to:on:)，它使我們能夠在每次 Publisher 發(fā)出新值時(shí)更新對(duì)象屬性的值。

此外，Combine 還提供了觀察單個(gè)變量變化的機(jī)制：

• 它為符合 KVO（Key-Value Observing）的對(duì)象的任何屬性提供 Publisher。
• ObservableObject 協(xié)議處理多個(gè)變量可能發(fā)生變化的情況。

publisher(for:options:)

KVO 一直是 Objective-C 的重要組成部分。 Foundation、UIKit 和 AppKit 類的大量屬性都符合 KVO 的要求。我們可以使用 KVO 來(lái)觀察它們的變化。

下面是一個(gè)對(duì) OperationQueue 的 operationCount 屬性 KVO 的示例：

let queue = OperationQueue()
let subscription = queue.publisher(for: \.operationCount)
  .sink {
    print("Outstanding operations in queue: \($0)")
  }

每次向隊(duì)列添加新 Operation 時(shí)，它的 operationCount 都會(huì)增加，并且我們的 sink 會(huì)收到新的計(jì)數(shù)值。當(dāng)隊(duì)列消耗了一個(gè) Operation 時(shí)，計(jì)數(shù)也相應(yīng)會(huì)減少，并且我們的 sink 會(huì)再次收到更新的計(jì)數(shù)值。

還有許多其他框架類公開(kāi)了符合 KVO 的屬性。只需將 publisher(for:) 與 KVO 兼容的屬性一起使用，我們將獲得一個(gè)能夠發(fā)出值變化的 Publisher。

自定義的 KVO 兼容屬性

我們還可以在自己的代碼中使用 Key-Value Observing，前提是：

• 對(duì)象是 NSObject 子類；
• 使用 @objc dynamic 標(biāo)記屬性。

完成此操作后，我們標(biāo)記的對(duì)象和屬性將與 KVO 兼容，并且可以使用 Combine。

注意：雖然 Swift 語(yǔ)言不直接支持 KVO，但將屬性標(biāo)記為 @objc dynamic 會(huì)強(qiáng)制編譯器生成觸發(fā) KVO 機(jī)制的方法，該機(jī)制依賴 NSObject 協(xié)議中的特定方法。

在 Playground 上嘗試一個(gè)例子：

example("KVO") { 
    class TestObject: NSObject {
      @objc dynamic var value: Int = 0
    }
    let obj = TestObject()
    let subscription = obj.publisher(for: \.value)
      .sink {
        print("value changes to \($0)")
      }
    obj.value = 100
    obj.value = 200
}

在上面的代碼中，我們創(chuàng)建了一個(gè) TestObject 類，繼承自 NSObject 這是 KVO 所必需的。將我們要使其可觀察的屬性標(biāo)記為 @objc dynamic。創(chuàng)建并訂閱 obj 的 value 屬性的 Publisher。更新屬性幾次：

--- KVO ---
value changes to 0
value changes to 100
value changes to 200

我們注意到在 TestObject 中我們使用的是 Swift 類型 Int，而作為 Objective-C 特性的 KVO 仍然有效？ KVO 可以與任何 Objective-C 類型以及任何橋接到 Objective-C 的 Swift 類型一起正常工作。這包括所有原生 Swift 類型以及數(shù)組和字典，只要它們的值都可以橋接到 Objective-C。

Observation options

publisher(for:options:) 的 options 參數(shù)是一個(gè)具有四個(gè)值的選項(xiàng)集：.initial、.prior、.old 和 .new。 默認(rèn)值為 [.initial]，這就是為什么我們會(huì)看到 Publisher 在發(fā)出任何更改之前發(fā)出初始值。以下是選項(xiàng)的細(xì)分：

.initial 發(fā)出初始值。

.prior 在發(fā)生更改時(shí)發(fā)出先前的值和新的值。

.old 和 .new 在此 Publisher 中未使用，它們都什么都不做（只是讓新值通過(guò)）。

如果我們不想要初始值，你可以簡(jiǎn)單地寫(xiě)：

obj.publisher(for: \.value, options: [])

如果我們指定 .prior，則每次發(fā)生更改時(shí)都會(huì)獲得兩個(gè)單獨(dú)的值。 修改 integerProperty 示例：

let subscription = obj.publisher(for: \.value, options: [.prior])

你現(xiàn)在將在 integerProperty 訂閱的調(diào)試控制臺(tái)中看到以下內(nèi)容：

--- KVO ---
value changes to 0
value changes to 100
value changes to 100
value changes to 200

該屬性首先從 0 更改為 100，因此我們獲得兩個(gè)值：0 和 100。然后，它從 100 更改為 200，因此我們?cè)俅潍@得兩個(gè)值：100 和 200。

ObservableObject

Combine 的 ObservableObject 協(xié)議不僅僅適用于派生自 NSObject 的對(duì)象，而且適用于 Swift 的對(duì)象。 它與 @Published 屬性包裝器合作，幫助我們使用編譯器生成的 objectWillChange Publisher 創(chuàng)建類。

它使我們免于編寫(xiě)大量重復(fù)代碼，并允許創(chuàng)建可以自我監(jiān)控的屬性，并在它們中的任何一個(gè)發(fā)生更改時(shí)通知的對(duì)象。

這是一個(gè)例子：

example("ObservableObject") { 
    class MonitorObject: ObservableObject {
      @Published var someProperty = false
      @Published var someOtherProperty = ""
    }
    let object = MonitorObject()
    let subscription = object.objectWillChange.sink {
      print("object will change")
    }
    object.someProperty = true
}

--- ObservableObject ---
object will change

ObservableObject 協(xié)議使編譯器自動(dòng)生成 objectWillChange 屬性。 它是一個(gè) ObservableObjectPublisher，它發(fā)出 Void 值并且永不失敗。

每次對(duì)象的 @Published 變量之一發(fā)生更改時(shí)，都會(huì)觸發(fā) objectWillChange。不幸的是，我們無(wú)法知道實(shí)際更改了哪個(gè)屬性。 這旨在與 SwiftUI 很好地配合使用，它可以合并事件以簡(jiǎn)化屏幕更新。

資源管理

在前面的內(nèi)容中，我們發(fā)現(xiàn)有時(shí)我們希望共享網(wǎng)絡(luò)請(qǐng)求、圖像處理和文件解碼等資源，而不是進(jìn)行重復(fù)的工作。換句話說(shuō)，我們希望在多個(gè)訂閱者之間共享單個(gè)資源的結(jié)果—— Publisher 發(fā)出的值，而不是復(fù)制該結(jié)果。

Combine 提供了兩個(gè)操作符來(lái)管理資源：share() Operator 和 multicast(_:) Operator。

share()

該 Operator 的目的是讓我們通過(guò)引用而不是通過(guò)值來(lái)獲取 Publisher。 Publisher 通常是結(jié)構(gòu)體：當(dāng)我們將 Publisher 傳遞給函數(shù)或?qū)⑵浯鎯?chǔ)在多個(gè)屬性中時(shí)，Swift 會(huì)多次復(fù)制它。當(dāng)我們訂閱每個(gè)副本時(shí)，Publisher 只能做一件事：開(kāi)始其工作并交付值。

share() Operator 返回 Publishers.Share 類的實(shí)例。通常，Publisher 被實(shí)現(xiàn)為結(jié)構(gòu)，但在 share() 的情況下， Operator 獲取對(duì) Publisher 的引用而不是使用值語(yǔ)義，這允許它共享底層 Publisher。

這個(gè)新 Publisher “共享”上游 Publisher。它將與第一個(gè)傳入的 Subscriber 一起訂閱一次上游 Publisher。然后它將從上游 Publisher 接收到的值轉(zhuǎn)發(fā)給這個(gè) Subscriber 以及所有在它之后訂閱的 Subscriber。

注意：新 Subscriber 只會(huì)收到上游 Publisher 在訂閱后發(fā)出的值。不涉及緩沖或重放。如果 Subscriber 在上游Publisher 完成后訂閱 share Publisher，則該新 Subscriber 只會(huì)收到完成事件。

假設(shè)我們正在執(zhí)行一個(gè)網(wǎng)絡(luò)請(qǐng)求，你希望多個(gè) Subscriber 無(wú)需多次請(qǐng)求即可接收結(jié)果：

example("share") { 
    let shared = URLSession.shared
      .dataTaskPublisher(for: URL(string: "https://random-data-api.com/api/v2/appliances")!)
      .map(\.data)
      .print("shared")
      .share()
    print("subscribing first")
    let subscription1 = shared.sink(
      receiveCompletion: { _ in },
      receiveValue: { print("subscription1 received: '\($0)'") }
    )
    .store(in: &subscriptions)
    print("subscribing second")
    let subscription2 = shared.sink(
      receiveCompletion: { _ in },
      receiveValue: { print("subscription2 received: '\($0)'") }
    )
    .store(in: &subscriptions)
}

第一個(gè) Subscriber 觸發(fā) share() 的上游 Publisher 工作(執(zhí)行網(wǎng)絡(luò)請(qǐng)求)。 第二個(gè) Subscriber 將簡(jiǎn)單地“連接”到它并與第一個(gè) Subscriber 同時(shí)接收值。

在 Playground 中運(yùn)行此代碼：

--- share ---
subscribing first
shared: receive subscription: (DataTaskPublisher)
shared: request unlimited
subscribing second
shared: receive value: (91 bytes)
subscription2 received: '91 bytes'
subscription1 received: '91 bytes'
shared: receive finished

我們可以看到，第一個(gè) Subscription 觸發(fā)對(duì) DataTaskPublisher 的訂閱。第二個(gè) Subscription 沒(méi)有任何改變：Publisher 繼續(xù)運(yùn)行，沒(méi)有第二個(gè)請(qǐng)求發(fā)出。當(dāng)請(qǐng)求完成時(shí)，Publisher 將結(jié)果數(shù)據(jù)發(fā)送給兩個(gè) Subscriber，然后完成。

要驗(yàn)證請(qǐng)求只發(fā)送一次，我們可以注釋掉 share()，輸出將類似于以下內(nèi)容：

--- share ---
subscribing first
shared: receive subscription: (DataTaskPublisher)
shared: request unlimited
subscribing second
shared: receive subscription: (DataTaskPublisher)
shared: request unlimited
shared: receive value: (109 bytes)
subscription1 received: '109 bytes'
shared: receive finished
shared: receive value: (94 bytes)
subscription2 received: '94 bytes'
shared: receive finished

可以清楚的看到，當(dāng) DataTaskPublisher 不共享時(shí)，它收到了兩個(gè) Subscription！ 在這種情況下，請(qǐng)求會(huì)運(yùn)行兩次。

但是有一個(gè)問(wèn)題：如果第二個(gè)訂閱者是在共享請(qǐng)求完成之后來(lái)的呢？ 我們可以通過(guò)延遲第二次訂閱來(lái)模擬這種情況：

example("share") { 
    let shared = URLSession.shared
        .dataTaskPublisher(for: URL(string: "https://random-data-api.com/api/v2/appliances")!)
        .map(\.data)
        .print("shared")
        .share()
    print("subscribing first")
    let subscription1 = shared.sink(
        receiveCompletion: { _ in },
        receiveValue: { print("subscription1 received: '\($0)'") }
    )
    .store(in: &subscriptions)
    DispatchQueue.main.asyncAfter(deadline: .now() + 5) {
        print("subscribing second")
        let subscription2 = shared.sink(
            receiveCompletion: { print("subscription2 completion \($0)") },
            receiveValue: { print("subscription2 received: '\($0)'") }
        )
        .store(in: &subscriptions)
    }
}

運(yùn)行 Playground，我們會(huì)看到 subscription2 什么值也沒(méi)有收到：

--- share ---
subscribing first
shared: receive subscription: (DataTaskPublisher)
shared: request unlimited
shared: receive value: (102 bytes)
subscription1 received: '102 bytes'
shared: receive finished
subscribing second
subscription2 completion finished

在創(chuàng)建 subscription2 時(shí)，請(qǐng)求已經(jīng)完成并且結(jié)果數(shù)據(jù)已經(jīng)發(fā)出。如何確保兩個(gè) Subscription 都收到請(qǐng)求結(jié)果？

multicast(_:)

在上游 Publisher 完成后，要與 Publisher 共享單個(gè) Subscription 并將值重播給新 Subscriber，我們需要類似 shareReplay() Operator。不幸的是，這個(gè) Operator 不是 Combine 的一部分。我們將在后續(xù)文章中創(chuàng)建一個(gè)。

在“網(wǎng)絡(luò)”中，我們使用了 multicast(_:)。此 Operator 基于 share() 構(gòu)建，并使用我們選擇的 Subject 將值發(fā)布給Subscriber。 multicast(_:) 的獨(dú)特之處在于它返回的 Publisher 是一個(gè) ConnectablePublisher。這意味著它不會(huì)訂閱上游 Publisher，直到我們調(diào)用它的 connect() 方法。這讓你有足夠的時(shí)間來(lái)設(shè)置我們需要的所有 Subscriber，然后再讓它連接到上游 Publisher 并開(kāi)始工作。

要調(diào)整前面的示例以使用 multicast(_:)，我們可以編寫(xiě)：

example("multicast") { 
    let subject = PassthroughSubject<Data, URLError>()
    let multicasted = URLSession.shared
        .dataTaskPublisher(for: URL(string: "https://random-data-api.com/api/v2/appliances")!)
        .map(\.data)
        .print("multicast")
        .multicast(subject: subject)
    let subscription1 = multicasted
        .sink(
            receiveCompletion: { _ in },
            receiveValue: { print("subscription1 received: '\($0)'") }
        )
        .store(in: &subscriptions)
    let subscription2 = multicasted
        .sink(
            receiveCompletion: { _ in },
            receiveValue: { print("subscription2 received: '\($0)'") }
        )
        .store(in: &subscriptions)
    let cancellable = multicasted.connect()
        .store(in: &subscriptions)
}

我們準(zhǔn)備一個(gè) subject，它傳遞上游 Publisher 發(fā)出的值和完成事件。使用上述 subject 準(zhǔn)備多播 Publisher。

運(yùn)行 Playground，結(jié)果輸出：

--- multicast ---
multicast: receive subscription: (DataTaskPublisher)
multicast: request unlimited
multicast: receive value: (116 bytes)
subscription1 received: '116 bytes'
subscription2 received: '116 bytes'
multicast: receive finished

一個(gè)多播 Publisher，和所有的 ConnectablePublisher 一樣，也提供了一個(gè) autoconnect() 方法，這使它像 share() 一樣工作：第一次訂閱它時(shí)，它會(huì)連接到上游 Publisher 并立即開(kāi)始工作。

Future

雖然 share() 和 multicast(_:) 為你提供了成熟的 Publisher，Combine 還提供了另一種讓我們共享計(jì)算結(jié)果的方法：Future：

example("future") { 
    func performSomeWork() throws -> Int {
        print("Performing some work and returning a result")
        return 5
    }
    let future = Future<Int, Error> { fulfill in
        do {
            let result = try performSomeWork()
            fulfill(.success(result))
        } catch {
            fulfill(.failure(error))
        }
    }
    print("Subscribing to future...")
    let subscription1 = future
        .sink(
            receiveCompletion: { _ in print("subscription1 completed") },
            receiveValue: { print("subscription1 received: '\($0)'") }
        )
        .store(in: &subscriptions)
    DispatchQueue.main.asyncAfter(deadline: .now() + 3) { 
        let subscription2 = future
            .sink(
                receiveCompletion: { _ in print("subscription2 completed") },
                receiveValue: { print("subscription2 received: '\($0)'") }
            )
            .store(in: &subscriptions)
    }
}

運(yùn)行將輸出：

--- future ---
Performing some work and returning a result
Subscribing to future...
subscription1 received: '5'
subscription1 completed
subscription2 received: '5'
subscription2 completed

在代碼中，我們提供一個(gè)模擬 Future 執(zhí)行的工作。創(chuàng)造新的 Future, 工作立即開(kāi)始，無(wú)需等待 Subscriber。

如果成功，則給 Promise 提供值。如果失敗，將錯(cuò)誤傳遞給 Promise。Subscription 一次表明我們收到了結(jié)果。第二次 Subscription 表明我們也收到了結(jié)果，沒(méi)有執(zhí)行兩次工作。

從資源的角度來(lái)看：

• Future 是一個(gè)類，而不是一個(gè)結(jié)構(gòu)。
• 創(chuàng)建后，它立即調(diào)用閉包開(kāi)始計(jì)算結(jié)果。
• 它存儲(chǔ) Promise 的結(jié)果并將其交付給當(dāng)前和未來(lái)的 Subscriber。

在實(shí)踐中，這意味著 Future 是一種便捷的方式，可以立即開(kāi)始執(zhí)行某些工作，同時(shí)只執(zhí)行一次工作并將結(jié)果交付給任意數(shù)量的 Subscriber。但它執(zhí)行工作并返回單個(gè)結(jié)果，而不是結(jié)果流，因此使用場(chǎng)景比成熟的 Subscriber 要更少。當(dāng)我們需要共享網(wǎng)絡(luò)請(qǐng)求產(chǎn)生的單個(gè)結(jié)果時(shí)，它是一個(gè)很好的選擇！

內(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全面使用詳解的詳細(xì)內(nèi)容，更多關(guān)于Combine 特定用例的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！