并行执行数千个操作然后等待结束

Question

我想加快某些进程的速度，因此我编写了一个快速的 CLI 脚本，该脚本可以并行处理数千个文件，并将每个文件的处理结果写入一个文件中。 （文件的顺序并不重要）

所以我写了下面的代码，它在 Xcode 单元测试中工作（即使有大约 1200 个文件的列表！）但是，当我在没有 Xcode 的情况下从命令行执行程序并使用相同的文件列表时，它永远不会结束。看起来快到最后了。

我读到，有时跨越太多线程会导致程序停止，因为它耗尽了资源，但我认为 DispatchQueue.concurrentPerform 会解决这个问题...我不知道为什么这在 XCTests 中有效，而在终端。

我尝试过 DispatchGroup 和 Semaphore 方法，但都有同样的问题...

非常感谢任何帮助。

let filePaths: [String] = Array with thousands of file paths to process

let group = DispatchGroup()
let concurrentQueue = DispatchQueue(label: "my.concurrent.queue", qos: .userInitiated, attributes: .concurrent)
let serialQueue = DispatchQueue(label: "my.serial.queue", qos: .userInitiated)

group.enter()
concurrentQueue.async {
    DispatchQueue.concurrentPerform(iterations: filePaths.count) { (fileIndex) in
        let filePath = filePaths[fileIndex]
        let result = self.processFile(path: filePath)
        group.enter()
        serialQueue.async {
            self.writeResult(result)
            group.leave()
        }
    }
    group.leave()
}
group.wait()

Answer 1

一些观察：

1. 您的代码是

   group.enter()
   serialQueue.async {
       self.writeResult(result)
       group.leave()
   }
   

   可以简化为：

   serialQueue.async(group: group) {
       self.writeResult(result)
   }
   

2. 考虑：

   group.enter()
   concurrentQueue.async {
       DispatchQueue.concurrentPerform(iterations: filePaths.count) { (fileIndex) in
           …                       // the synchronous work you want off the main thread
       }
       group.leave()
   }
   group.wait()
   

   如果你真的要

   wait

   ，那就会使

   concurrentQueue

   变得多余。这可以简化为：

   DispatchQueue.concurrentPerform(iterations: filePaths.count) { (fileIndex) in
       …                           // the synchronous work you want off the main thread
   }
   

3. 或者，更好的是，避免

   wait

   ，它会阻塞当前队列：

   DispatchQueue.global().async {
       DispatchQueue.concurrentPerform(iterations: filePaths.count) { (fileIndex) in
           …                       // the synchronous work you want off the main thread
       }
       DispatchQueue.main.async {
           …                       // whatever you want to do when this is all done
       }
   }
   

因此，也许：

DispatchQueue.global().async { [self] in
    let group = DispatchGroup()
    let serialQueue = DispatchQueue(label: "my.serial.filewrite")

    DispatchQueue.concurrentPerform(iterations: filePaths.count) { fileIndex in
        let filePath = filePaths[fileIndex]
        let result = processFile(path: filePath)
        serialQueue.async(group: group) {
            writeResult(result)
        }
    }
    
    group.notify(queue: .main) {
        …                           // whatever you want to do when this is all done
    }
} 

或者更简单地说：

DispatchQueue.global().async { [self] in
    DispatchQueue.concurrentPerform(iterations: filePaths.count) { fileIndex in
        let filePath = filePaths[fileIndex]
        let result = processFile(path: filePath)
        writeResult(result)
    }
    
    DispatchQueue.main.async {
        …                           // whatever you want to do when this is all done
    }
}