DispatchSemaphore+DispatchQueue未按预期并行，求技术排查方案

问题描述 #

我尝试通过多线程/并行处理缩短视频轨道CMSampleBuffer的处理耗时，处理后将结果存入数组备用。由于每个处理操作的RAM和CPU开销较高，我使用DispatchSemaphore将并行任务数限制为5个，避免设备因内存问题崩溃，但当前实现几乎是串行执行而非并行，需要排查问题。

原代码如下：

class MyService {

  let semaphore = DispatchSemaphore(value: 5)
  let processQueue = DispatchQueue(label: "custom.process", attributes: .concurrent)

    func startReading() {

        for sampleBuffer in sampleBuffers {
            // signal wait
            semaphore.wait()

            // async queue
            processQueue.async {

                // run taks
                self.process(buffer: sampleBuffer) { pixelBuffer in

                    // singal to semaphore
                    self.semaphore.signal()

                }
            }
        }
    }


  func process(buffer: CMSampleBuffer, completion: @escaping (CVPixelBuffer) -> (Void)) {

        // run on a background thread to avoid UI freeze
        DispatchQueue.global(qos: .userInteractive).async {

            // Do something
            // Do something
            // Do something
            // Do something
            completion(processedBuffer)
        }   
       
    }

}

问题根源 #

核心问题在于semaphore.wait()的位置错误：
它被放在了调用startReading的线程（大概率是主线程）的for循环中，每次循环都会先阻塞等待信号量，再提交任务到并行队列。但processQueue.async的闭包本身执行极快（只是把处理任务调度到全局队列），而semaphore.signal()要等到process方法的异步处理完成后才会触发，这就导致for循环被wait()卡住，必须等前一个任务的completion回调触发后，才能继续提交下一个任务，最终变成了串行提交，自然无法并行执行。

修复方案 #

把semaphore.wait()移到processQueue.async的闭包内部，让for循环一次性把所有任务提交到并行队列，由信号量在队列内部控制并发数：

class MyService {

    let semaphore = DispatchSemaphore(value: 5)
    let processQueue = DispatchQueue(label: "custom.process", attributes: .concurrent)

    func startReading() {
        for sampleBuffer in sampleBuffers {
            processQueue.async {
                // 将wait移到并行队列的任务内部，避免阻塞主线程/调用线程
                self.semaphore.wait()
                
                self.process(buffer: sampleBuffer) { pixelBuffer in
                    self.semaphore.signal()
                }
            }
        }
    }

    func process(buffer: CMSampleBuffer, completion: @escaping (CVPixelBuffer) -> Void) {
        DispatchQueue.global(qos: .userInteractive).async {
            // 处理逻辑
            // ...
            completion(processedBuffer)
        }   
    }
}

额外优化建议 #

如果process方法里的处理逻辑不需要强制切换到全局队列，直接在processQueue的线程上执行，可以减少线程切换的开销，进一步提升并行效率：

func process(buffer: CMSampleBuffer, completion: @escaping (CVPixelBuffer) -> Void) {
    // 直接在当前队列（processQueue）执行处理逻辑，无需额外切换线程
    // 处理逻辑
    // ...
    completion(processedBuffer)
}

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