Combine 中 subscribe(on:)/receive(on:) 的位置与 .share() 缓存

2026-11-04技术老齐

结论

用于 SwiftUI 更新的 .receive(on: DispatchQueue.main)，通常应放在 <2>，也就是合并输出之后、进入 UI 订阅者之前。框架内部的缓存访问、消息处理和状态维护，应统一放在串行 sessionQueue 上。

这里有两个问题需要分开处理：

.share() 让多个订阅者共享上游订阅，但不会向后来的订阅者回放最新值。
receive(on:) 和 subscribe(on:) 负责调度，不会自动保护缓存，也不会补发已经错过的消息。

把主队列切换放在 <1> 后恢复正常，说明执行时机或调用上下文影响了结果，但现有代码还不足以确定原因。补充描述中“订阅后能收到变化，订阅时没有当前值”的表现，符合 .share() 不回放历史值的语义，可以先从这里排查。

`subscribe(on:)` 和 `receive(on:)` 分别影响什么

操作	作用	不会保证的事情
`subscribe(on:)`	调度向上游订阅、请求 demand 和取消的操作	不会强制 WebSocket 的异步回调在该队列执行
`receive(on:)`	将经过该位置的值和 completion 交给指定 scheduler，再向下游传递	不会改变此前已经执行的普通函数调用所在的队列，也不控制 `receive(subscription:)` 的调度

Apple 的 subscribe(on:options:) 和 receive(on:options:) 文档说明了这些调度范围。

例如：

let publisher = try makePublisher()

publisher
    .subscribe(on: sessionQueue)
    .receive(on: DispatchQueue.main)
    .sink(
        receiveCompletion: { print($0) },
        receiveValue: { print($0) }
    )
    .store(in: &observers)

makePublisher() 在第一行就执行了。如果它立即读取缓存或发送请求，后面添加的两个 operator 都不会改变这次调用的执行位置。

如果需要把构造推迟到订阅时，可以使用 Deferred。构造函数会抛错时，还需要在闭包内将错误转换为 Fail。即使推迟了构造，底层请求实现仍须保证监听与发送的顺序，不能据此认定响应监听已经就绪。

`<1>` 和 `<2>` 的区别

放在 <1>：

.receive(on: DispatchQueue.main)
.flatMap { ... }

后面的 flatMap 闭包会在主队列处理连接输出，闭包中的缓存读取和 Publisher 构造也会随之在主队列执行。不过，内部 Publisher 后续从其他队列发出的值，仍可能在其他队列到达。

放在 <2>：

.flatMap { ... }
.receive(on: DispatchQueue.main)
.sink(...)

合并后的输出会交给主队列，适合放在 UI 更新之前。Apple 的事件处理示例也采用了这种做法。

标准 Combine 没有内置 tryFlatMap。如果它不是项目中的扩展，可以在 flatMap 闭包内用 do/catch 返回 Fail。tryMap 配合 switchToLatest() 也能处理返回 Publisher 的抛错调用，但会取消先前的内部订阅，使用前需要确认这符合连接事件的语义。

缓存必须在同一个隔离范围内完成查找和创建

原代码中的：

.receive(on: sessionQueue)
.handleEvents(receiveCompletion: { ... })

只能约束经过这里的 completion 回调，不会让 <3> 的缓存查询和 <4> 的缓存写入自动进入 sessionQueue。

应把查询缓存、创建实例和写入缓存的整个过程放进同一个串行队列。否则，两个调用者可能同时读到 nil，各自创建一个 .share()。

下面的代码保留 .share() 来处理缓存隔离，仍不提供最新值回放。代码沿用已有的项目类型和成员，前提是请求 Publisher 在订阅时才启动请求。

public func studioModeStatePublisher() -> AnyPublisher<Bool, Error> {
    Just(())
        .setFailureType(to: Error.self)
        .receive(on: sessionQueue)
        .flatMap { [weak self] _ -> AnyPublisher<Bool, Error> in
            guard let self = self else {
                return Empty<Bool, Error>().eraseToAnyPublisher()
            }

            do {
                return try self.studioModeStateOnQueue()
            } catch {
                return Fail<Bool, Error>(error: error)
                    .eraseToAnyPublisher()
            }
        }
        .eraseToAnyPublisher()
}

