Swift中如何创建并管理适量Actor以最大化CPU使用率

开发中经常会遇到需要对单个资源执行大量批量操作的场景，最典型的就是把同一张图像导出为多种不同分辨率的版本。

• 如果目标资源遵循Sendable协议（比如CGImage），直接调用withTaskGroup为每个操作创建子任务就能最大化CPU利用率——前提是这个Sendable资源本身没有内置限制并发访问的同步逻辑。
• 但很多场景下资源并不遵循Sendable协议，比如CGPDFPage，这类资源常规有三种处理思路：
  • 完全不使用并发任务，在单线程for循环里串行执行所有操作
  • 把非Sendable资源封装在actor中，依然用withTaskGroup给每个操作创建子任务，子任务调用actor的隔离方法执行逻辑
  • 用withTaskGroup创建子任务，每个子任务内部单独初始化一份全新的资源副本再执行操作

方案1和方案2的实际执行效率几乎没有差别：方案2哪怕用了withTaskGroup，受actor的串行隔离特性限制，所有操作还是排队串行执行，完全利用不上多核性能。
方案3确实能跑满CPU，但要求每次执行操作前都重新初始化资源，批量操作的量级往往达到数千次，而资源初始化通常有不小的开销——比如从文件系统读取文件加载资源。哪怕算上这部分额外开销，方案3的整体速度可能还是比前两种快，但实现逻辑既不优雅也存在不必要的性能浪费。

理想的实现逻辑是创建和CPU算力刚好匹配的资源/actor实例，比如8核设备就创建8个资源实例，每个实例封装在独立actor里。后续执行数千次批量操作时依然为每个操作创建任务，每个任务从「Actor池」里选取当前空闲的actor来执行对应操作。

方案结论 #

你提出的固定大小Actor池方案就是当前场景下的最优解，既规避了单actor串行执行的性能瓶颈，也不用为每个任务重复初始化资源带来冗余开销，资源实例数和CPU算力对齐，不会出现多余的资源占用和初始化浪费。

Actor池的实现逻辑 #

核心逻辑非常简单，就是维护固定数量的worker actor+一个待执行任务的等待队列，不需要引入复杂的调度逻辑：

1. 初始化阶段先获取当前设备的活跃CPU核心数，直接调用ProcessInfo.processInfo.activeProcessorCount就能拿到准确值，按照这个数量创建封装了非Sendable资源的worker actor，每个actor初始化时就完成资源加载，后续全程复用不重复初始化
2. 维护一个FIFO的异步等待队列，存所有待执行任务的continuation
3. 每个worker actor完成当前操作后，主动检查等待队列，有等待任务就取下一个继续执行，没有任务就标记为空闲状态
4. 新任务提交时先检查有没有空闲的worker，有就直接分配执行，没有就把任务挂起存入等待队列

给一个可直接复用的最小实现代码：

// 封装非Sendable资源的Worker Actor
actor ResourceWorker<Resource> {
    private let resource: Resource
    private var isIdle = true
    
    init(resourceLoader: () throws -> Resource) rethrows {
        self.resource = try resourceLoader()
    }
    
    func execute<T>(_ operation: (Resource) throws -> T) rethrows -> T {
        isIdle = false
        defer { isIdle = true }
        return try operation(resource)
    }
    
    func isAvailable() -> Bool { isIdle }
}

// Actor池本体
actor ResourcePool<Resource> {
    private var workers: [ResourceWorker<Resource>]
    private var waitQueue: [CheckedContinuation<ResourceWorker<Resource>, Never>] = []
    
    init(
        workerCount: Int = ProcessInfo.processInfo.activeProcessorCount,
        resourceLoader: () throws -> Resource
    ) rethrows {
        self.workers = try (0..<workerCount).map { _ in
            try ResourceWorker(resourceLoader: resourceLoader)
        }
    }
    
    // 获取可用worker
    func acquireWorker() async -> ResourceWorker<Resource> {
        // 优先查找现有空闲worker
        for worker in workers {
            if await worker.isAvailable() {
                return worker
            }
        }
        // 无空闲worker则挂起等待
        return await withCheckedContinuation { continuation in
            waitQueue.append(continuation)
        }
    }
    
    // 任务执行完成后触发队列调度
    func dispatchPending() {
        guard !waitQueue.isEmpty else { return }
        let continuation = waitQueue.removeFirst()
        Task {
            let availableWorker = await self.acquireWorker()
            continuation.resume(returning: availableWorker)
        }
    }
}

使用方式也很简单，配合withTaskGroup做批量操作即可，不需要手动管理worker的分配释放：

// 初始化资源池，示例为加载PDF第一页资源
let pdfPagePool = ResourcePool {
    guard let pdfDoc = CGPDFDocument(pdfFileURL as CFURL),
          let firstPage = pdfDoc.page(at: 1) else {
        fatalError("资源初始化失败")
    }
    return firstPage
}

// 批量执行任务
await withTaskGroup(of: OperationResult.self) { group in
    for renderConfig in allRenderConfigs {
        group.addTask {
            let worker = await pdfPagePool.acquireWorker()
            let renderResult = await worker.execute { page in
                // 执行具体的PDF渲染逻辑
                return renderPage(page, config: renderConfig)
            }
            await pdfPagePool.dispatchPending()
            return renderResult
        }
    }
    // 汇总所有任务结果...
}

几个实现的注意点：

• 纯CPU计算场景下，worker数量保持和活跃CPU核心数一致即可，多创建worker只会带来额外的线程调度开销，不会提升执行效率
• 资源初始化只在池创建阶段执行一次，后续所有任务复用已有实例，完全规避重复加载的开销
• 等待队列采用FIFO顺序，不会出现任务长时间排队拿不到资源的饥饿问题
• 如果批量操作包含部分IO等待逻辑（比如把处理结果写磁盘），可以适当把worker数量调整为核心数的1.5-2倍，进一步提升利用率

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