如何在Swift并发体系中实现异步队列？

我们可以用Actor保障线程安全，结合CheckedContinuation实现无数据时的挂起等待，完全替代DispatchSemaphore的功能，同时完美适配async/await体系。

实现代码 #

actor AsyncQueue<Element> {
    private var elements: [Element] = []
    private var waitingContinuations: [CheckedContinuation<Element, Never>] = []
    
    func push(_ element: Element) {
        elements.append(element)
        
        // 若有等待的任务，唤醒最早发起等待的那个
        if let continuation = waitingContinuations.removeFirst() {
            continuation.resume(returning: element)
        }
    }
    
    func pop() async -> Element {
        // 队列有数据时直接返回
        if let element = elements.popLast() {
            return element
        }
        
        // 无数据时挂起任务，直到被push操作唤醒
        return await withCheckedContinuation { continuation in
            waitingContinuations.append(continuation)
            
            // 处理Task取消：任务取消时从等待列表移除当前continuation，避免内存泄漏
            Task { [weak self] in
                try await Task.sleep(nanoseconds: 0) // 进入取消检查点
                guard Task.isCancelled else { return }
                self?.waitingContinuations.removeAll { $0 === continuation }
            }
        }
    }
}

代码说明 #

• Actor自动保证elements和waitingContinuations的线程安全，无需手动管理锁。
• push添加数据后，会检查是否有等待的continuation，存在则立即唤醒，将新数据返回给对应的pop调用。
• pop先检查队列状态，有数据直接返回；无数据时通过withCheckedContinuation挂起当前任务，等待push操作唤醒。
• 内置Task取消处理逻辑，避免任务取消后continuation残留导致的内存问题。

使用示例 #

let queue = AsyncQueue<String>()

// 启动任务等待获取数据
Task {
    let value = await queue.pop()
    print("拿到数据：\(value)") // 会在push执行后输出：拿到数据：Hello World
}

// 延迟1秒后添加数据
Task {
    try await Task.sleep(nanoseconds: 1_000_000_000)
    await queue.push("Hello World")
}

对比原DispatchSemaphore实现的优势 #

• 完全适配Swift Concurrency体系，无需混用Dispatch框架，避免线程阻塞问题。
• 代码更简洁安全，Actor自动处理线程安全，无需手动维护锁或信号量。
• 原生支持Task取消，逻辑符合现代Swift并发设计理念。

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