private func studioModeStateOnQueue() throws
    -> AnyPublisher<Bool, Error> {
    dispatchPrecondition(condition: .onQueue(sessionQueue))

    if let cached = publishers.studioModeState {
        return cached
    }

    let initial = try getStudioModeStateOnce()
    let changes = try listenForEvent(
        OBSEvents.StudioModeStateChanged.self,
        firstOnly: false
    )
    .map(\.studioModeEnabled)

    let shared = initial
        .merge(with: changes)
        .receive(on: sessionQueue)
        .removeDuplicates()
        .handleEvents(receiveCompletion: { [weak self] _ in
            self?.publishers.studioModeState = nil
        })
        .share()
        .eraseToAnyPublisher()

    publishers.studioModeState = shared
    return shared
}

外部方法改为通过 Publisher 报告构造错误，不再同步 throws。UI 端可以这样订阅：

try connectToOBS()
    .flatMap { _ in
        self.studioModeStatePublisher()
    }
    .receive(on: DispatchQueue.main)
    .sink(
        receiveCompletion: { print("Completion:", $0) },
        receiveValue: { enabled in
            // 更新 UI 模型
            print("Studio mode:", enabled)
        }
    )
    .store(in: &observers)

其他缓存访问入口也必须遵守同一规则，包括断线清理、重连失效和按键删除。串行队列只保护提交给它的操作，其他线程直接访问属性时仍不受保护。

如果允许主动替换缓存，还应检查条目标识或连接代次，避免旧 Publisher 的 completion 清掉新条目。

`<5>` 可以切换输出队列，但发送顺序要在底层保证

在 <5> 添加：

.receive(on: sessionQueue)

后续解析、状态更新等操作就可以在 sessionQueue 上处理，但前面 wsPublisher.send(...) 调用已经产生的副作用不会因此改变。

需要检查 send 的具体契约：

调用方法时就发送消息，还是等订阅建立后才发送？
响应过滤订阅和 request ID 登记，是否在发送之前完成？
响应很快返回时，是否可能还没有订阅者接收？
取消、失败和断线时，能否正确移除响应登记？

如果底层使用 PassthroughSubject，监听建立之前发送的响应不会自动保留。调整调度位置，可能只是暂时改变了这一竞争条件的触发概率。

subscribe(on: sessionQueue) 适用于确实需要在该队列执行的订阅工作。WebSocket 回调队列和实际发送队列，仍需要由底层实现明确安排。

`.share()`、缓存实例和缓存最新值是三件事

重复请求与 Publisher 是否为 struct 没有直接对应关系。需要检查每次订阅是否会启动新的工作，以及调用者是否拿到了同一个共享实例。

share() 返回的 Publishers.Share 本身是引用类型，行为相当于结合 multicast、PassthroughSubject 和自动连接。它不会回放先前的输出。

例如：

A 订阅，初始请求返回 true。
B 随后订阅同一个共享实例。
OBS 一直没有变化。

此时 B 收不到 true 是预期行为：上游仍在运行，但没有新的值供 B 接收。removeDuplicates() 也不会为新订阅者补发当前状态。

对于 studio mode 这样的状态接口，可以选择：

使用经过验证、回放容量为 1 的共享实现，并明确连接和缓存的生命周期。
由会话对象持续订阅请求与事件流，把最新状态存进 CurrentValueSubject，再向各个页面提供状态 Publisher。

CurrentValueSubject 会保留最近发布的一个元素。尚未取得初始值时，宜使用 Bool? 或明确的 .unknown 状态，避免用 false 表示“还不知道”。

使用会话存储状态时，还需要确定：

