特定用例下的Combine全面使用詳解

更新時間：2022年12月26日 10:32:26 作者：Layer

這篇文章主要為大家介紹了特定用例下的Combine全面使用詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪

引言

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

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

網(wǎng)絡(luò)

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

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

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

URLSession Extension

URLSession 是 Apple 平臺下執(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ù)公開了一個 Combine Publisher。 Combine 使用具有兩個變體的單個 API 處理這些任務(wù)。入?yún)?shù)為 URLRequest 或 URL：

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

下面看看如何使用這個 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ǔ)代碼后，進入我們的 example 函數(shù)。我們使用 URL 作為參數(shù)的 dataTaskPublisher(for:) 。確保我們處理了錯誤。請求結(jié)果是包含 Data 和 URLResponse 的元組。Combine 在 URLSession.dataTask 上提供了 Publisher 而不是閉包。最后保留 Subscription，否請求它會立即被取消，并且請求永遠(yuǎn)不會執(zhí)行。

Codable

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

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

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

我們在 tryMap Operator 中解碼 JSON，該方法有效，但 Combine 提供了一個在該場景下更合適的 Operator 來幫助減少代碼：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: &amp;subscriptions)

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

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

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

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

還有另一種解決方案：使用 multicast() Operator，它返回一個 ConnectablePublisher，該 Publisher 為每個 Subscriber 創(chuàng)建一個單獨的 Subject。它允許我們多次訂閱 Subject，然后在我們準(zhǔn)備好時，調(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&lt;Data, URLError&gt;() }
    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: &amp;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: &amp;subscriptions)
    let subscription = publisher.connect()
}

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

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

調(diào)試

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

打印事件

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

即使是這樣的簡單案例：

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

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

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

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

我們可以創(chuàng)建一個簡單的記錄器來顯示每個字符串之間的時間間隔，以便了解發(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é)果顯示每條打印行之間的時間：

--- 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í)行副作用

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

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

考慮這段代碼：

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

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

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

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

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

我們忘記保留 AnyCancellable。因此 Subscription 開始但立即被取消。

使用 Debugger Operator

Debugger 操作符是我們在萬不得已的時候確實需要使用的 Operator。

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

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

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

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

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

Timer

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

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

使用 RunLoop

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

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

RunLoop 實現(xiàn)了我們將后續(xù)文章中了解的 Scheduler 協(xié)議。它定義了幾種相對低級別的方法，并且是唯一一種可以讓你創(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: &amp;subscriptions)
}

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

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 類會更好。

使用 Timer 類

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

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

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

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

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

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

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

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

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

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

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

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: &amp;subscriptions)
}

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

使用 DispatchQueue

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

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

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

KVO

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

此外，Combine 還提供了觀察單個變量變化的機制：

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

publisher(for:options:)

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

下面是一個對 OperationQueue 的 operationCount 屬性 KVO 的示例：

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

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

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

自定義的 KVO 兼容屬性

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

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

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

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

在 Playground 上嘗試一個例子：

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)建了一個 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ù)是一個具有四個值的選項集：.initial、.prior、.old 和 .new。默認(rèn)值為 [.initial]，這就是為什么我們會看到 Publisher 在發(fā)出任何更改之前發(fā)出初始值。以下是選項的細(xì)分：

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

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

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

如果我們不想要初始值，你可以簡單地寫：

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

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

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

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

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

該屬性首先從 0 更改為 100，因此我們獲得兩個值：0 和 100。然后，它從 100 更改為 200，因此我們再次獲得兩個值：100 和 200。

ObservableObject

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

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

這是一個例子：

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é)議使編譯器自動生成 objectWillChange 屬性。它是一個 ObservableObjectPublisher，它發(fā)出 Void 值并且永不失敗。

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

資源管理

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

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

share()

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

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

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

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

假設(shè)我們正在執(zhí)行一個網(wǎng)絡(luò)請求，你希望多個 Subscriber 無需多次請求即可接收結(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)
}

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

在 Playground 中運行此代碼：

--- 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

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

要驗證請求只發(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 不共享時，它收到了兩個 Subscription！在這種情況下，請求會運行兩次。

但是有一個問題：如果第二個訂閱者是在共享請求完成之后來的呢？我們可以通過延遲第二次訂閱來模擬這種情況：

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: &amp;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: &amp;subscriptions)
    }
}

運行 Playground，我們會看到 subscription2 什么值也沒有收到：

--- 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 時，請求已經(jīng)完成并且結(jié)果數(shù)據(jù)已經(jīng)發(fā)出。如何確保兩個 Subscription 都收到請求結(jié)果？

multicast(_:)

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

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

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

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)備一個 subject，它傳遞上游 Publisher 發(fā)出的值和完成事件。使用上述 subject 準(zhǔn)備多播 Publisher。

運行 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

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

Future

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

example("future") { 
    func performSomeWork() throws -&gt; Int {
        print("Performing some work and returning a result")
        return 5
    }
    let future = Future&lt;Int, Error&gt; { 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: &amp;subscriptions)
    DispatchQueue.main.asyncAfter(deadline: .now() + 3) { 
        let subscription2 = future
            .sink(
                receiveCompletion: { _ in print("subscription2 completed") },
                receiveValue: { print("subscription2 received: '\($0)'") }
            )
            .store(in: &amp;subscriptions)
    }
}

運行將輸出：

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

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

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

從資源的角度來看：

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

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

內(nèi)容參考