从响应式编程到 Combine 实践-响应式编程函数式编程

作者｜何星

大约一年前，Resso 接入了 Combine，利用响应式编程简化了代码逻辑，也积累了很多实践经验。本文会从响应式编程的基本思想并逐步深入介绍 Combine 的概念与最佳实践，希望能帮助更多的同学顺利上手并实践响应式编程，少踩坑。

等等，Resso 是什么？Resso 来源于 Resonate（共鸣），是字节跳动推出的一个社交音乐流媒体平台，专为下一代音乐发烧友设计，使他们能够通过对音乐的热爱来表达和与他人建立联系。

书回正文，所谓的响应式编程到底是什么呢？

熟悉 Combine 的同学可以直接跳到实践建议部分。

响应式编程

维基百科对响应式编程的定义是：

在计算中，响应式编程是一种面向数据流和变化传播的声明式编程范式。

虽然定义中每个字都认识，但连起来却十分费解。我们可以把定义中的内容分开来理解，逐个击破。首先，让我们来看下声明式编程。

声明式编程

声明式和指令式编程是常见的编程范式。在指令式编程中，开发者通过组合运算、循环、条件等语句让计算机执行程序。声明式与指令式正相反，如果说指令式像是告诉计算机 How to do，而声明式则是告诉计算机 What to do。其实大家都接触过声明式编程，但在编码时并不会意识到。各类 DSL 和函数式编程都属于声明式编程的范畴。

举个例子，假设我们想要获取一个整形数组里的所有奇数。按照指令式的逻辑，我们需要把过程拆解为一步一步的语句：

遍历数组中的所有元素。
判断是否为奇数。
如果是的话，加入到结果中。继续遍历。

var results = [Int]()
for num in values {
    if num %2 != 0 {
        results.append(num)
    }
}

如果按声明式编程来，我们的想法可能是“过滤出所有奇数”，对应的代码就十分直观：

var results = values.filter { $0 % 2 != 0 }

可见上述两种编程方式有着明显的区别：

指令式编程：描述过程（How），计算机直接执行并得结果。
声明式编程：描述结果（What），让计算机为我们组织出具体过程，最后得到被描述的结果。

“面向数据流和变化传播”

用说人话的方式解释，面向数据流和变化传播是响应未来发生的事件流。

事件发布：某个操作发布了事件A，事件A 可以携带一个可选的数据B 。
操作变形：事件A 与数据B 经过一个或多个的操作发生了变化，最终得到事件A' 与数据B'。
订阅使用：在消费端，有一个或多个订阅者来消费处理后的A' 和B'，并进一步驱动程序其他部分 (如 UI )

在这个流程中，无数的事件组成了事件流，订阅者不断接受到新的事件并作出响应。

至此，我们对响应式编程的定义有了初步的理解，即以声明的方式响应未来发生的事件流。在实际编码中，很多优秀的三方库对这套机制进一步抽象，为开发者提供了功能各异的接口。在 iOS 开发中，有三种主流的响应式“流派“。

响应式流派

ReactiveX：RxSwift
Reactive Streams：Combine
Reactive*：ReactiveCocoa / ReactiveSwift ／ReactiveObjc

这三个流派分别是 ReactiveX、Reactive Streams 和 Reactive*。ReactiveX 接下来会详细介绍。Reactive Stream 旨在定义一套非阻塞式异步事件流处理标准，Combine 选择了它作为实现的规范。以 ReactiveCocoa 为代表的 Reactive* 在 Objective-C 时代曾非常流行，但随着 Swift 崛起，更多开发者选择了 RxSwift 或 Combine，导致 Reactive* 整体热度下降不少。

ReactiveX (Reactive Extension)

ReactiveX 最初是微软在 .NET 上实现的一个响应式的拓展。它的接口命名并不直观，如 Observable （可观测的）和 Observer（观测者）。ReactiveX 的优势在于创新地融入了许多函数式编程的概念，使得整个事件流的变形非常灵活。这个易用且强大的概念迅速被各个语言的开发者青睐，因此 ReactiveX 在很多语言都有对应版本的实现（如 RxJS，RxJava，RxSwift），都非常流行。Resso 的 Android 团队就在重度使用 RxJava。