没有 UI 订阅者时，会话是否继续监听事件？如果停止监听，保留的值就可能过期。
断线后是否将状态标记为未知或失效？
将错误作为状态发布，还是让流终止？Subject 一旦收到 completion，就不能继续承载重连后的更新。
谁持有上游订阅，何时取消？只保存 Publisher 并不代表已经建立并持续持有订阅。

在共享实例前放置 receive(on:)，也不能保证自定义回放值的交付队列。向后来订阅者回放值的过程可能直接发生在订阅调用中，因此 UI 端仍应在回放 operator 后面添加 .receive(on: DispatchQueue.main)。

自定义 `shareReplay` 为什么可能在测试中正常、在 SwiftUI 中丢值

引用的 ReplaySubject 示例实现中，订阅建立时会立即遍历缓存，但只有当前 demand 大于零才会发送值。后续 request(_:) 只增加 demand，不会重新发送此前未交付的缓存。

这可能导致以下情况：

订阅者收到 Subscription，此时尚未请求 demand。
Subject 尝试回放，缓存值因 demand 为零而被跳过。
订阅者稍后请求 demand。
未交付的缓存没有再次发送，订阅者只能等待新事件。

直接使用 sink 时，它会请求无限 demand，因此这种实现容易通过简单测试。SwiftUI 的实际订阅链是否会触发上述过程，需要通过日志或独立测试确认，不能直接断言所有 SwiftUI 订阅者都会延迟 demand。

可以先做一个确定性的测试：让自定义 Subscriber 保存 Subscription，暂不请求 demand；确认 Subject 已尝试回放后，再调用 request(.max(1))。只要上游仍在活动、缓存中已有值，回放实现就应在收到 demand 后交付该值。

完整实现还需要正确处理并发的 request、cancel、新值发送和 completion，以及订阅者的移除。仅调整 .share() 的调用位置，无法修复这些契约问题。

按数据经过的位置排查

先去掉测试中的 .output(in: 0..<4)。它在收满四个输出后会结束下游订阅，而取消不保证触发上游的 receiveCompletion。因此，只在收到 completion 时清理缓存，还不能完整管理活动订阅。

然后在各处分别记录订阅、输出、completion、取消和 demand：

.handleEvents(
    receiveSubscription: { _ in print("state: subscribed") },
    receiveOutput: { print("state: output", $0) },
    receiveCompletion: { print("state: completion", $0) },
    receiveCancel: { print("state: cancelled") },
    receiveRequest: { print("state: demand", $0) }
)

记录位置应覆盖初始请求、事件监听、共享或回放之前，以及 UI 订阅之前。日志同时带上连接代次、缓存条目标识和 request ID，便于确认两边使用的是同一个会话和共享实例。

根据日志中的现象继续排查：

回放前有值、回放后没有值：检查回放缓冲与 demand。
收到订阅后立即取消：检查 AnyCancellable、ViewModel 和会话对象的生命周期。
框架有值、UI 端没有值：检查两边是否使用同一个实例，以及 UI 端的过滤和取消。
初始请求没有输出、事件流正常：检查请求发送与响应监听的先后顺序。
UI sink 已收到值、界面没有变化：检查模型持有关系和 UI 属性更新位置。

SwiftUI 中应避免在反复求值的 body 内创建会话或订阅。使用 ObservableObject 模式时，由视图创建并拥有的模型通常使用 @StateObject；外部传入的模型则按实际所有权使用 @ObservedObject 或环境注入。

merge 还有一个独立的状态一致性问题：它按到达顺序转发两个来源，不保证初始快照先到、后续变化再到。如果新事件先到、较旧的请求结果后到，旧状态仍可能覆盖新状态。应先确保事件监听就绪，再结合协议提供的顺序信息、版本或明确的缓冲策略，协调快照与事件。共享和回放都会保留最终收到的值，无法判断哪个值在业务上更新。

Combine 中 subscribe(on:)/receive(on:) 的位置与 .share() 缓存

结论

subscribe(on:) 和 receive(on:) 分别影响什么

<1> 和 <2> 的区别