如何使用 Swift Concurrency 约束并发（如 maxConcurrentOperationCount）？

Question

我正在尝试执行一系列网络请求，并希望限制新的 Swift 并发系统中的并发任务数量。对于操作队列，我们​​将使用maxConcurrentOperationCount. 在组合中，flatMap(maxPublishers:_:). 新的 Swift Concurrency 系统中的等价物是什么？

例如，它并不是非常相关，但请考虑：

func downloadAll() async throws {
    try await withThrowingTaskGroup(of: Void.self) { group in
        for index in 0..<20 {
            group.addTask { try await self.download(index) }
        }

        try await group.waitForAll()
    }
}

这会导致所有请求同时运行：

URLSession不尊重这一事实httpMaximumConnectionsPerHost很有趣，但这不是这里的突出问题。更一般地说，我正在寻找如何限制一系列并行运行的异步任务的并发程度。

Answer 1

达到一定计数后，可以group.next()在循环内插入调用，例如：

\n

func downloadAll() async throws {\n    try await withThrowingTaskGroup(of: Void.self) { group in\n        for index in 0..<20 {\n            if index >= 6 { try await group.next() }\n            group.addTask { try await self.download(index) }\n        }\n\n        try await group.waitForAll()\n    }\n}\n

\n

这会导致一次不超过六个：

\n

\n\n

为了完整起见，我应该指出，在 WWDC 2023 中，除了结构化并发的基础知识之外，Apple 还建议了一种替代模式：

\n

withTaskGroup(of: Something.self) { group in\n    for _ in 0 ..< maxConcurrentTasks {\n        group.addTask { \xe2\x80\xa6 }\n    }\n    while let <partial result> = await group.next() {\n        if !shouldStop {\n            group.addTask { \xe2\x80\xa6 }\n        }\n    }\n}\n

\n

在这个例子中，这可能会转化为：

\n

func downloadAll() async throws {\n    try await withThrowingTaskGroup(of: Void.self) { group in\n        for index in 0..<6 {\n            group.addTask { try await self.download(index) }\n        }\n        var index = 6\n        while try await group.next() != nil {\n            if index < 20 {\n                group.addTask { [index] in try await self.download(index) }\n            }\n            index += 1\n        }\n    }\n}\n

\n

产量（以仪器计）：

\n

\n

这个想法非常相似，即您group.addTask {\xe2\x80\xa6}达到所需的最大并发度，但然后group.next()在添加每个后续任务之前。这是另一种破解坚果的方法。

\n