axiom-combine-patterns

Combine 模式

概述

Combine 仍广泛应用于大型生产代码库中——UIKit 代理、NotificationCenter 桥接、KVO 观察和 @Published 属性随处可见。新代码更倾向于使用 async/await，但互操作以及维护现有 Combine 管道仍是日常工作。本技能涵盖重要的决策和常见陷阱：何时使用 Combine 与 async/await、如何避免内存泄漏，以及如何在这两种范式之间进行桥接。

核心原则：Combine 并没有过时——它已经成熟。问题不是“我该不该使用 Combine？”，而是“对于这个特定的数据流，Combine 是合适的工具吗？”

何时使用本技能

• 处理现有的 Combine 管道
• 为新的数据流选择 Combine 还是 async/await
• 调试 AnyCancellable 内存泄漏或管道静默失效的问题
• 使用 @Published 或 ObservableObject
• 将 Combine 发布者与 async/await 代码桥接
• 使用 Subject（PassthroughSubject、CurrentValueSubject）

何时不使用本技能

• Timer.publish 模式 → 通过 axiom-ios-concurrency 转到 timer-patterns 技能（专门涵盖计时器生命周期）
• 从 ObservableObject 迁移到 @Observable → 使用 axiom-swift-concurrency（现代观察机制）
• UIKit ↔ SwiftUI 桥接 → 转到 axiom-ios-ui（视图封装，而非数据流）
• 通用 async/await 模式 → 使用 axiom-swift-concurrency

示例提示词

• “这个场景应该用 Combine 还是 async/await？”
• “我的 Combine 管道会悄无声息地停止产生值”
• “如何将发布者转换为异步序列？”
• “AnyCancellable 泄漏了——我应该把它存在哪里？”
• “combineLatest 和 zip 有什么区别？”
• “如何使用 Combine 对文本字段输入进行防抖？”
• “我的 @Published 属性更新没有传递到视图”
• “如何将 Combine 发布者桥接到 async/await 代码中？”

---

第 1 部分：Combine 与 async/await 决策树

使用场景	Combine	async/await	原因
一次性网络请求	否	是	async/await 更简单，无需管理可取消对象
随时间产生值的流	是	AsyncStream	Combine 的操作符（debounce、combineLatest）更丰富
对用户输入进行防抖/节流	是	不太方便	Combine 内置了 debounce/throttle；AsyncStream 需要手动实现
合并多个来源	是	TaskGroup	Combine 的 merge/combineLatest 可自然处理异构流
现有 UIKit KVO/Notification	是	桥接	publisher(for:) 和 NotificationCenter.default.publisher 是符合惯用方式的 Combine 用法
新建 iOS 17+ 项目	否	是	@Observable + async/await 是现代模式
使用 Combine 的现有代码库	维护	逐步迁移	不要重写可正常工作的管道——在边界处进行桥接

快速决策

Is it a one-shot operation (network call, file read)?
├─ Yes → async/await (simpler, no cancellable management)
│
Does it need time-based operators (debounce, throttle, delay)?
├─ Yes → Combine (built-in operators, no manual implementation)
│
Are you combining multiple ongoing streams?
├─ Yes → Combine (combineLatest, merge, zip are purpose-built)
│
Is this new code on iOS 17+?
├─ Yes → async/await + @Observable (modern pattern)
│
Is it existing Combine code that works?
└─ Yes → Keep it. Bridge at boundaries when async/await code needs the data.

---

第 2 部分：发布者/订阅者生命周期

AnyCancellable 存储规则

AnyCancellable 在释放时会取消其订阅。如果不存储它，管道会在设置完成后立即取消。

❌ 管道立即终止

func setupPipeline() {
    publisher
        .sink { value in
            self.handle(value)  // Never called
        }
    // AnyCancellable returned by sink is discarded → subscription cancelled
}

✅ 存储到 Set 中

private var cancellables = Set<AnyCancellable>()

func setupPipeline() {
    publisher
        .sink { [weak self] value in
            self?.handle(value)
        }
        .store(in: &cancellables)
}

为什么选择 Set，而不是 Array

Set<AnyCancellable> 是符合惯用方式的选择，原因如下：

• store(in:) 既可用于 Set，也可用于 RangeReplaceableCollection（包括 Array），但通常使用 Set
• 订阅的顺序无关紧要
• 如果重复执行设置，可避免意外重复

4 种内存泄漏模式

泄漏 1：在 sink 中强引用 self

