在具體介紹 Combine 之前，有兩個(gè)重要的概念需要簡(jiǎn)要介紹一下：

• 觀(guān)察者模式
• 響應(yīng)式編程

觀(guān)察者模式

觀(guān)察者模式（Observer Pattern）是一種設(shè)計(jì)模式，用來(lái)描述一對(duì)多關(guān)系：一個(gè)對(duì)象發(fā)生改變時(shí)將自動(dòng)通知其他對(duì)象，其他對(duì)象將相應(yīng)做出反應(yīng)。這兩類(lèi)對(duì)象分別被稱(chēng)為被 觀(guān)察目標(biāo)（Observable）和 觀(guān)察者（Observer），也就是說(shuō)一個(gè)觀(guān)察目標(biāo)可以對(duì)應(yīng)多個(gè)觀(guān)察者，觀(guān)察者可以訂閱它們感興趣的內(nèi)容，當(dāng)觀(guān)察目標(biāo)內(nèi)容改變時(shí)，它會(huì)向這些觀(guān)察者廣播通知（調(diào)用 Observer 的更新方法）。有一點(diǎn)需要說(shuō)明的是，觀(guān)察者之間彼此時(shí)互相獨(dú)立的，也并不知道對(duì)方的存在。

在 Swift 中，一個(gè)簡(jiǎn)單的觀(guān)察者模式可以被描述為：

protocol Observable {
    associatedtype T: Observer
    mutating func attach(observer: T)
}

protocol Observer {
    associatedtype State
    func notify(_ state: State)
}

struct AnyObservable<T: Observer>: Observable{

    var state: T.State {
        didSet {
            notifyStateChange()
        }
    }

    var observers: [T] = []

    init(_ state: T.State) {
        self.state = state
    }

    mutating func attach(observer: T) {
        observers.append(observer)
        observer.notify(state)
    }

    private func notifyStateChange() {
        for observer in observers {
            observer.notify(state)
        }
    }
}

struct AnyObserver<S>: Observer {

    private let name: String

    init(name: String) {
        self.name = name
    }

    func notify(_ state: S) {
        print("\(name)'s state updated to \(state)")
    }
}

var observable = AnyObservable<AnyObserver<String>>("hello")
let observer = AnyObserver<String>(name: "My Observer")
observable.attach(observer: observer)
observable.state = "world"

// "My Observer's state updated: hello"
// "My Observer's state updated: world"

響應(yīng)式編程

響應(yīng)式編程（Reactive Programming）是一種編程思想，相對(duì)應(yīng)的也有面向過(guò)程編程、面向?qū)ο缶幊獭⒑瘮?shù)式編程等等。不同的是，響應(yīng)式編程的核心是面向異步數(shù)據(jù)流和變化的。

在現(xiàn)在的前端世界中，我們需要處理大量的事件，既有用戶(hù)的交互，也有不斷的網(wǎng)絡(luò)請(qǐng)求，還有來(lái)自系統(tǒng)或者框架的各種通知，因此也無(wú)可避免產(chǎn)生紛繁復(fù)雜的狀態(tài)。使用響應(yīng)式編程后，所有事件將成為異步的數(shù)據(jù)流，更加方便的是可以對(duì)這些數(shù)據(jù)流可以進(jìn)行組合變換，最終我們只需要監(jiān)聽(tīng)所關(guān)心的數(shù)據(jù)流的變化并做出響應(yīng)即可。

舉個(gè)有趣的例子來(lái)解釋一下：

當(dāng)你早上起床之后，你需要一邊洗漱一邊烤個(gè)面包，最后吃早飯。

傳統(tǒng)的編程方法：

func goodMorning() {
    wake {
        let group = DispatchGroup()
        group.enter()
        wash {
            group.leave()
        }
        group.enter()
        cook {
            group.leave()
        }
        group.notify(queue: .main) {
            eat {
                print("Finished")
            }
        }
    }
}

響應(yīng)式編程：

func reactiveGoodMorning() {
    let routine = wake.rx.flapLatest {
        return Observable.zip(wash, cook)
    }.flapLatest {
        return eat.rx
    }
    routine.subsribe(onCompleted: {
        print("Finished")
    })
}