为何选择 Combine

Combine 是 Apple 在 2019 年推出的一个类似 RxSwift 的异步事件处理框架。

通过对事件处理的操作进行组合 (combine) ，来对异步事件进行自定义处理 (这也正是 Combine 框架的名字的由来)。Combine 提供了一组声明式的 Swift API，来处理随时间变化的值。这些值可以代表用户界面的事件，网络的响应，计划好的事件，或者很多其他类型的异步数据。

Resso iOS 团队也曾短暂尝试过 RxSwift，但在仔细考察 Combine 后，发现 Combine 无论是在性能、调试便捷程度上都优于 RxSwift，此外还有内置框架和 SwiftUI 官配的特殊优势，受其多方面优势的吸引，我们全面切换到了 Combine。

Combine 的优势

相较于 RxSwift，Combine 有很多优势：

Apple 出品

内置在系统中，对 App 包体积无影响

性能更好
Debug 更便捷
SwiftUI 官配

性能优势

Combine 的各项操作相较 RxSwift 有 30% 多的性能提升。

Reference: Combine vs. RxSwift Performance Benchmark Test Suite

Debug 优势

由于 Combine 是一方库，在 Xcode 中开启了 Show stack frames without debug symbols and between libraries 选项后，无效的堆栈可以大幅的减少，提升了 Debug 效率。

// 在 GlobalQueue 中接受并答应出数组中的值
[1, 2, 3, 4].publisher
    .receive(on: DispatchQueue.global())
    .sink { value in
        print(value)
    }

Combine 接口

上文提到，Combine 的接口是基于 Reactive Streams Spec 实现的，Reactive Streams 中已经定义好了 Publisher， Subscriber，Subscription 等概念，Apple 在其上有一些微调。

具体到接口层面，Combine API 与 RxSwift API 比较类似，更精简，熟悉 RxSwift 的开发者能无缝快速上手 Combine。Combine 中缺漏的接口可以通过其他已有接口组成替代，少部分操作符也有开源的第三方实现，对生产环境的使用不会产生影响。

OpenCombine

细心的读者可能有发现 Debug 优势的图中出现了一个 OpenCombine。Combine 万般好，但有一个致命的缺点：它要求的最低系统版本是 iOS 13，许多要维护兼容多个系统版本的 App 并不能使用。好在开源社区给力，实现了一份仅要求 iOS 9.0 的 Combine 开源实现：OpenCombine。经内部测试，OpenCombine 的性能与 Combine 持平。OpenCombine 使用上与 Combine 差距很小，未来如果 App 的最低版本升级至 iOS 13 之后，从 OpenCombine 迁移到 Combine 的成本也很低，基本只有简单的文本替换工作。公司内 Resso、剪映、醒图、Lark 都有使用 OpenCombine。

Combine 基础概念

上文提到，Combine 的概念基于 Reactive Streams。响应式编程中的三个关键概念，事件发布／操作变形／订阅使用，分别对应到 Combine 中的 Publisher, Operator 与 Subscriber。

在简化的模型中，首先有一个 Publisher，经过 Operater 变换后被 Subscriber消费。而在实际编码中， Operator 的来源可能是复数个 Publisher，Operator 也可能会被多个 Publisher 订阅，通常会形成一个非常复杂的图。

Publisher

Publisher<Output, Failure: Error>

Publisher 是事件产生的源头。事件是 Combine 中非常重要的概念，可以分成两类，一类携带了值（Value），另外一类标志了结束（Completion）。结束的可以是正常完成（Finished）或失败（Failure）。

Events：
- Value：Output
- Completion
    - Finished
    - Failure(Error)

通常情况下，一个 Publisher 可以生成 N 个事件后结束。需要注意的是，一个 Publisher一旦发出了Completion（可以是正常完成或失败），整个订阅将结束，之后就不能发出任何事件了。