// ❌ LEAK: sink closure captures self strongly
publisher
    .sink { value in
        self.handle(value)  // Strong capture → retain cycle
    }
    .store(in: &cancellables)

// ✅ FIX: weak self
publisher
    .sink { [weak self] value in
        self?.handle(value)
    }
    .store(in: &cancellables)

泄漏 2：缺少 store(in:)

// ❌ LEAK: cancellable assigned to local var, not stored
let cancellable = publisher.sink { handle($0) }
// cancellable deallocated at end of scope → pipeline cancelled

// ✅ FIX: store in instance property
publisher.sink { [weak self] in self?.handle($0) }
    .store(in: &cancellables)

泄漏 3：过度保留可取消对象

// ❌ LEAK: cancellables set never cleared, old pipelines accumulate
func refreshData() {
    // Each call adds another subscription without removing the previous one
    dataPublisher
        .sink { [weak self] in self?.update($0) }
        .store(in: &cancellables)
}

// ✅ FIX: clear before re-subscribing
func refreshData() {
    cancellables.removeAll()  // Cancel previous subscriptions
    dataPublisher
        .sink { [weak self] in self?.update($0) }
        .store(in: &cancellables)
}

泄漏 4：assign(to:on:) 强引用捕获

assign(to:on:) 会对 on: 参数进行强引用。当目标是 self 时，就会产生循环引用：self → cancellables → subscription → self。

// ❌ LEAK: assign(to:on:) retains self strongly — deinit never called
userPublisher
    .map { $0.name }
    .assign(to: \.displayName, on: self)
    .store(in: &cancellables)

// ✅ FIX: use assign(to:) with @Published projected value (iOS 14+)
userPublisher
    .map { $0.name }
    .assign(to: &$displayName)
// No store(in:) needed — subscription tied to @Published property lifetime

关键区别：assign(to: &$prop) 不会返回 AnyCancellable——订阅由内部管理，并会在 @Published 属性的所有者释放时取消。不会产生循环引用，也不需要存储可取消对象。

如果必须支持 iOS 13，请改用带有 [weak self] 的 sink。

---

第 3 部分：必备运算符

每组提供一个经典示例。这些示例涵盖了 90% 的实际用法。

转换

// map: transform each value
publisher.map { $0.name }

// compactMap: transform + filter nil
publisher.compactMap { Int($0) }

// flatMap: one-to-many (each value produces a new publisher)
searchText
    .flatMap { query in
        api.search(query)  // Returns a publisher
    }

flatMap 的注意事项：如果没有 .switchToLatest() 或 maxPublishers: .max(1)，flatMap 会为每个上游值创建一个新的内部发布者。对于边输入边搜索的场景，请改用 map + switchToLatest：

searchText
    .map { query in api.search(query) }
    .switchToLatest()  // Cancels previous search when new query arrives

合并多个来源

// combineLatest: latest value from each, fires when ANY changes
Publishers.CombineLatest(namePublisher, agePublisher)
    .map { name, age in "\(name), \(age)" }

// merge: interleave values from same-type publishers
Publishers.Merge(localUpdates, remoteUpdates)
    .sink { update in handle(update) }

// zip: pairs values 1:1 (waits for both to produce)
Publishers.Zip(requestA, requestB)
    .sink { responseA, responseB in /* both complete */ }

运算符	触发时机	使用场景
combineLatest	任意输入发生变化	表单验证（所有字段）
merge	任意输入产生值	合并事件流
zip	所有输入都产生一个值	必须一起完成的并行请求

基于时间的运算符

// debounce: wait until values stop arriving (search-as-you-type)
searchTextPublisher
    .debounce(for: .milliseconds(300), scheduler: RunLoop.main)
    .sink { [weak self] query in self?.search(query) }
    .store(in: &cancellables)

// throttle: emit at most once per interval (scroll position)
scrollOffsetPublisher
    .throttle(for: .milliseconds(100), scheduler: RunLoop.main, latest: true)
    .sink { [weak self] offset in self?.updateHeader(offset) }
    .store(in: &cancellables)

运算符	行为	使用场景
debounce	等待输入静默，然后发送最后一个值	搜索字段、自动保存
throttle(latest: true)	按固定间隔发送最新值	滚动跟踪、传感器数据
throttle(latest: false)	按固定间隔发送第一个值	限制按钮点击频率

错误处理

// tryMap: transform that can throw
publisher.tryMap { data in
    try JSONDecoder().decode(Model.self, from: data)
}