不同于傳統(tǒng)可以看到 wake/wash/cook/eat 這幾個(gè)事件通過(guò)一些組合轉(zhuǎn)換被串聯(lián)成一個(gè)流，我們也只需要訂閱這個(gè)流就可以在應(yīng)該響應(yīng)的時(shí)候得到通知。

Marble Diagram

為了更方便理解數(shù)據(jù)流，我們通常用一種叫 Marble Diagram 的圖象形式來(lái)表示數(shù)據(jù)流基于時(shí)間的變化。

我們用從左至右的箭頭表示時(shí)間線(xiàn)，不同的形狀代表 Publisher 發(fā)出的值（Input），豎線(xiàn)代表正常終止，叉代表發(fā)生錯(cuò)誤而終止。上圖中，上面一條時(shí)間線(xiàn)是一個(gè)數(shù)據(jù)流，中間的方塊代表組合變換，下方的另外一條時(shí)間線(xiàn)代表經(jīng)過(guò)變換后的數(shù)據(jù)流。

Combine

現(xiàn)在可以進(jìn)入正題了，什么是 Combine？

官方文檔的介紹是：

The Combine framework provides a declarative Swift API for processing values over time. These values can represent user interface events, network responses, scheduled events, and many other kinds of asynchronous data.
By adopting Combine, you’ll make your code easier to read and maintain, by centralizing your event-processing code and eliminating troublesome techniques like nested closures and convention-based callbacks.

簡(jiǎn)而言之，Combine 可以使代碼更加簡(jiǎn)潔、易于維護(hù)，也免除了飽受詬病的嵌套閉包和回調(diào)地獄。Combine 是 Reactive Programming 在 Swift 中的一個(gè)實(shí)現(xiàn)，更確切的說(shuō)是對(duì) ReactiveX (Reactive Extensions, 簡(jiǎn)稱(chēng) Rx) 的實(shí)現(xiàn)，而這個(gè)實(shí)現(xiàn)正是基于觀(guān)察者模式的。

Combine 是基于泛型實(shí)現(xiàn)的，是類(lèi)型安全的。它可以無(wú)縫地接入已有的工程，用來(lái)處理現(xiàn)有的 Target/Action、Notification、KVO、callback/closure 以及各種異步網(wǎng)絡(luò)請(qǐng)求。

在 Combine 中，有幾個(gè)重要的組成部分：

• 發(fā)布者：Publiser
• 訂閱者：Subscriber
• 操作符：Operator

Publisher

在 Combine 中，Publisher 是觀(guān)察者模式中的 Observable，并且可以通過(guò)組合變換（利用 Operator）重新生成新的 Publisher。

public protocol Publisher {
    /// The kind of values published by this publisher.
    associatedtype Output

    /// The kind of errors this publisher might publish.
    ///
    /// Use `Never` if this `Publisher` does not publish errors.
    associatedtype Failure : Error

    /// This function is called to attach the specified `Subscriber` to 
    /// this `Publisher` by `subscribe(_:)`
    ///
    /// - SeeAlso: `subscribe(_:)`
    /// - Parameters:
    ///     - subscriber: The subscriber to attach to this `Publisher`.
    ///                   once attached it can begin to receive values.
    func receive<S>(subscriber: S) where S : Subscriber, 
        Self.Failure == S.Failure, Self.Output == S.Input
}

在 Publisher 的定義中，Output 代表數(shù)據(jù)流中輸出的值，值的更新可能是同步，也可能是異步，F(xiàn)ailure 代表可能產(chǎn)生的錯(cuò)誤，也就是說(shuō) Pubslier 最核心的是定義了值與可能的錯(cuò)誤。Publisher 通過(guò) receive(subscriber:) 用來(lái)接受訂閱，并且要求 Subscriber 的值和錯(cuò)誤類(lèi)型要一致來(lái)保證類(lèi)型安全。

例如來(lái)自官方的一個(gè)例子：

extension NotificationCenter {
        struct Publisher: Combine.Publisher {
                typealias Output = Notification
                typealias Failure = Never
                init(center: NotificationCenter, name: Notification.Name, object: Any? = nil)
        }
}

