Swift Combine使用MergeMany时如何限制并发网络请求数避免超时

Publishers.MergeMany本身不提供并发控制能力，订阅触发时会立刻订阅所有传入的Publisher，一次性发起全部网络请求，请求量过大时很容易触发服务端超时。Combine原生提供了对应能力实现类似DispatchSemaphore的并发限制效果，不需要自行引入信号量逻辑。

推荐方案：使用flatMap的maxPublishers参数（原生支持） #

Combine的flatMap算子内置了背压控制能力，通过maxPublishers参数可以指定下游同时持有的活跃Publisher最大数量，本质就是限制并发执行的任务数，完全匹配需求，且不会出现手动管理信号量带来的线程阻塞、死锁问题。

改造后的代码 #

假设需要限制最多同时发起3个网络请求，代码如下，和原有输出逻辑完全一致：

// 配置最大并发请求数，根据服务端承载能力调整即可
let maxConcurrentRequests = 3

let mergedPubs = urlRequests.publisher
    .flatMap(maxPublishers: .max(maxConcurrentRequests)) { request in
        dataTaskPublisher(for: request)
            .decode(type: RawJSON.self, decoder: JSONDecoder())
            .mapError { _ in
                URLError(.badServerResponse)
            }
    }
    .collect()
    .eraseToAnyPublisher()

运行逻辑说明 #

• 首先通过urlRequests.publisher将请求数组转为按顺序输出单个请求的序列Publisher
• flatMap(maxPublishers: .max(3))收到前3个请求时，会立刻启动对应的网络任务，此时并发数达到上限
• 后续到达的请求会被暂存，不会立刻发起；每当有一个在途请求完成（成功/失败都会触发），就会自动取出下一个待处理请求发起，始终保持同时运行的请求数不超过设置的阈值
• 最后的collect()算子保持原有逻辑：等所有请求全部完成后，将所有结果拼接为数组返回

该方案的结果顺序和原有MergeMany实现完全一致：collect输出的数组顺序为请求完成的顺序，而非原始请求数组的顺序。如果需要保持结果和原始请求顺序一致，可以在flatMap内给每个结果绑定原始索引，最终收集完成后再按索引排序即可。

不推荐的方案：手动接入DispatchSemaphore #

如果有特殊场景需要自定义并发控制逻辑，确实可以接入DispatchSemaphore，但必须注意不能阻塞Combine和URLSession的内部工作线程，否则很容易出现请求卡死、回调不触发的问题，稳定性远不如原生的flatMap方案。

注意事项 #

• 禁止在dataTaskPublisher的事件回调中直接调用semaphore.wait()，会阻塞URLSession的调度队列，导致请求完成后无法发送事件，触发死锁
• 手动管理信号量需要配对处理成功、失败、取消三种场景的signal()调用，漏写任意一种场景都会导致并发计数异常，后续请求无法发起

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