// mapError: convert error types
publisher.mapError { error in
    AppError.network(error)
}

// replaceError: provide fallback value (terminates error path)
publisher.replaceError(with: defaultValue)

// retry: re-subscribe on failure
publisher.retry(3)  // Retry up to 3 times before propagating error

错误处理的顺序很重要：retry 应放在 replaceError 之前。retry 会重新订阅上游发布者；replaceError 会终止错误，并使管道不再产生错误。

api.fetchData()
    .retry(3)                    // Try 3 more times on failure
    .replaceError(with: cached)  // If all retries fail, use cache
    .sink { data in update(data) }
    .store(in: &cancellables)

在 flatMap 之后使用 replaceError 会终止外层管道：如果 replaceError 位于 flatMap 下游，单个内部发布者出错就会终止整个管道，而不只是该请求。应将错误处理移入 flatMap，让每个内部发布者自行处理错误：

// ❌ One API error kills the entire pipeline
$searchText
    .flatMap { query in api.search(query) }
    .replaceError(with: [])  // Pipeline completes on first error
    .sink { ... }

// ✅ Each search handles its own errors independently
$searchText
    .flatMap { query in
        api.search(query)
            .replaceError(with: [])  // Only this search affected
    }
    .sink { ... }

---

第 4 部分：@Published + ObservableObject

willSet 时机

@Published 会在 willSet 中触发其发布者，而不是在 didSet 中。这意味着订阅者会在属性实际设置到对象上之前看到新值。

class ViewModel: ObservableObject {
    @Published var count = 0

    init() {
        $count.sink { newValue in
            // 'newValue' is the incoming value
            // BUT self.count is still the OLD value here
            print("New: \(newValue), Current: \(self.count)")
            // Prints "New: 1, Current: 0" when count is set to 1
        }
        .store(in: &cancellables)
    }
}

如果需要在属性设置后读取其值，请不要订阅 $count——改用 didSet 观察器，或短暂延迟后读取 self.count。$ 发布者用于响应即将传入的值，而不是读取变更后的状态。

嵌套 ObservableObject 陷阱

SwiftUI 不会观察嵌套的 ObservableObject 变更。只有顶层对象的 objectWillChange 会触发视图更新。

// ❌ View won't update when settings.theme changes
class AppState: ObservableObject {
    @Published var settings = Settings()  // Settings is also ObservableObject
}

class Settings: ObservableObject {
    @Published var theme = "light"  // Changes here don't propagate
}

// ✅ FIX: Forward objectWillChange manually
class AppState: ObservableObject {
    @Published var settings = Settings()
    private var cancellables = Set<AnyCancellable>()

    init() {
        settings.objectWillChange
            .sink { [weak self] _ in
                self?.objectWillChange.send()
            }
            .store(in: &cancellables)
    }
}

iOS 17+ 的更佳修复方案：迁移到 @Observable，它会自动处理嵌套观察。迁移模式请参阅 axiom-swift-concurrency。

线程安全警告

@Published 不是线程安全的。在后台线程设置 @Published 属性会导致 objectWillChange 在主线程之外触发，这可能导致 SwiftUI 视图崩溃。

// ❌ CRASH: @Published set from background thread
class ViewModel: ObservableObject {
    @Published var data: [Item] = []

    func fetch() {
        Task {
            let items = await api.fetchItems()
            data = items  // Background thread → crash
        }
    }
}

// ✅ FIX: Ensure main thread
@MainActor
class ViewModel: ObservableObject {
    @Published var data: [Item] = []

    func fetch() {
        Task {
            let items = await api.fetchItems()
            data = items  // Safe — @MainActor ensures main thread
        }
    }
}

---

第 5 部分：桥接 Combine 和 async/await

Publisher → AsyncSequence

使用 .values 将任意发布者作为异步序列使用：

let cancellable = notificationPublisher
    .sink { notification in handle(notification) }

// ✅ Modern equivalent using .values
for await notification in notificationPublisher.values {
    handle(notification)
}

使用 .values 时的注意事项：

• for await 循环会无限运行，直到发布者完成或 Task 被取消
• 发布者抛出的错误会终止循环
• 只能有一个使用者——如果两个 for await 循环消费同一个 .values，行为未定义

async/await → Publisher

将异步函数封装在 Future 中，以便 Combine 使用：

func fetchUser(id: String) async throws -> User { ... }

// Wrap as a Combine publisher
let userPublisher = Future<User, Error> { promise in
    Task {
        do {
            let user = try await fetchUser(id: "123")
            promise(.success(user))
        } catch {
            promise(.failure(error))
        }
    }
}

