Swift Concurrency：如何用async/await实现DispatchSemaphore的并发保护？

我有一个单例类，需通过网络请求完成初始化，希望确保load()方法不会被并发调用，以此缓存结果、避免不必要的网络往返。在GCD时代，我可以使用DispatchQueue、DispatchGroup或DispatchSemaphore实现该需求，但在async/await的新环境下，尝试用Actor却无法达到预期效果。

示例说明 #

Actor默认同一时间仅能执行一个函数，但函数中包含异步调用时例外。

先定义一个日志函数：

func log(_ string: String) {
    print("\(Date.now) - \(string)")
}

对比以下两个Actor：

actor AsyncActor {
    static var shared = AsyncActor()
    
    func load() async throws {
        log("AsyncActor load() start")
        try await Task.sleep(nanoseconds: 1_000_000_000)
        log("AsyncActor load() end")
    }
}

Task { try? await AsyncActor.shared.load() }
Task { try? await AsyncActor.shared.load() }


actor SyncActor {
    static var shared = SyncActor()
    
    func load() async throws {
        log("SyncActor load() start")
        sleep(1)
        log("SyncActor load() end")
    }
}

Task { try? await SyncActor.shared.load() }
Task { try? await SyncActor.shared.load() }

输出结果：

2022-10-09 13:45:55 +0000 - AsyncActor load() start
2022-10-09 13:45:55 +0000 - SyncActor load() start
2022-10-09 13:45:55 +0000 - AsyncActor load() start
2022-10-09 13:45:56 +0000 - SyncActor load() end
2022-10-09 13:45:56 +0000 - SyncActor load() start
2022-10-09 13:45:56 +0000 - AsyncActor load() end
2022-10-09 13:45:56 +0000 - AsyncActor load() end
2022-10-09 13:45:57 +0000 - SyncActor load() end

（可在Playground中测试）

使用信号量的异步代码可以正常工作：

struct AsyncSemaphore {
    let semaphore = DispatchSemaphore(value: 1)
    
    static var shared = Self()
    
    func load() async throws {
        log("AsyncSemaphore load() barriers")
        semaphore.wait()
        defer { semaphore.signal() }
        log("AsyncSemaphore load() start")
        try await Task.sleep(nanoseconds: 1_000_000_000)
        log("AsyncSemaphore load() end")
    }
}

Task { try await AsyncSemaphore.shared.load() }
Task { try await AsyncSemaphore.shared.load() }

输出结果：

2022-10-09 13:49:24 +0000 - AsyncSemaphore load() barriers
2022-10-09 13:49:24 +0000 - AsyncSemaphore load() barriers
2022-10-09 13:49:24 +0000 - AsyncSemaphore load() start
2022-10-09 13:49:25 +0000 - AsyncSemaphore load() end
2022-10-09 13:49:25 +0000 - AsyncSemaphore load() start
2022-10-09 13:49:26 +0000 - AsyncSemaphore load() end

但semaphore.wait()会触发Xcode警告：

Instance method 'wait' is unavailable from asynchronous contexts; Await a Task handle instead; this is an error in Swift 6

解决方案：用Actor缓存异步任务实现串行化 #

原来的Actor失效是因为当load()执行到await时，Actor会释放执行权，允许下一个load()调用进入执行。要解决这个问题，核心是在Actor内部维护一个待完成的异步任务引用，让后续调用直接等待已有任务的结果，而非重新发起操作。

基础版：确保串行执行 #

actor SafeLoader {
    static let shared = SafeLoader()
    
    // 缓存正在执行的加载任务
    private var loadingTask: Task<Void, Error>?
    
    func load() async throws {
        // 已有任务则等待其完成
        if let existingTask = loadingTask {
            log("SafeLoader waiting for existing task")
            try await existingTask.value
            return
        }
        
        // 创建新任务并执行
        let task = Task {
            log("SafeLoader load() start")
            try await Task.sleep(nanoseconds: 1_000_000_000)
            log("SafeLoader load() end")
        }
        
        loadingTask = task
        
        defer {
            // 任务完成后清空引用，允许后续重新加载
            loadingTask = nil
        }
        
        try await task.value
    }
}

// 测试调用
Task { try? await SafeLoader.shared.load() }
Task { try? await SafeLoader.shared.load() }

输出结果：

2022-10-09 14:00:00 +0000 - SafeLoader load() start
2022-10-09 14:00:00 +0000 - SafeLoader waiting for existing task
2022-10-09 14:00:01 +0000 - SafeLoader load() end

进阶版：缓存加载结果 #

如果需要彻底避免重复请求，还可以缓存加载后的结果：

actor CachedLoader {
    static let shared = CachedLoader()
    
    private var cachedResult: Data? // 根据实际业务调整结果类型
    private var loadingTask: Task<Data, Error>?
    
    func load() async throws -> Data {
        // 已有缓存直接返回
        if let result = cachedResult {
            log("CachedLoader returning cached result")
            return result
        }
        
        // 已有任务则等待其完成
        if let existingTask = loadingTask {
            log("CachedLoader waiting for existing task")
            return try await existingTask.value
        }
        
        // 创建新任务执行网络请求
        let task = Task {
            log("CachedLoader load() start")
            // 替换为实际网络请求代码
            try await Task.sleep(nanoseconds: 1_000_000_000)
            let mockData = "Loaded Data".data(using: .utf8)!
            log("CachedLoader load() end")
            return mockData
        }
        
        loadingTask = task
        
        defer {
            loadingTask = nil
        }
        
        let result = try await task.value
        cachedResult = result // 缓存结果
        return result
    }
}

逻辑说明 #

• Actor的串行执行特性保证了loadingTask的读写是线程安全的，无需额外同步。
• 当第一个请求发起时，创建任务并保存引用；后续请求直接等待该任务完成，避免并发调用。
• 缓存结果的版本进一步优化，后续调用直接返回缓存值，彻底消除不必要的网络往返。

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