Swift Concurrency：如何结合TaskGroup与AsyncStream实现并行图片下载？

背景 #

假设我想要使用Swift Concurrency在后台下载300张图片，有两个核心需求：

• 尽可能高的并行度
• 调用方能够在每张图片下载完成后立即接收结果，而非等待所有图片下载完毕

现有方案分析 #

• TaskGroup：具备高并行度、轻量子任务、支持取消操作等优势，但必须等待全部300个子任务完成后才会返回所有结果。
• AsyncStream：允许在图片下载完成后立即返回结果，但本身不支持并行，下载任务会逐个按顺序执行。

问题 #

我尝试用AsyncStream包装TaskGroup来结合两者的优势，示例代码如下：

let stream = AsyncStream(NSImage.self) { continuation in
    
    _ = await withTaskGroup(of: NSImage.self, returning: [NSImage].self) { taskGroup in
        
        let imageURLs: [URL] = ... // 包含300个图片下载链接的数组
        for imageURL in imageURLs {
            taskGroup.addTask { await downloadImage(url: imageURL) }
        }

        for await result in taskGroup {
            continuation.yield(result)
        }

        continuation.finish()
        return []
    }
}

但AsyncStream的初始化闭包不支持async，那么在Swift Concurrency中，实现这种“高并行+实时返回结果”行为的最佳方式是什么？

解决方案 #

你可以在AsyncStream的初始化闭包中启动一个独立的Task来执行TaskGroup的异步逻辑，这样既不会阻塞初始化过程，又能实现并行下载并实时返回结果。修改后的代码如下：

let stream = AsyncStream(NSImage.self) { continuation in
    // 启动后台Task处理并行下载逻辑
    Task {
        await withTaskGroup(of: NSImage.self) { taskGroup in
            let imageURLs: [URL] = ... // 300个图片URL数组
            
            for imageURL in imageURLs {
                taskGroup.addTask {
                    await downloadImage(url: imageURL)
                }
            }
            
            // 遍历TaskGroup结果，每拿到一张图就通过stream返回
            for await image in taskGroup {
                continuation.yield(image)
            }
            
            // 所有任务完成后结束stream
            continuation.finish()
        }
    }
}

补充说明 #

1. 取消支持：如果需要支持取消操作，可以保存启动的Task实例，或者让调用方通过AsyncStream的迭代器触发取消——TaskGroup会自动将取消信号传播给所有子任务。
2. 错误处理：如果下载过程可能出现错误，可以将TaskGroup的类型改为Result<NSImage, Error>，把错误也通过AsyncStream返回：

let stream = AsyncStream(Result<NSImage, Error>.self) { continuation in
    Task {
        await withTaskGroup(of: Result<NSImage, Error>.self) { taskGroup in
            let imageURLs: [URL] = ...
            
            for imageURL in imageURLs {
                taskGroup.addTask {
                    do {
                        let image = await downloadImage(url: imageURL)
                        return .success(image)
                    } catch {
                        return .failure(error)
                    }
                }
            }
            
            for await result in taskGroup {
                continuation.yield(result)
            }
            
            continuation.finish()
        }
    }
}

内容的提问来源于stack exchange，提问作者Bryan