Future 会立即执行——它会在创建时运行闭包，而不是在订阅时运行。如果需要延迟执行，请用 Deferred 封装：

let lazyPublisher = Deferred {
    Future<User, Error> { promise in
        Task {
            do {
                let user = try await fetchUser(id: "123")
                promise(.success(user))
            } catch {
                promise(.failure(error))
            }
        }
    }
}

渐进式迁移策略

不要重写能正常工作的 Combine 代码。在边界处进行桥接：

Combine pipeline  →  .values  →  async/await code
                     (bridge)

async function    →  Future   →  Combine pipeline
                     (bridge)

迁移优先级：

1. 新代码：使用 async/await 编写
2. 边界：使用 .values 或 Future 进行桥接
3. 现有 Combine 代码：保留能正常工作的管道
4. 重写：仅当管道本就需要进行重大更改时

---

第 6 部分：Subject

PassthroughSubject 与 CurrentValueSubject

特性	PassthroughSubject	CurrentValueSubject
初始值	无	必须提供
较晚订阅的订阅者	错过之前的值	立即获取当前值
.value 属性	否	是（读取当前值）
使用场景	事件（按钮点击、通知）	状态（当前选择、加载状态）

// Event-driven: no initial value, late subscribers miss past events
let taps = PassthroughSubject<Void, Never>()
taps.send()

// State-driven: always has a current value
let isLoading = CurrentValueSubject<Bool, Never>(false)
isLoading.value = true  // Direct access
isLoading.send(false)   // Also works

完成后发送的陷阱

Subject 收到完成事件后，之后所有的 send() 调用都会被静默忽略。不会崩溃，也不会报错——一片静默。

let subject = PassthroughSubject<Int, Never>()

subject.send(1)           // Delivered
subject.send(completion: .finished)
subject.send(2)           // Silently ignored — no crash, no warning

// This is the most common cause of "my pipeline stopped working"

诊断：如果管道不再产生值且没有任何提示，请检查上游是否发送了 .finished 或 .failure 完成事件。完成后，管道就失效了。

---

第 7 部分：冷发布者与热发布者（share/multicast）

大多数 Combine 发布者都是冷发布者——它们在收到订阅后才开始工作，并且每个订阅者都有各自独立的执行过程。每个订阅者都会触发 URLSession.dataTaskPublisher 发起新的 HTTP 请求。

// ❌ Two subscribers = two network requests
let publisher = URLSession.shared
    .dataTaskPublisher(for: url)
    .map(\.data)
    .eraseToAnyPublisher()

publisher.sink { cache.store($0) }.store(in: &cancellables)  // Request 1
publisher.sink { display($0) }.store(in: &cancellables)       // Request 2

share()

.share() 会将冷发布者变为热发布者——第一个订阅者触发工作，后续订阅者共享输出：

// ✅ One request, shared result
let publisher = URLSession.shared
    .dataTaskPublisher(for: url)
    .map(\.data)
    .share()
    .eraseToAnyPublisher()

publisher.sink { cache.store($0) }.store(in: &cancellables)  // Triggers request
publisher.sink { display($0) }.store(in: &cancellables)       // Shares result

share() 的注意事项

注意事项	影响	解决方法
较晚订阅的订阅者会错过值	share() 使用 PassthroughSubject，不会重放	在第一个值到达前添加所有订阅者，或使用带有 CurrentValueSubject 的 multicast
上游在订阅者订阅前已完成	较晚订阅的订阅者会立即收到 .finished，且没有任何值	确保订阅顺序正确，或在 Combine 之外缓存结果
所有订阅者都取消订阅 → 上游取消	此后新订阅者会触发一次新的上游执行	这是预期行为，但如果你以为结果已缓存，就会感到意外

何时使用 share()

Multiple subscribers to the same expensive publisher?
├─ No → Don't use share() (unnecessary complexity)
│
├─ Yes, all subscribe at the same time?
│   └─ Yes → share() works
│
└─ Yes, subscribers attach at different times?
    └─ Use multicast(subject:) with CurrentValueSubject, or cache the result in a property

---

避免自我合理化