Apple 为官方基础库中的很多常用类提供了 Combine 拓展 Publisher，如 Timer, NotificationCenter, Array, URLSession, KVO 等。利用这些拓展我们可以快速组合出一个 Publisher，如：

// `cancellable` 是用于取消订阅的 token，下文会详细介绍
cancellable = URLSession.shared
    // 生成一个 https://example.com 请求的 Publisher
    .dataTaskPublisher(for: URL(string: "https://example.com")!)
    // 将请求结果中的 Data 转换为字符串，并忽略掉空结果，下面会详细介绍 compactMap
    .compactMap {
        String(data: $0.data, encoding: .utf8)
    }
    // 在主线程接受后续的事件 （上面的 compactMap 发生在 URLSession 的线程中）
    .receive(on: RunLoop.main)
    // 对最终的结果（请求结果对应的字符串）进行消费
    .sink { _ in
        //
    } receiveValue: { resultString in
        self.textView.text = resultString
    }

此外，还有一些特殊的 Publisher 也十分有用：

Future：只会产生一个事件，要么成功要么失败，适用于大部分简单回调场景
Just：对值的简单封装，如Just(1)
@Published：下文会详细介绍在大部分情况下，使用这些特殊的Publisher 以及下文介绍的Subject 可以灵活组合出满足需要的事件源。极少的情况下，需要实现自定义的 Publisher ，可以看这篇文章。

Subscriber

Subscriber<Input, Failure: Error>

Subsriber 作为事件的订阅端，它的定义与 Publisher 对应，Publisher 中的 Output对应Subscriber 的 Input。常用的 Subscriber 有 Sink 和 Assign。

Sink 直接对事件流进行订阅使用，可以对 Value 和 completion 分别进行处理。

Sink 这个单词在初次看到会令人非常费解。这个术语可来源于网络流中的汇点（Sink），我们也可以理解为 The stream goes down the sink。

// 从数组生成一个 Publisher
cancellable = [1, 2, 3, 4, 5].publisher
    .sink { completion in
        // 处理事件流结束
    } receiveValue: { value in
        // 打印会每个值，会依次打印出 1, 2, 3, 4, 5
        print(value)
    }

Assign 是一个特化版的 Sink ，支持通过 KeyPath 直接进行赋值。

let textLabel = UILabel()
cancellable = [1, 2, 3].publisher
    // 将 数字 转换为 字符串，并忽略掉 nil ，下面会详细介绍这个 Operator
    .compactMap { String($0) }
    .assign(to: \.text, on: textLabel)

需要留意的是，如果用 assign 对 self 进行赋值，可能会形成隐式的循环引用，这种情况需要改用 sink 与 weak self 手动进行赋值。

Cancellable & AnyCancellable

细心的读者可能发现了上面出现了一个 cancellable。每一个订阅都会生成一个 AnyCancellable 对象，用于控制订阅的生命周期。通过这个对象，我们可以取消订阅。当这个对象被释放时，订阅也会被取消。

// 取消订阅
cancellable.cancel()

需要注意的是，每一个订阅我们都需要持有这个 cancellable，否则整个订阅会立即被取消并结束掉。

Subscription

Publisher 和 Subscriber 之间是通过 Subscription 建立连接。理解整个订阅过程对后续深入使用 Combine 非常有帮助。

图片来自《SwiftUI 和 Combine 编程》

Combine 的订阅过程其实是一个拉取模型。

Subscriber 发起一个订阅，告诉Publisher 我需要一个订阅。
Publisher 返回一个订阅实体（Subscription）。
Subscriber 通过这个Subscription 去请求固定数量（Demand）的数据。
Publisher 根据Demand 返回事件。单次的Demand 发布完成后，如果Subscriber继续请求事件，Publisher 会继续发布。
继续发布流程。
当Subscriber 请求的事件全部发布完成后，Publisher 会发送一个Completion。

Subject

Subject<Output, Failure: Error>

Subject 是一类特殊的 Publisher，我们可以通过方法调用（如 send()）手动向事件流中注入新的事件。

private let isPlayingPodcastSubject = CurrentValueSubject<Bool, Never>(false)
// 向 isPlayingPodcastPublisher 注入一个新的事件，它的值是 true
isPlayingPodcastSubject.send(true)

