Swift AsyncStream 背压与流控工程指南

AsyncStream 很容易被当成“异步版本回调”，写起来快，坑也来得快。生产者持续 yield，消费者偶尔卡住，短时间看不到问题，几分钟后内存开始爬升，几十分钟后掉帧、超时、甚至 OOM。很多团队把这归因于“设备差”或“网络波动”，但真正根因通常是没有背压协议。

在流式系统里，背压不是优化项，而是生存条件。只要生产速度可能超过消费速度，你就必须明确回答三件事：队列多大、满了怎么办、谁负责降速。

一、`AsyncStream` 的运行语义：先搞清楚再谈策略

AsyncStream 是生产者与消费者之间的桥。生产者通过 continuation 推数据，消费者用 for await 拉数据。默认情况下如果你不设置策略，系统不会替你做业务级流控决策。

核心事实：

yield 成功不代表消费者已处理，只代表元素进入了缓冲流程。
缓冲区是资源，不是无限吞吐器。
如果不显式定义满队列行为，就会把风险留给运行时。

AsyncStream.BufferingPolicy 给了三个方向：

unbounded：几乎等于“无上限风险自担”。
bufferingNewest(n)：保留最新 n 条，牺牲历史。
bufferingOldest(n)：保留最早 n 条，牺牲新鲜度。

这不是语法选择，而是业务选择。

二、先画一张流水线图：速率失配在哪里发生

flowchart LR
    S[事件源: Socket/传感器/日志] --> P[Producer Task]
    P --> B{AsyncStream Buffer}
    B --> C[Consumer Task]
    C --> D[Decode + Validate]
    D --> E[业务处理]
    E --> F[存储/渲染]

    B -->|队列积压| M[内存上涨]
    C -->|处理变慢| L[端到端延迟上升]

这张图要落到具体指标，否则只是概念图。

推荐最小指标集合：

生产速率（events/s）。
消费速率（events/s）。
缓冲区当前深度与峰值。
丢弃率（按策略）。
端到端延迟 P95/P99。

三、背压策略选型：先按业务分类

1）实时类（行情、位置、传感器）

目标是“新鲜度优先”。通常用 bufferingNewest(n)，允许丢旧数据，保证用户看到的是最新状态。

2）审计类（订单事件、计费流水）

目标是“不丢数据”。不能靠 AsyncStream 内存缓冲兜底，必须落盘或消息队列持久化，再由消费者拉取。

3）混合类（聊天消息 + 在线状态）

同一通道里不同事件类型诉求不同。建议拆流：

状态流可丢旧，保新。
消息流不可丢，持久化。

流不拆，背压策略一定互相牵制。

四、工程实现：别让生产者“盲推”

推荐把 continuation 包成受控组件，而不是裸露到各层：

actor StreamGate<Element: Sendable> {
    private let continuation: AsyncStream<Element>.Continuation
    private var dropped = 0

    init(_ continuation: AsyncStream<Element>.Continuation) {
        self.continuation = continuation
    }

    func push(_ element: Element) {
        let result = continuation.yield(element)
        if case .dropped = result { dropped += 1 }
    }

    func finish() {
        continuation.finish()
    }

    func droppedCount() -> Int { dropped }
}

关键点：

统计必须内建，不要靠外层猜。
推送入口要单一，便于限流和熔断。
终止信号必须明确，避免消费者永久等待。

五、风险面与防线

风险 1：无上限缓冲导致内存失控

防线：禁止 unbounded 直上生产。若业务确需无损，必须持久化队列，不把内存当数据库。

风险 2：消费者抖动触发级联延迟

防线：把消费链拆阶段，在重处理前做快速过滤；对慢依赖设置独立并发池和 timeout。

风险 3：丢弃策略与业务语义冲突

防线：在事件 schema 层增加“可丢弃等级”，并在流水线里显式分流。

风险 4：取消后仍继续推送

防线：在生产者循环里检查取消状态，收到取消后立刻停止并 finish()。

六、性能调优：先控速，再扩容

很多团队第一反应是“加消费者并发”。这常常只会加剧资源争用。

更稳定的顺序：

优先降生产速率（采样、聚合、去抖）。
压缩单条事件处理成本（减少解码和复制）。
再评估是否提高消费者并发。

建议优化清单：

对高频事件做批量提交，减少函数边界与分配开销。
使用值语义快照，避免跨任务共享可变引用。
把日志采集与主链路解耦，避免日志反向拖慢消费。
用 Instruments 检查分配风暴和异步调度开销。

七、测试设计：背压必须被“证明”

1）速率失配压测

构造生产 10k/s、消费 2k/s 场景，验证内存不会无限上涨，系统会触发预期丢弃或限流。

2）策略正确性测试

验证 bufferingNewest 与 bufferingOldest 在业务语义上的输出是否符合预期，而不是只看程序不崩。

3）取消恢复测试

取消后检查：

生产者停止推送。
消费者正确退出。
资源句柄释放。

4）长时稳定测试

持续运行数小时，观察缓冲深度、丢弃率和尾延迟是否稳定在预算内。

八、可观测方案：把“慢”变成可解释事件

推荐每条流都输出统一指标：

stream_produced_total
stream_consumed_total
stream_dropped_total
stream_buffer_depth
stream_latency_ms

并给关键阶段打 signpost，形成时间线证据。否则你只能看到“卡”，看不到“卡在何处”。

九、排障流程：遇到堆积先做什么

flowchart TD
    A[告警: 延迟或内存上升] --> B[核对生产速率与消费速率]
    B --> C{生产是否长期高于消费}
    C -- 是 --> D[启用限流/采样/批处理]
    C -- 否 --> E[检查消费者阶段耗时分布]
    E --> F{是否单阶段过慢}
    F -- 是 --> G[拆分阶段并优化热点]
    F -- 否 --> H[检查缓冲策略与丢弃语义]
    D --> I[回归压测与阈值校验]
    G --> I
    H --> I

这个流程能避免“盲目扩容”导致的次生问题。

十、落地案例模板（可直接复用）

假设你在做实时行情页面：

Socket 事件源每秒 2000 条。
UI 每帧只需最新快照。
策略选 bufferingNewest(256)。
每 50ms 聚合一次再渲染。
页面退出时父任务取消，流立刻 finish。

效果通常是：内存稳定、延迟可控、用户体验更平滑。

十一、治理清单

所有 AsyncStream 必须标注缓冲策略。
所有流必须定义“满队列行为”。
高价值事件必须有无损通道，不允许仅靠内存流。
取消路径必须覆盖生产者与消费者。
每次改动必须跑速率失配压测。

十二、结语

AsyncStream 不是“更优雅的回调”，它是流式系统接口。接口背后没有背压协议，系统就会把风险转换成内存和延迟债务。把策略、指标、测试和门禁做全，流式链路才能在真实流量下稳定运行。

一、AsyncStream 的运行语义：先搞清楚再谈策略