想法	现实
“Combine 已经过时了，直接用 async/await 吧”	Combine 没有弃用通知。数千个生产环境应用都在使用它。重写能正常工作的管道既浪费时间，又会引入 bug。应改为渐进式桥接。
“我到处用 .sink 就行了”	如果没有 [weak self] 和正确的 store(in:)，每个 sink 都可能造成内存泄漏。第 2 部分中的生命周期规则可以避免最常见的 4 种泄漏模式。
“assign(to:on:) 没问题，这是标准 API”	它会强引用 on:，如果目标是 self，就会形成循环引用。改用 assign(to: &$prop)（第 2 部分，泄漏 4）。
“debounce 和 throttle 是一样的”	debounce 会等待静默期；throttle 会按间隔发出值。用错会导致响应延迟或事件丢失。第 3 部分有决策表。
“我知道 @Published 的工作方式”	@Published 在 willSet 时触发，而不是 didSet。嵌套的 ObservableObject 不会传播。后台线程访问会崩溃。第 4 部分涵盖了这三个陷阱。
“我要一次性把所有内容都迁移到 async/await”	全面重写能正常工作的 Combine 代码会引入 bug、浪费时间。应在边界处进行桥接（第 5 部分）。仅当管道本就需要进行重大更改时才重写。

---

压力场景

场景 1：“把所有 Combine 代码迁移到 async/await”

设置：技术负责人想要现代化代码库。“Combine 已经过时了，我们把它彻底移除吧。”

压力：权威压力 + 范围蔓延。整个数据层都使用 Combine 发布者、@Published 属性和运算符链。

使用此技能时的预期做法：提出渐进式迁移策略（第 5 部分）。新代码使用 async/await。在边界处使用 .values 和 Future。现有且正常工作的流水线保持不变，直到需要修改时再处理。全面重写是成本最高、收益最低的选项。

应对模板：“Combine 并未弃用——Apple 仍在每个 SDK 中提供它。全面重写正常工作的流水线会引入我们目前没有的错误。我们可以在边界处进行桥接：新代码使用 async/await，通过 .values 使用现有发布者，并且只在对某条流水线进行重大修改时才重写它。”

---

场景 2：“流水线悄无声息地停止了——直接重新创建吧”

设置：一次重构后，Combine 流水线不再产生值。没有崩溃，也没有错误。

压力：时间压力。“直接拆掉重建吧。”

使用此技能时的预期做法：先诊断，再重建。检查：(1) 上游是否发送了完成事件？（完成后仍发送事件，第 6 部分）。(2) AnyCancellable 是否仍然存活？（存储规则，第 2 部分）。(3) 发布者是否在没有错误处理的情况下出错？（replaceError / catch，第 3 部分）。这三种原因涵盖了 90% 的流水线静默故障。

诊断清单：

1. AnyCancellable 是否仍被存储？（集合未清空，且未被释放）
2. 上游是否发送了 .finished 或 .failure？
3. 是否存在没有错误处理的 tryMap 或其他可能抛出错误的运算符？
4. 外层发布者已完成的情况下，是否使用了 switchToLatest？

应对模板：“重建之前，我先检查四件事：可取消对象的生命周期、上游完成事件、未处理的错误，以及 switchToLatest 的完成状态。这些通常就是问题所在。诊断只需 5 分钟，而重建和测试需要 30 分钟。”

---

场景 3：“设置更改后，UI 没有更新”

设置：设置界面使用了一个嵌套的 ObservableObject。父级 AppState 持有一个 Settings 对象。当用户更改 settings.theme 时，UI 没有更新。

压力：“单独使用时绑定没问题，这肯定是 SwiftUI 的 bug。我直接用 objectWillChange.send() 强制刷新吧。”

使用此技能时的预期做法：识别嵌套 ObservableObject 陷阱（第 4 部分）。SwiftUI 不会观察嵌套的 ObservableObject——只有顶层对象的 objectWillChange 才会触发视图更新。解决方法是将嵌套对象的 objectWillChange 转发出去，或者迁移到 @Observable（iOS 17+），它会自动处理嵌套情况。

不具备此技能时的反模式：在代码中到处添加 objectWillChange.send() 调用；给每个嵌套属性添加 @Published（这并没有帮助）；或者重新组织模型，把所有内容都扁平化到一个对象中（失去关注点分离）。

应对模板：“SwiftUI 只会观察顶层 ObservableObject。嵌套对象需要将其 objectWillChange 转发给父对象。第 4 部分有确切的实现模式——只需在父对象的 init 中添加 5 行代码，而不是修复 SwiftUI 的 bug。”

---

资源

WWDC：2019-722、2019-721、2020-10034

文档：/combine、/combine/anycancellable、/combine/published

技能：swift-concurrency、memory-debugging