Combine 提供了两个常用的 Subject：PassthroughSubject 与 CurrentValueSubject。

PassthroughSubject：透传事件，不会持有最新的Output
CurrentValueSubject：除了传递事件之外，会持有最新的Output

@Published

对于刚接触 Combine 的同学来说，最困扰的问题莫过于难以找到可以直接使用的事件源。Combine 提供了一个 Property Wrapper @Pubilshed 可以快速封装一个变量得到一个 Publisher。

// 声明变量
class Alarm {
    @Published
    public var countDown = 0
}

let alarm = Alarm()

// 订阅变化
let cancellable = alarm.$countDown // Published<Int>.Publisher
    .sink { print($0) }

// 修改 countDown，上面 sink 的闭包会触发
alarm.countDown += 1

上面比较有趣的是 $countDown 访问到的一个 Publisher，这其实是一个语法糖，$ 访问到其实是 countDown 的 projectedValue，正是对应的 Publisher。

@propertyWrapper public struct Published<Value> {
    // ...
    /// The property for which this instance exposes a publisher
    ///
    /// The ``Published/projectedValue` is the property accessed with the `$` operator
    public var projectedValue: Published<Value>.Publisher { mutating get set }
}

@Published 非常适合在模块内对事件进行封装，类型擦除后提供外部进行订阅消费。

实际实践中，对于已有的代码逻辑，使用 @Published 可以在不改动其他代码快速让属性得到 Publisher 的能力。而新编写的代码，如果不会发生错误且需要使用到当前的 Value，@Published 也是很好的选择，除此之外则需要按需考虑使用 PassthroughSubject 或 CurrentValueSubject。

Operator

现实编码中，Publisher 携带的数据类型可能并不满足我们的需求，这时需要使用 Operator 对数据进行变换。Combine 自带了非常丰富的 Operator，接下来会针对其中常用的几个进行介绍。

map, filter, reduce

熟悉函数式编程的同学对这几个 Operator 应该非常熟悉。它们的作用与在数组上的效果非常相似，只不过这次是在异步的事件流中。

例如，对于 map 来说，他会对每个事件中的值进行变换：

[1, 2, 3].publisher
    .map { $0 * 10 }
    .sink { value in
        // 将会答应出 10, 20, 30
        print(value)
    }

filter 也类似，会对每个事件用闭包里的条件进行过滤。reduce 则会对每个事件的值进行计算，最后将计算结果传递给下游。

compactMap

对于 Value 是 Optional 的事件流，可以使用 compactMap 得到一个 Value 为非空类型的 Publisher。

// Publiser<Int?, Never> -> Publisher<Int, Never>
cancellable = [1, nil, 2, 3].publisher
        .compactMap { $0 }
        .map { $0 * 10 }
        .sink { print($0) }

flatMap

flatMap 是一个特殊的操作符，它将每一个的事件转换为一个事件流并合并在一起。举例来说，当用户在搜索框输入文本时，我们可以订阅文本的变化，并针对每一个文本生成对应的搜索请求 Publisher，并将所有 Publisher 的事件汇聚在一起进行消费。

其他常见的 Operator 还有 zip, combineLatest 等。

实践建议

类型擦除

Combine 中的 Publisher 在经过各种 Operator 变换之后会得到一个多层泛型嵌套类型：

URLSession.shared.dataTaskPublisher(for: URL(string: "https://resso.com")!)
    .map { $0.data }
    .decode(type: String.self, decoder: JSONDecoder())
// 这个 publisher 的类型是 Publishers.Decode<Publishers.Map<URLSession.DataTaskPublisher, JSONDecoder. Input>, String, JSONDecoder>

如果在 Publisher 创建变形完成后立即订阅消费，这并不会带来任何问题。但一旦我们需要把这个 Publisher 提供给外部使用时，复杂的类型会暴露过多内部实现细节，同时也会让函数／变量的定义非常臃肿。Combine 提供了一个特殊的操作符 erasedToAnyPublisher，让我们可以擦除掉具体类型：

