Swift Actor内使用DispatchQueue.concurrentPerform的问题及合规性咨询

问题描述 #

我有一个Swift Actor，它包含一个长时间运行的update方法，该方法会修改大量共享状态。我希望在后台队列并行执行这项耗时工作，但不想因使用TaskGroup而让update函数变为async——因为在update函数内使用await会使Actor可重入，导致其他Actor函数在update完成前运行。

过去我使用DispatchQueue.concurrentPerform()处理此类CPU密集型计算，效果很好。但在Swift Playground的Actor中应用该方法时，出现了编译错误（代码中注释标注了问题位置）：

import Cocoa

actor Renderer {
    var items = [Int: Int]()

    func performUpdate() {
        var localItems = [Int: Int]()
        let lock = NSLock()

        DispatchQueue.global().sync {
            DispatchQueue.concurrentPerform(iterations: 10) { index in
                Thread.sleep(forTimeInterval: 10) //模拟CPU密集型计算
                Swift.print("Done: \(index)")

                lock.lock()
                localItems[index] = Int.random(in: 0...100) //并发执行代码中修改捕获的变量'localItems'，此处编译报错
                lock.unlock()
            }
        }

        items = localItems
    }
}

Task {
    let renderer = Renderer()
    await renderer.performUpdate()
    await Swift.print(renderer.items)
}

请问：

1. 如何将concurrentPerform()内的计算结果导出到performUpdate()函数中？
2. 这种结构是否符合Swift结构化并发运行时约定？
3. 有没有更Swift化的方式，在不使performUpdate()变为async的情况下执行并行密集计算？

解决方案与解释 #

1. 导出concurrentPerform()计算结果的正确方式 #

编译错误的核心原因是：concurrentPerform的闭包要求符合@Sendable协议，但直接在并发闭包中修改捕获的普通字典localItems，违反了Sendable的线程安全约定。

推荐两种解决方式：

• 方式一：使用线程安全容器收集结果
  封装一个线程安全的字典，避免在并发场景下直接修改普通字典：

  actor Renderer {
      var items = [Int: Int]()
  
      func performUpdate() {
          let threadSafeItems = ThreadSafeDictionary<Int, Int>()
  
          DispatchQueue.global().sync {
              DispatchQueue.concurrentPerform(iterations: 10) { index in
                  Thread.sleep(forTimeInterval: 1) //模拟计算
                  print("Done: \(index)")
                  threadSafeItems[index] = Int.random(in: 0...100)
              }
          }
  
          items = threadSafeItems.dictionary
      }
  }
  
  // 简易线程安全字典实现
  class ThreadSafeDictionary<Key: Hashable, Value> {
      private var dictionary = [Key: Value]()
      private let lock = NSLock()
  
      subscript(key: Key) -> Value? {
          get {
              lock.lock()
              defer { lock.unlock() }
              return dictionary[key]
          }
          set {
              lock.lock()
              defer { lock.unlock() }
              dictionary[key] = newValue
          }
      }
  
      var dictionary: [Key: Value] {
          lock.lock()
          defer { lock.unlock() }
          return dictionary
      }
  }
  

• 方式二：先独立计算再统一合并
  让每个并发任务先生成独立的键值对，最后一次性合并到字典，减少锁的使用频率：

  actor Renderer {
      var items = [Int: Int]()
  
      func performUpdate() {
          var results = [(Int, Int)]()
          let lock = NSLock()
  
          DispatchQueue.global().sync {
              DispatchQueue.concurrentPerform(iterations: 10) { index in
                  Thread.sleep(forTimeInterval: 1)
                  let value = Int.random(in: 0...100)
                  lock.lock()
                  results.append((index, value))
                  lock.unlock()
              }
          }
  
          // 最后统一合并到字典
          var localItems = [Int: Int]()
          for (key, value) in results {
              localItems[key] = value
          }
          items = localItems
      }
  }
  

2. 该结构是否符合Swift结构化并发约定？ #

不符合。DispatchQueue.concurrentPerform()和sync属于非结构化并发，脱离了Swift结构化并发的管理范畴。

Swift结构化并发的核心是通过Task、TaskGroup等API让并发任务的生命周期与调用栈绑定，自动管理任务的创建、执行和销毁，还能提供自动取消、任务层级追踪等特性。而你的代码使用Dispatch API，无法享受这些结构化并发的优势。

3. 更Swift化的同步并行计算方式 #

如果不想让performUpdate()变为async，可以结合Swift并发的Task.detached和信号量实现同步等待的并行计算，既利用Swift并发模型，又保持方法的同步性：

actor Renderer {
    var items = [Int: Int]()

    func performUpdate() {
        var results = [(Int, Int)]()
        let lock = NSLock()
        let semaphore = DispatchSemaphore(value: 0)
        let taskCount = 10

        for index in 0..<taskCount {
            Task.detached {
                Thread.sleep(forTimeInterval: 1)
                let value = Int.random(in: 0...100)
                lock.lock()
                results.append((index, value))
                lock.unlock()
                semaphore.signal()
            }
        }

        // 同步等待所有任务完成
        for _ in 0..<taskCount {
            semaphore.wait()
        }

        // 合并结果
        var localItems = [Int: Int]()
        for (key, value) in results {
            localItems[key] = value
        }
        items = localItems
    }
}

这种方式用Swift并发的Task替代Dispatch的并发执行，同时用信号量同步等待所有任务完成，既符合Swift的并发模型，又保持了performUpdate()的非async特性，避免Actor重入。

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