這個(gè) Publisher 提供的值就是 Notification 類(lèi)型，而且永遠(yuǎn)不會(huì)產(chǎn)生錯(cuò)誤（Never）。這個(gè)擴(kuò)展非常有用，可以很方便地將任何 Notification 轉(zhuǎn)換成 Publisher，便于我們將應(yīng)用改造成 Reactive。

讓我們?cè)倏匆粋€(gè)的例子：

let justPubliser = Publishers.Just("Hello")

justPubliser 會(huì)給每個(gè)訂閱者發(fā)送一個(gè) "Hello" 消息，然后立即結(jié)束（這個(gè)數(shù)據(jù)流只包含一個(gè)值）。

類(lèi)似的，Combine 為我們提供了一些便捷的 Publisher 的實(shí)現(xiàn)，除了上面的 Just，這里也列出一些很有用的 Publisher。

Empty

Empty 不提供任何值的更新，并且可以選擇立即正常結(jié)束。

Once 可以提供以下兩種數(shù)據(jù)流之一：

發(fā)送一次值更新，然后立即正常結(jié)束（和 Just 一樣）
立即因錯(cuò)誤而終止
Fail

Fail 和 Once 很像，也是提供兩種情況之一：

發(fā)送一次值更新，然后立即因錯(cuò)誤而終止
立即因錯(cuò)誤而終止

Sequence 將給定的一個(gè)序列按序通知到訂閱者。

Future 初始化需要提供執(zhí)行具體操作的 closure，這個(gè)操作可以是異步的，并且最終返回一個(gè) Result，所以 Future 要么發(fā)送一個(gè)值，然后正常結(jié)束，要么因錯(cuò)誤而終止。在將一些異步操作轉(zhuǎn)換為 Publisher 時(shí)非常有用，尤其是網(wǎng)絡(luò)請(qǐng)求。例如：

let apiRequest = Publishers.Future { promise in
    URLSession.shared.dataTask(with: url) { data, _, _ in
        promise(.success(data))
    }.resume()
}

Deferred 初始化需要提供一個(gè)生成 Publisher 的 closure，只有在有 Subscriber 訂閱的時(shí)候才會(huì)生成指定的 Publisher，并且每個(gè) Subscriber 獲得的 Publisher 都是全新的。

那么，在之前的 Just Publisher 例子中，假設(shè)我們要實(shí)現(xiàn)延遲兩秒后再發(fā)送消息呢？也很簡(jiǎn)單。我們前面說(shuō)到 Publisher 都是可以組合變換的，這些組合變換可以通過(guò)操作符來(lái)實(shí)現(xiàn)的。

上面的例子利用 delay 操作符就可以改寫(xiě)為：

let delayedJustPublisher = justPubliser.delay(for: 2, scheduler: DispatchQueue.main)

在文章的后半部分，我們將了解一些常用操作符以及它們的使用場(chǎng)景。

Subscriber

和 Publisher 相對(duì)應(yīng)的，Subscriber 就是觀(guān)察者模式中 Observer。

public protocol Subscriber : CustomCombineIdentifierConvertible {
    /// The kind of values this subscriber receives.
    associatedtype Input

    /// The kind of errors this subscriber might receive.
    ///
    /// Use `Never` if this `Subscriber` cannot receive errors.
    associatedtype Failure : Error

    /// Tells the subscriber that it has successfully subscribed to 
    /// the publisher and may request items.
    ///
    /// Use the received `Subscription` to request items from the publisher.
    /// - Parameter subscription: A subscription that represents the connection 
    /// between publisher and subscriber.
    func receive(subscription: Subscription)

    /// Tells the subscriber that the publisher has produced an element.
    ///
    /// - Parameter input: The published element.
    /// - Returns: A `Demand` instance indicating how many more elements 
    /// the subcriber expects to receive.
    func receive(_ input: Self.Input) -> Subscribers.Demand

    /// Tells the subscriber that the publisher has completed publishing, 
    /// either normally or with an error.
    ///
    /// - Parameter completion: A `Completion` case indicating 
    /// whether publishing completed normally or with an error.
    func receive(completion: Subscribers.Completion<Self.Failure>)
}