// 生成一个类型擦除后的请求。函数的返回值更简洁
func requestRessoAPI() -> AnyPublisher<String, Error> {
    let request = URLSession.shared.dataTaskPublisher(for: URL(string: "https://resso.com")!)
        .map { $0.data }
        .decode(type: String.self, decoder: JSONDecoder())
    // Publishers.Decode<Publishers.Map<URLSession.DataTaskPublisher, JSONDecoder. Input>, String, JSONDecoder>
    // to
    // AnyPublisher<String, Error>
    return request.eraseToAnyPublisher()
}

// 在模块外，不用关心 `requestRessoAPI()` 返回的具体类型，直接进行消费
cancellable = requestRessoAPI().sink { _ in

} receiveValue: {
    print($0)
}

通过类型擦除，最终暴露给外部的是一个简单的 AnyPublisher<String, Error>。

Debugging

响应式编程写起来非常的行云流水，但 Debug 起来就相对没有那么愉快了。对此，Combine 也提供了几个 Operator 帮助开发者 Debug。

Debug Operator

print 和 handleEvents

print 可以打印出整个订阅过程从开始到结束的 Subscription 变化与所有值，例如：

cancellable = [1, 2, 3].publisher
  .receive(on: DispatchQueue.global())
  // 使用 `Array Publisher` 作为所有打印内容的前缀
  .print ( "Array Publisher")
  .sink { _ in }

可以得到：

Array Publisher: receive subscription: (ReceiveOn)
Array Publisher: request unlimited
Array Publisher: receive cancel
Array Publisher: receive value: (1)
Array Publisher: receive value: (2)
Array Publisher: receive value: (3)
Array Publisher: receive finished

在一些情况下，我们只对所有变化中的部分事件感兴趣，这时候可以用 handleEvents 对部分事件进行打印。类似的还有 breakpoint，可以在事件发生时触发断点。

画图法

到了万策尽的地步，用图像理清思路也是很好的方法。对于单个 Operator，可以在 RxMarble 找到对应 Operator 确认理解是否正确。对于复杂的订阅，可以画图确认事件流的传递是否符合预期。

let greetings = PassthroughSubject<String, Never>()
let names = PassthroughSubject<String, Never>()
let years = PassthroughSubject<Int, Never>()
// CombineLatest 会选用两个事件流中最新的值生成新的事件流
let greetingNames = Publishers.CombineLatest(greetings, names)
        .map {"\($1) \($0)" }
let wholeSentence = Publishers.CombineLatest(greetingNames, years)
        .map { ")($0), \($1)" }
        .sink { print($0) }

greetings.send("Hello")
names.send("Combine")
years.send(2022)

常见错误

立即开始的 Just 和 Future

对于大部分的 Publisher来说，它们在订阅后才会开始生产事件，但也有一些例外。Just 和 Future 在初始化完成后会立即执行闭包生产事件，这可能会让一些耗时长的操作在不符合预期的时机提前开始，也可能会让第一个订阅错过一些太早开始的事件。

func makeMyPublisher () -> AnyPublisher<Int, Never> {
    Just(calculateTimeConsumingResult())
        .eraseToAnyPublisher()
}

一个可行的解法是在这类 Publisher 外封装一层 Defferred，让它在接收到订阅之后再开始执行内部的闭包。

func makeMyFuture2( ) -> AnyPublisher<Int, Never> {
    Deferred {
        return Just(calculateTimeConsumingResult())
    }.eraseToAnyPublisher()
}

发生错误导致 Subscription 意外结束

func requestingAPI() -> AnyPublisher<String, Error> {
    return URLSession.shared
        .dataTaskPublisher(for: URL(string: "https://resso.com")!)
        .map { $0.data }
        .decode(type: String.self, decoder: JSONDecoder())
        .eraseToAnyPublisher()
}

cancellable = NotificationCenter.default
    .publisher(for: UserCenter.userStateChanged)
    .flatMap({ _ in
        return requestingAPI()
    })
    .sink { completion in

    } receiveValue: { value in
        textLabel.text = value
    }

