并行执行异步任务

Task Group作为Swift结构化并发的核心工具，擅长并行执行一组异步任务，显著提升多任务场景的效率。前一节介绍了Task Group的基本概念和用法，本节将深入探讨如何利用它实现并行执行，包括任务添加、结果收集和错误处理。通过一个实际案例，你将学会如何设计并行异步流程，并在实践中优化性能。

并行执行的原理

Task Group利用Swift并发运行时和GCD线程池，让组内的子任务并行运行。与串行执行相比，并行执行将多个任务的等待时间重叠，缩短总耗时。例如，串行执行三个1秒任务需3秒，而并行只需约1秒。

基本流程：

  1. 创建组：用withTaskGroup或withThrowingTaskGroup定义任务组。
  2. 添加任务：通过addTask提交异步操作。
  3. 等待结果：迭代组或聚合结果。

示例：

func fetchData(id: Int) async -> String {
    try? await Task.sleep(nanoseconds: 1_000_000_000) // 模拟1秒延迟
    return "数据\(id)"
}

Task {
    let start = Date()
    let results = await withTaskGroup(of: String.self) { group in
        for id in 1...3 {
            group.addTask {
                await fetchData(id: id)
            }
        }
        var collected = [String]()
        for await result in group {
            collected.append(result)
        }
        return collected
    }
    let duration = Date().timeIntervalSince(start)
    print("结果：\(results), 耗时：\(duration)秒")
}

输出：

结果：["数据1", "数据2", "数据3"], 耗时：1.0xxx秒
  • 三个任务并行，总耗时接近单个任务时间。

并行任务的实现

以下是实现并行执行的关键步骤：

1. 添加并行任务

使用addTask动态添加子任务：

let items = await withTaskGroup(of: Int.self) { group in
    for i in 1...5 {
        group.addTask {
            try? await Task.sleep(nanoseconds: 500_000_000)
            return i * i
        }
    }
    return await group.reduce(0, +) // 平方和
}
print("平方和：\(items)") // 输出：55 (1+4+9+16+25)
  • 任务立即并行启动。
  • reduce聚合结果。

2. 处理抛出错误的并行任务

用withThrowingTaskGroup管理可能失败的任务：

enum FetchError: Error {
    case invalidData
}

func fetchWithError(id: Int) async throws -> Int {
    try await Task.sleep(nanoseconds: 500_000_000)
    if id % 2 == 0 { throw FetchError.invalidData }
    return id
}

Task {
    do {
        let total = try await withThrowingTaskGroup(of: Int.self) { group in
            for id in 1...4 {
                group.addTask {
                    try await fetchWithError(id: id)
                }
            }
            return try await group.reduce(0, +)
        }
        print("总和：\(total)")
    } catch {
        print("错误：\(error)")
    }
}

输出：

错误：invalidData
  • 任意任务抛出错误，组立即结束，未完成任务取消。

3. 动态添加任务

在组内动态扩展任务：

let numbers = await withTaskGroup(of: Int.self) { group in
    group.addTask { 1 } // 初始任务
    for await result in group {
        if result < 3 {
            group.addTask { result + 1 } // 动态添加
        }
    }
    return await group.collectAll() // 自定义扩展函数
}

extension AsyncTaskGroup {
    func collectAll() async -> [Element] {
        var results: [Element] = []
        for await result in self {
            results.append(result)
        }
        return results
    }
}

print("结果：\(numbers)") // 输出：[1, 2, 3]
  • 任务执行中添加新任务，前提是迭代尚未完成。

实战案例：批量下载图片

假设我们要并行下载多张图片并显示：

struct ImageDownloader {
    func downloadImage(from url: URL) async throws -> UIImage {
        let (data, _) = try await URLSession.shared.data(from: url)
        guard let image = UIImage(data: data) else { throw ImageError.invalidData }
        return image
    }
}

enum ImageError: Error {
    case invalidData
}

@MainActor
class GalleryViewController: UIViewController {
    @IBOutlet weak var collectionView: UICollectionView!
    private var images: [UIImage] = []

    override func viewDidLoad() {
        super.viewDidLoad()
        loadImages()
    }

    private func loadImages() {
        Task {
            let urls = [
                URL(string: "https://example.com/image1.jpg")!,
                URL(string: "https://example.com/image2.jpg")!,
                URL(string: "https://example.com/image3.jpg")!
            ]
            
            let downloader = ImageDownloader()
            images = try await withThrowingTaskGroup(of: UIImage.self) { group in
                for url in urls {
                    group.addTask {
                        try await downloader.downloadImage(from: url)
                    }
                }
                var downloaded: [UIImage] = []
                for try await image in group {
                    downloaded.append(image)
                }
                return downloaded
            }
            
            collectionView.reloadData()
        }
    }
}

// UICollectionViewDataSource
extension GalleryViewController: UICollectionViewDataSource {
    func collectionView(_ collectionView: UICollectionView, numberOfItemsInSection section: Int) -> Int {
        images.count
    }

    func collectionView(_ collectionView: UICollectionView, cellForItemAt indexPath: IndexPath) -> UICollectionViewCell {
        let cell = collectionView.dequeueReusableCell(withReuseIdentifier: "ImageCell", for: indexPath)
        if let imageCell = cell as? ImageCollectionViewCell {
            imageCell.imageView.image = images[indexPath.item]
        }
        return cell
    }
}

分析

  • 并行下载：三个图片同时获取，总耗时取决于最慢的请求。
  • 错误处理：任意下载失败抛出错误。
  • 主线程更新：@MainActor确保UI刷新安全。

性能与限制

  • 优势：并行缩短总时间，适合IO密集型任务。
  • 限制：线程池容量有限，过多任务可能导致竞争；CPU密集型任务需谨慎并行。

最佳实践

  1. 控制任务数量
    避免创建过多子任务，建议根据设备核心数调整。

  2. 优先级分配
    为组设置优先级：

    Task(priority: .userInitiated) {
    await withTaskGroup(of: Int.self) { ... }
}

  3. 提前退出
    用group.cancelAll()取消剩余任务：

    if shouldStop { group.cancelAll() }

  4. 结果顺序
    结果按完成顺序返回，需手动排序若需特定顺序。

小结

Task Group通过并行执行异步任务，提升了多任务处理的效率。本节通过图片下载案例展示了其实现细节和应用场景，强调了结构化并发的优势。掌握并行执行，你将能优化批量操作的性能。下一节将探讨非结构化并发，与本章内容形成对比，进一步完善你的并发知识体系。

内容说明

  • 结构：从原理到实现，再到案例和实践，最后总结。
  • 代码：包含简单并行、抛错和图片下载示例，突出实用性。
  • 语气：实践性且深入，适合技术书籍核心章节。
  • 衔接：承接前节（Task Group简介），预告后续（非结构化并发）。
Last Updated::