可以看出，從概念上和我們之前定義的簡(jiǎn)單觀(guān)察者模式相差無(wú)幾。Publisher 在自身狀態(tài)改變時(shí)，調(diào)用 Subscriber 的三個(gè)不同方法（receive(subscription), receive(_:Input), receive(completion:)）來(lái)通知 Subscriber。

這里也可以看出，Publisher 發(fā)出的通知有三種類(lèi)型：

• Subscription：Subscriber 成功訂閱的消息，只會(huì)發(fā)送一次，取消訂閱會(huì)調(diào)用它的 Cancel 方法來(lái)釋放資源
• Value（Subscriber 的 Input，Publisher 中的 Output）：真正的數(shù)據(jù)，可能發(fā)送 0 次或多次
• Completion：數(shù)據(jù)流終止的消息，包含兩種類(lèi)型：.finished 和 .failure(Error)，最多發(fā)送一次，一旦發(fā)送了終止消息，這個(gè)數(shù)據(jù)流就斷開(kāi)了，當(dāng)然有的數(shù)據(jù)流可能永遠(yuǎn)沒(méi)有終止

大部分場(chǎng)景下我們主要關(guān)心的是后兩種消息，即數(shù)據(jù)流的更新和終止。

Combine 內(nèi)置的 Subscriber 有三種：

• Sink
• Assign
• Subject

Sink 是非常通用的 Subscriber，我們可以自由的處理數(shù)據(jù)流的狀態(tài)。

例如：

let once: Publishers.Once<Int, Never> = Publishers.Once(100)
let observer: Subscribers.Sink<Publishers.Once<Int, Never>> = Subscribers.Sink(receiveCompletion: {
    print("completed: \($0)")
}, receiveValue: {
    print("received value: \($0)")
})
once.subscribe(observer)

// received value: 100
// completed: finished

Publisher 甚至提供了 sink(receiveCompletion:, receiveValue:) 方法來(lái)直接訂閱。

Assign 可以很方便地將接收到的值通過(guò) KeyPath 設(shè)置到指定的 Class 上（不支持 Struct），很適合將已有的程序改造成 Reactive。

例如：

class Student {
    let name: String
    var score: Int

    init(name: String, score: Int) {
        self.name = name
        self.score = score
    }
}

let student = Student(name: "Jack", score: 90)
print(student.score)
let observer = Subscribers.Assign(object: student, keyPath: \Student.score)
let publisher = PassthroughSubject<Int, Never>()
publisher.subscribe(observer)
publisher.send(91)
print(student.score)
publisher.send(100)
print(student.score)

// 90
// 91
// 100

在例子中，一旦 publisher 的值發(fā)生改變，相應(yīng)的，student 的 score 也會(huì)被更新。

PassthroughSubject 這里是 Combine 內(nèi)置的一個(gè) Publisher，可能你會(huì)好奇：前面不是說(shuō) Subject 是一種 Observer 嗎，這里怎么又是 Pulisher 呢？

Subject

有些時(shí)候我們想隨時(shí)在 Publisher 插入值來(lái)通知訂閱者，在 Rx 中也提供了一個(gè) Subject 類(lèi)型來(lái)實(shí)現(xiàn)。Subject 通常是一個(gè)中間代理，即可以作為 Publisher，也可以作為 Observer。Subject 的定義如下：

public protocol Subject : AnyObject, Publisher {

    /// Sends a value to the subscriber.
    ///
    /// - Parameter value: The value to send.
    func send(_ value: Self.Output)

    /// Sends a completion signal to the subscriber.
    ///
    /// - Parameter completion: A `Completion` instance which indicates 
    /// whether publishing has finished normally or failed with an error.
    func send(completion: Subscribers.Completion<Self.Failure>)
}

作為 Observer 的時(shí)候，可以通過(guò) Publisher 的 subscribe(_:Subject) 方法訂閱某個(gè) Publisher。

