Swift中如何用Actor实现资源并行读取、读写互斥？

现有实现对比 #

基于DispatchQueue的安全字典（支持并行读） #

通过并发队列加barrier标记，可实现资源的并行读取，写入时阻塞所有读写操作，保证线程安全：

struct SafeDict<Element> {
    private var dict = [String: Element]()
    private var queue = DispatchQueue(label: "WriteQueue", attributes: .concurrent)
    
    func get(_ key: String) -> Element? {
        queue.sync { return dict[key] }
    }
    
    mutating func set(_ key: String, value: Element) {
        queue.sync(flags: .barrier) { dict[key] = value }
    }
}

var safeDict = SafeDict<Int>()
for i in 1...4 {
    DispatchQueue.global().async {
        switch i {
        case 1:
            safeDict.get("one")
        case 2:
            safeDict.set("one", value: 1) // 等待get完成（符合预期）
        case 3:
            safeDict.get("one") // 写入完成后与其他get并行执行
        case 4:
            safeDict.get("two") // 写入完成后与其他get并行执行
        default:
            print("done")
        }
    }
}

基于Actor的安全字典（读取串行，存在问题） #

直接用Actor实现的话，由于Actor所有方法会串行执行，即使是无关的读取操作也会互相等待，无法实现并行读取：

actor SafeDictActor<Element> {
    private var dict = [String: Element]()
    
    func get(_ key: String) -> Element? { dict[key] }
    
    func set(_ key: String, value: Element) { dict[key] = value }
}

let safeDictActor = SafeDictActor<Int>()
for i in 1...4 {
    Task {
        switch i {
        case 1:
            await safeDictActor.get("one")
        case 2:
            await safeDictActor.set("one", value: 1) // 等待get完成（符合预期）
        case 3:
            await safeDictActor.get("one") // 等待写入完成（符合预期）
        case 4:
            await safeDictActor.get("two") // 等待前一个get完成（不符合预期，本该并行）
        default:
            print("done")
        }
    }
}

简洁解决方案：Actor内部结合并发队列 #

在Actor内部维护一个并发DispatchQueue，用它处理读写逻辑，既保留Actor的线程安全特性，又能实现并行读：

actor SafeDictActor<Element> {
    private var dict = [String: Element]()
    private let queue = DispatchQueue(label: "SafeDictQueue", attributes: .concurrent)
    
    func get(_ key: String) async -> Element? {
        await withCheckedContinuation { continuation in
            queue.sync {
                continuation.resume(returning: dict[key])
            }
        }
    }
    
    func set(_ key: String, value: Element) async {
        await withCheckedContinuation { continuation in
            queue.async(flags: .barrier) {
                self.dict[key] = value
                continuation.resume()
            }
        }
    }
}

方案说明： #

• 读取操作通过queue.sync提交到并发队列，多个读请求可并行执行
• 写入操作通过queue.async(flags: .barrier)提交，会阻塞队列内所有操作，完成写入后再恢复后续读写
• 用withCheckedContinuation把同步的队列操作包装成异步方法，符合Actor的异步调用规范

测试效果 #

用同样的测试代码验证，第三个和第四个get操作会在写入完成后并行执行，解决了原Actor实现的串行问题：

let safeDictActor = SafeDictActor<Int>()
for i in 1...4 {
    Task {
        switch i {
        case 1:
            let value = await safeDictActor.get("one")
            print("读取one: \(value ?? 0)")
        case 2:
            await safeDictActor.set("one", value: 1)
            print("写入one为1")
        case 3:
            let value = await safeDictActor.get("one")
            print("写入后读取one: \(value ?? 0)")
        case 4:
            let value = await safeDictActor.get("two")
            print("读取two: \(value ?? 0)")
        default:
            print("完成")
        }
    }
}

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