上面的代码中将用户状态的通知转化成了一个网络请求，并将请求结果更新到一个 Label 上。需要留意的是，一旦某次网络请求发生错误，整个订阅会被结束掉，后续新的通知并不会被转化为请求。

cancellable = NotificationCenter.default
    .publisher(for: UserCenter.userStateChanged)
    .flatMap { value in
        return requestingAPI().materialize()
    }
    .sink { text in
        titleLabel.text = text
    }

解决这个问题的方式有很多，上面使用 materialize 将事件从 Publisher<Output, MyError> 转换为 Publisher<Event<Output, MyError>, Never> 从而避免了错误发生。

Combine 官方并没有实现 materialize ，CombineExt 提供了开源的实现。

Combine In Resso

Resso 在很多场景使用到了 Combine，其中最经典的例子莫过于音效功能中多个属性的获取逻辑。音效需要使用专辑封面，专辑主题色以及歌曲对应的特效配置来驱动音效播放。这三个属性分别需要使用三个网络请求来获取，如果使用 iOS 中经典的闭包回调来编写这部分逻辑，那嵌套三个闭包，陷入回调地狱，更别提其中的错误分支很有可能遗漏。

func startEffectNormal() {
    // 1. 获取歌曲封面
    WebImageManager.shared.requestImage(trackCoverURL) { result in
        switch result {
        case .success(let image):
            // 2. 获取特效配置
            fetchVisualEffectConfig(for: trackID) { result in
                switch result {
                case .success(let path):
                    // 3. 获取封面主题色
                    fetchAlbumColor(trackID: trackID) { result in
                        switch result {
                        case .success(let albumColor):
                            self.startEffect(coverImage: coverImage, effectConfig: effectConfig, coverColor: coverColor)
                        case .failure:
                            // 处理获取封面颜色错误
                            break
                        }
                    }
                case .failure(let error):
                    // 处理获取特效配置错误
                    break
                }
            }
        case .failure(let error):
            // 处理下载图片错误
            break
        }
    }
}

使用 Combine，我们可以把三个请求封装成单独的 Publisher，再通过 combineLatest 将三个结果合并在一起进行使用：

func startEffect() {
    // 获取歌曲封面的 Publisher
    cancellable = fetchTrackCoverImagePublisher(for: trackCoverURL)
        // 并与 获取特效配置的 Publisher 和 获取专辑主题色的 Publisher 中的最新结果组成新的 Publisher
        .combineLatest(fetchVisualEffectPathPublisher(for: trackID), fetchAlbumColorPublisher(trackID: trackID))
        // 对最终的结果进行使用
        .sink { completion in
            if case .failure(let error) = completion {
                // 对错误进行处理
            }
        } receiveValue: { (coverImage, effectConfig, coverColor) in
            self.startEffect(coverImage: coverImage, effectConfig: effectConfig, coverColor: coverColor)
        }
}

这样的实现方式带来了很多好处：

代码结构更紧凑，可读性更好
错误处理更集中，不易遗漏
可维护性更好，后续如果需要新的请求，只需继续 combine 新的 Publisher 即可

此外，Resso 也对自己的网络库实现了 Combine 拓展，方便更多的同学开始使用 Combine：

func fetchSomeResource() -> RestfulClient<SomeResponse>.DataTaskPublisher{
    let request = SomeRequest()
    return RestfulClient<SomeResponse>(request: request)
            .dataTaskPublisher
}

总结

一言以蔽之，响应式编程的核心在于用声明的方式响应未来发生的事件流。在日常的开发中，合理地使用响应式编程可以大幅简化代码逻辑，但在不适宜的场景（甚至是所有场景）滥用则会让同事 🤬。常见的多重嵌套回调、自定义的通知都是非常适合切入使用的场景。

Combine 是响应式编程的一种具体实现，系统原生内置与优秀的实现让它相较于其他响应式框架有着诸多的优势，学习并掌握 Combine 是实践响应式编程的绝佳途径，对日常开发也有诸多毗益。

从响应式编程到 Combine 实践