作為 Publisher 的時(shí)候，可以主動(dòng)通過(guò) Subject 的兩個(gè) send 方法，我們可以在數(shù)據(jù)流中隨時(shí)插入數(shù)據(jù)。目前在 Combine 中，有三個(gè)已經(jīng)實(shí)現(xiàn)對(duì) Subject: AnySubject，CurrentValueSubject 和 PassthroughSubject 。

利用 Subject 可以很輕松的將現(xiàn)在已有代碼的一部分轉(zhuǎn)為 Reactive，一個(gè)比較典型的使用場(chǎng)景是：

// Before
class ContentManager {

    var content: [String] {
        didSet {
            delegate?.contentDidChange(content)
        }
    }

    func getContent() {
        content = ["hello", "world"]
    }
}

// After
class RxContentController {

    var content = CurrentValueSubject<[String], NSError>([])

    func getContent() {
        content.value = ["hello", "world"]
    }
}

CurrentValueSubject 的功能很簡(jiǎn)單，就是包含一個(gè)初始值，并且會(huì)在每次值變化的時(shí)候發(fā)送一個(gè)消息，這個(gè)值會(huì)被保存，可以很方便的用來(lái)替代 Property Observer。在上例中，以前需要實(shí)現(xiàn) delegate 來(lái)獲取變化，現(xiàn)在只需要訂閱 content 的變化即可，并且它作為一個(gè) Publisher，可以很方便的利用操作符進(jìn)行組合變換。

PassthroughSubject 和 CurrentValueSubject 幾乎一樣，只是沒(méi)有初始值，也不會(huì)保存任何值。

AnyPublisher、AnySubscriber、AnySubject

前面說(shuō)到 Publisher、Subscriber、Subject 是類(lèi)型安全的，那么在使用中不同類(lèi)型的 Publisher、Subscriber、Subject 勢(shì)必會(huì)造成一些麻煩。

考慮下面這種情況：

class StudentManager {

    let namesPublisher: ??? // what's the type?

    func updateStudentsFromLocal() {
        let student1 = Student(name: "Jack", score: 75)
        let student2 = Student(name: "David", score: 80)
        let student3 = Student(name: "Alice", score: 96)
        let namesPublisher: Publishers.Sequence<[String], Never> = Publishers.Sequence<[Student], Never>(sequence: [student1, student2, student3]).map { $0.name }
        self.namesPublisher = namesPublisher
    }

    func updateStudentsFromNetwork() {
        let namesPublisher: Publishers.Future<[String], Never> = Publishers.Future { promise in
            getStudentsFromNetwork {
                let names: [String] = ....
                promise(.success([names]))
            }
        }
        self.namesPublisher = namesPublisher
    }
}

這里 namesPublisher 究竟該是什么類(lèi)型呢？第一個(gè)方法返回的是 Publishers.Sequence<[String], Never>，第二個(gè)方法是 Publishers.Future<[String], Never>，而我們是無(wú)法定義成 Publisher<[String], Never> 的。

AnyPublisher、AnySubscriber、AnySubject 正是為這樣的場(chǎng)景設(shè)計(jì)的，它們是通用類(lèi)型，任意的 Publisher、Subscriber、Subject 都可以通過(guò) eraseToAnyPublisher()、eraseToAnySubscriber()、eraceToAnySubject() 轉(zhuǎn)化為對(duì)應(yīng)的通用類(lèi)型。

class StudentManager {

    let namePublisher: AnyPublisher<[String, Never]>

    func updateStudentsFromLocal() {
        let namePublisher: AnyPublisher<[String, Never]> = Publishers.Sequence<[Student], Never>(sequence: students).map { $0.name }.eraseToAnyPublisher()
        self.namePublisher = namePublisher
    }

    func updateStudentsFromNetwork() {
        let namePublisher: AnyPublisher<[String, Never]> = Publishers.Future { promise in
            promise(.success([names]))
        }.eraseToAnyPublisher()
        self.namePublisher = namePublisher
    }
}

AnyPublisher、AnySubscriber、AnySubject 的另外一個(gè)實(shí)用場(chǎng)景是創(chuàng)建自定義的 Publisher/Subscriber/Subject，因?yàn)樵诳蚣苤兴鼈円呀?jīng)是實(shí)現(xiàn)好相應(yīng)的協(xié)議了。

