iOS并发编程：如何使用Semaphore替代barrier实现分组并发任务

Semaphore版本实现 #

你之前的代码只能串行执行的核心原因是：每提交1个上传任务就立刻调用wait()阻塞，必须等前一个任务完成发信号才会提交下一个任务，自然无法实现A组任务并发。
正确逻辑是先把A组所有3个任务全部提交到并发队列，再统一等待所有任务执行完成，再提交B组任务，实现代码如下：

func workflow() {
    // 避免阻塞当前调用线程（如主线程），将等待逻辑放到全局队列异步执行
    DispatchQueue.global().async {
        let concurrentQueue = DispatchQueue(label: "test.concurrent.queue", qos: .background, attributes: .concurrent)
        let semaphore = DispatchSemaphore(value: 0)
        
        // 批量提交A组所有任务，全部并发执行
        for i in 0..<3 {
            concurrentQueue.async {
                self.uploadImg(idx: "A_\(i)")
                semaphore.signal()
            }
        }
        // 等待3次信号，即A组全部任务完成
        for _ in 0..<3 {
            semaphore.wait(timeout: .distantFuture)
        }
        print("group A done")
        
        // 批量提交B组所有任务
        for i in 0..<3 {
            concurrentQueue.async {
                self.uploadImg(idx: "B_\(i)")
                semaphore.signal()
            }
        }
        // 等待B组全部任务完成
        for _ in 0..<3 {
            semaphore.wait(timeout: .distantFuture)
        }
        print("group B done")
    }
}

func uploadImg(idx info: String){
    Thread.sleep(forTimeInterval: 1)
    print("img \(info) uploaded")
}

Combine版本实现 #

如果使用Combine框架，可以将单个上传任务封装为Future Publisher，合并同组所有Publisher等待全部完成后再触发下一组任务，实现代码如下：

import Combine

var cancellables = Set<AnyCancellable>()

func combineWorkflow() {
    let concurrentQueue = DispatchQueue(label: "test.concurrent.queue", qos: .background, attributes: .concurrent)
    
    // 封装单个上传任务为Publisher
    func uploadTaskPublisher(idx: String) -> AnyPublisher<Void, Never> {
        Future { promise in
            concurrentQueue.async {
                self.uploadImg(idx: idx)
                promise(.success(()))
            }
        }
        .eraseToAnyPublisher()
    }
    
    // 合并A组3个任务，等待全部完成后执行后续逻辑
    Publishers.Zip3(
        uploadTaskPublisher(idx: "A_0"),
        uploadTaskPublisher(idx: "A_1"),
        uploadTaskPublisher(idx: "A_2")
    )
    .sink { [weak self] _ in
        guard let self = self else { return }
        print("group A done")
        
        // 合并执行B组3个任务
        Publishers.Zip3(
            uploadTaskPublisher(idx: "B_0"),
            uploadTaskPublisher(idx: "B_1"),
            uploadTaskPublisher(idx: "B_2")
        )
        .sink { _ in
            print("group B done")
        }
        .store(in: &self.cancellables)
    }
    .store(in: &cancellables)
}

func uploadImg(idx info: String){
    Thread.sleep(forTimeInterval: 1)
    print("img \(info) uploaded")
}

如果任务数量不固定，可替换Zip3为Publishers.MergeMany配合collect()操作符，实现任意数量任务的批量等待。

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