Swift中如何用AsyncThrowingAppendSequence追加异步序列并处理错误？

AsyncThrowingAppendSequence 的追加顺序与惰性消费

Swift 的异步序列通常不是一上来就把所有数据读进内存，而是由消费方驱动。AsyncThrowingAppendSequence 可以理解为两个异步序列的串联容器：它先完整消费 base 序列，等 base 正常结束后，再开始消费 suffix 序列。只要 base 还在产出元素，suffix 的迭代器通常不会被推进，因此不会提前触发网络请求、文件读取或数据库查询。

这种顺序语义和同步数组的拼接不同。数组拼接会立即产生新数组，而异步序列拼接是惰性的。比如 base 是一个无限流，suffix 永远不会执行；base 是一个慢速流，suffix 也会等到 base 完全结束才出现。理解这一点，能避免把追加误认为并行合并。

在类型层面，AsyncThrowingAppendSequence 适合被追加的序列可能抛出错误的场景。它会把 suffix 的失败传播给消费者，让 for try await 的 catch 分支能够捕获。若两个序列都只产出元素而不失败，使用普通追加包装更符合语义；若希望错误也作为数据流的一部分，则需要在元素层面做包装。

下面这段示例展示最基本的追加消费方式。primary 是可能失败的异步流，secondary 是备用流，二者通过 append 串成一条流。

import Foundation

enum StreamError: Error {
    case failed
}

let primary = AsyncThrowingStream<Int, Error> { continuation in
    let task = Task {
        for value in 1...3 {
            continuation.yield(value)
        }
        continuation.finish()
    }
    continuation.onTermination = { _ in task.cancel() }
}

let secondary = AsyncThrowingStream<Int, Error> { continuation in
    let task = Task {
        for value in 101...103 {
            continuation.yield(value)
        }
        continuation.finish()
    }
    continuation.onTermination = { _ in task.cancel() }
}

let appended = primary.append(secondary)

do {
    for try await value in appended {
        print(value)
    }
} catch {
    print("追加序列失败: \(error)")
}

这段代码中，输出顺序通常是 1、2、3、101、102、103。如果 primary 在 yield 1 后调用 continuation.finish(throwing: StreamError.failed)，secondary 不会继续被消费，错误会直接跳到 catch。这里的关键不是语法，而是生命周期：追加序列把前一个流结束作为后一个流开始的条件。

错误处理边界：追加不是兜底

很多错误处理问题来自把 AsyncThrowingAppendSequence 当成 fallback。实际上它更像流水线，而不是保险箱。base 失败时，整条流失败；suffix 失败时，已经消费过的 base 元素不会回滚，错误在消费到 suffix 阶段时抛出。若业务希望 base 失败后自动切换到 suffix，就不能只依赖 append，而要在消费端捕获错误后手动启动 suffix。

func consumeWithFallback(
    primary: AsyncThrowingStream<Int, Error>,
    fallback: AsyncThrowingStream<Int, Error>
) async throws {
    do {
        for try await value in primary {
            print("primary: \(value)")
        }
    } catch {
        print("primary 失败，切换到 fallback: \(error)")
        for try await value in fallback {
            print("fallback: \(value)")
        }
    }
}

上面的写法保留了清晰的边界：只有 primary 正常结束后，fallback 才不会被启动；只有 primary 抛错时，fallback 才作为恢复路径出现。这与 AsyncThrowingAppendSequence 的顺序追加不同。追加是成功接力，兜底是失败接管。

另一种思路是把错误封装进元素，让流本身不抛出。这样 AsyncThrowingAppendSequence 可以稳定消费，错误由调用方按业务判断。示例中用枚举包装每个元素，后续可以决定是中断、忽略还是重试。

enum StreamOutcome<Element> {
    case value(Element)
    case failure(Error)
}

func safeStream(_ source: AsyncThrowingStream<Int, Error>) -> AsyncStream<StreamOutcome<Int>> {
    AsyncStream { continuation in
        let task = Task {
            do {
                for try await element in source {
                    continuation.yield(.value(element))
                }
                continuation.finish()
            } catch {
                continuation.yield(.failure(error))
                continuation.finish()
            }
        }
        continuation.onTermination = { _ in task.cancel() }
    }
}

这种方案适合需要跨多个序列统一处理错误的场景。代价是元素类型变复杂，调用方必须持续判断 .failure，否则错误会被当成普通数据吞掉。若系统只关心最终结果，直接抛出并让上层捕获可能更简洁。

取消与资源释放：追加链上的任务如何收尾

异步序列的消费往往发生在 Task 中。取消一个外部 Task 并不会自动让所有底层异步序列停止，除非这些序列内部检查取消状态或注册终止回调。AsyncThrowingAppendSequence 作为包装器，会把取消传递给当前正在消费的子序列，但子序列是否真正释放资源，取决于实现。

func cancellableNumbers(limit: Int) -> AsyncThrowingStream<Int, Error> {
    AsyncThrowingStream { continuation in
        let task = Task {
            for i in 1...limit {
                if Task.isCancelled {
                    continuation.finish(throwing: CancellationError())
                    return
                }
                continuation.yield(i)
                try await Task.sleep(nanoseconds: 50_000_000)
            }
            continuation.finish()
        }
        continuation.onTermination = { _ in
            task.cancel()
        }
    }
}

let first = cancellableNumbers(limit: 3)
let second = cancellableNumbers(limit: 3)
let combined = first.append(second)

let job = Task {
    do {
        for try await value in combined {
            print(value)
        }
    } catch {
        print("取消或失败: \(error)")
    }
}

job.cancel()

如果 first 或 second 内部没有检查 Task.isCancelled，取消可能只是让外层不再继续等待，底层 Task 仍在空转。因此编写自定义异步流时，onTermination 和 Task.isCancelled 都很重要。onTermination 负责释放外部资源，Task.isCancelled 负责在长循环中主动退出。

对于追加链，还要避免一个常见误区：认为 suffix 在 base 取消后也不会启动。若 base 因取消而抛出 CancellationError，追加链会停止，suffix 不会继续。若 base 正常结束，suffix 开始后才取消，则 suffix 需要自己处理取消。错误和取消都应在流的边界处被明确表达。

实践建议：什么时候用追加，什么时候用合并

选择 AsyncThrowingAppendSequence 的场景很明确：需要按顺序消费两段异步数据，例如先读缓存流，再读网络流；先播放本地片段，再播放远端片段；先处理日志主文件，再处理归档文件。只要业务要求严格先后顺序，并且后一段失败也要能抛出，它就很合适。

如果两段数据可以并行到达，或者希望错误只影响单个分支，就不该用追加。并行合并更适合 TaskGroup 或自定义 AsyncThrowingStream。例如两个网络源同时请求，任一失败可以降级，两个成功则合并输出。追加会把并行需求变成串行瓶颈。

最后，错误处理要提前设计契约。若消费者期望流失败就停止，直接抛出错误最简单。若消费者期望部分失败继续，把错误映射成枚举或结果类型更稳定。若消费者期望失败后自动切换，用显式捕获和 fallback 比追加更符合语义。AsyncThrowingAppendSequence 提供的是顺序拼接能力，不是完整的容错策略。

AsyncSequenceAsyncThrowingAppendSequence异步序列修改时间：2026-09-09 10:56:29