let justPubliser = AnyPublisher<String, NSError> { subscribe in
    _ = subscribe.receive("hello")  // ignore demand
    subscribe.receive(completion: .finished)
}
let subscriber = AnySubscriber<String, NSError>(receiveValue: { input in
    print("Received input: \(input)")
    return .unlimited
}, receiveCompletion: { completion in
    print("Completed with \(completion)")
})
justPubliser.subscribe(subscriber)

// Received input: hello
// Completed with finished

Cancellable

讓我們回到上文提到的 Future 的一個(gè)例子：

let apiRequest = Publishers.Future { promise in
    URLSession.shared.dataTask(with: url) { data, _, _ in
        promise(.success(data))
    }.resume()
}

它似乎能夠很好的工作，但是一個(gè)比較經(jīng)典的場(chǎng)景是：用戶(hù)上傳某個(gè)文件，上傳到一半發(fā)現(xiàn)選錯(cuò)了就點(diǎn)擊取消，我們也應(yīng)該取消這個(gè)上傳。類(lèi)似的還有一些臨時(shí)生成的資源需要在取消訂閱時(shí)釋放。這在 Combine 中該怎么實(shí)現(xiàn)呢？

Combine 中提供了 Cancellable 這個(gè)協(xié)議。定義很簡(jiǎn)單：

protocol Cancellable {
    /// Cancel the activity.
    func cancel()
}

前面我們提到 Publisher 在被訂閱時(shí)會(huì)給 Subscriber 發(fā)送一個(gè) Subscription 消息，這個(gè) Subscription 恰好也實(shí)現(xiàn)了 Cancellable 協(xié)議，在取消訂閱時(shí)，會(huì)調(diào)用它的 cancel 方法。

這樣，我們就可以實(shí)現(xiàn)一個(gè)簡(jiǎn)單的例子：

struct AnySubscription: Subscription {
    private let cancellable: Cancellable

    init(_ cancel: @escaping () -> Void) {
        cancellable = AnyCancellable(cancel)
    }

    func cancel() {
        cancellable.cancel()
    }
}

let downloadPublisher = AnyPublisher<Data?, Never> { subscribe in
    let task = URLSession.shared.uploadTask(with: request, fromFile: file) { (data, _, _) in
        _ = subscribe.receive(data) // ignore demand
        subscribe.receive(completion: .finished)
    }
    let subscription = AnySubscription {
        task.cancel()
    }
    subscribe.receive(subscription: subscription)
    task.resume()
}
let cancellable = downloadPublisher.sink { data in
    print("Received data: \(data)")
}

// Cancel the task before it finishes
cancellable.cancel()

操作符

操作符是 Combine 中非常重要的一部分，通過(guò)各式各樣的操作符，可以將原來(lái)各自不相關(guān)的邏輯變成一致的（unified）、聲明式的（declarative）的數(shù)據(jù)流。

轉(zhuǎn)換操作符：

• map/mapError
• flatMap
• replaceNil
• scan
• setFailureType

過(guò)濾操作符：

• filter
• compactMap
• removeDuplicates
• replaceEmpty/replaceError

reduce 操作符：

• collect
• ignoreOutput
• reduce

運(yùn)算操作符：

• count
• min/max

匹配操作符：

• contains
• allSatisfy

序列操作符：

• drop/dropFirst
• append/prepend
• prefix/first/last/output

組合操作符：

• combineLatest
• merge
• zip

錯(cuò)誤處理操作符：

• assertNoFailure
• catch
• retry

時(shí)間控制操作符：

measureTimeInterval
debounce
delay
throttle
timeout

其他操作符：

• encode/decode
• switchToLatest
• share
• breakpoint/breakpointOnError
• handleEvents

以下內(nèi)容僅供參考，目前并未完成所有的操作符的詳細(xì)介紹
map/mapError

map 將收到的值按照給定的 closure 轉(zhuǎn)換為其他值，mapError 則將錯(cuò)誤轉(zhuǎn)換為另外一種錯(cuò)誤類(lèi)型。

func map<T>(_ transform: (Output) -> T) -> Publishers.Just<T>