Swift Concurrency的Task是否按顺序执行？Combine与Async Sequence疑问

问题描述 #

找不到相关文档说明Swift Concurrency的Task是并发执行，还是在某个隐形队列中按顺序执行。以下是App中遇到的、可在Playground运行的简化问题：

import UIKit
import Foundation
import Combine

struct Info {
    var id: String
    var value: Int
}

class DataStore {
        // 模拟Core Data存储操作
    func store(info: Info, id: String) {
        print("    store \(info)")
        let start = CACurrentMediaTime()
        while CACurrentMediaTime() - start < 2 { }  
    }
}

let dataStore = DataStore()
let subj = PassthroughSubject<Info, Never>()
let cancel = subj.sink { info in
    print("Start task for \(info)")
    Task {
        print("  start \(info)")
        dataStore.store(info: info, id: info.id)
        print("  finish: \(info)")
    }
}

subj.send(Info(id: "A", value: 1))
subj.send(Info(id: "A", value: 2))
subj.send(Info(id: "A", value: 3))
subj.send(Info(id: "A", value: 4))

let queueA = DispatchQueue(label: "A", attributes: .concurrent)
let queueB = DispatchQueue(label: "B", attributes: .concurrent)

queueA.async {
    subj.send(Info(id: "A", value: 1))
    subj.send(Info(id: "A", value: 2))
    subj.send(Info(id: "A", value: 3))
    subj.send(Info(id: "A", value: 4))
}

queueB.async {
    subj.send(Info(id: "B", value: 1))
    subj.send(Info(id: "B", value: 2))
    subj.send(Info(id: "B", value: 3))
    subj.send(Info(id: "B", value: 4))
}

queueA.async {
    subj.send(Info(id: "A", value: 1))
    subj.send(Info(id: "A", value: 2))
    subj.send(Info(id: "A", value: 3))
    subj.send(Info(id: "A", value: 4))
}

queueB.async {
    subj.send(Info(id: "B", value: 1))
    subj.send(Info(id: "B", value: 2))
    subj.send(Info(id: "B", value: 3))
    subj.send(Info(id: "B", value: 4))
}

// 注意：一个闭包总是在前一个完成后才启动

实际运行发现闭包总是在前一个完成后才启动，不确定这是PassthroughSubject的特性还是Publisher的其他机制。App存在旧Combine代码与新async-await代码对接的场景，同时想了解换成Async Sequence是否会有差异。

核心解答 #

1. Swift Task的本质特性 #

默认创建的Task会被调度到全局并发执行器（对应GCD的全局并发队列），本身是支持并发执行的。你看到的顺序执行现象，和Combine的sink闭包调度逻辑有关，不是Task的固有特性。

2. Combine PassthroughSubject的调度逻辑 #

PassthroughSubject默认会在调用send的线程/队列上同步执行sink闭包。比如你在queueA.async里连续调用send，这些send会在queueA上同步触发sink闭包——虽然sink里创建Task的操作是瞬间完成的，但你的store方法用了阻塞线程的2秒循环，系统调度线程可能暂时没有空闲资源，导致Task看起来是顺序执行的。

如果要验证Task的并发能力，可以修改代码让Task明确指定后台优先级，或者把阻塞循环换成真正的异步休眠：

// 方式1：指定后台优先级Task
Task(priority: .background) {
    print("  start \(info)")
    dataStore.store(info: info, id: info.id)
    print("  finish: \(info)")
}

// 方式2：把store改成异步方法
func store(info: Info, id: String) async {
    print("    store \(info)")
    try await Task.sleep(nanoseconds: 2_000_000_000)
}

这样就能看到Task并发执行的效果。

3. Combine与Async Sequence的差异 #

换成Async Sequence后，核心差异在调度和订阅模式：

• Async Sequence的for await循环默认会在当前任务上下文执行，每次迭代会等待前一个异步操作完成，除非你为每个元素显式创建新Task。
• 不管是Combine还是Async Sequence，要实现「同一ID串行、不同ID并发」的需求，都需要额外的调度控制——比如为每个ID维护串行队列或Actor。

4. 场景解决方案（按ID排队执行） #

如果你的需求是同一ID的任务串行执行，不同ID的任务并发，可以用Actor字典或者GCD串行队列字典来实现：

方案1：用Actor实现串行化

// 单个ID的串行存储Actor
actor SerialStore {
    func store(info: Info) {
        print("    store \(info)")
        let start = CACurrentMediaTime()
        while CACurrentMediaTime() - start < 2 { }
    }
}

class DataStore {
    private var actors: [String: SerialStore] = [:]
    private let lock = NSLock()
    
    func store(info: Info, id: String) async {
        // 线程安全获取对应ID的Actor
        lock.lock()
        let actor = actors[id] ?? SerialStore()
        actors[id] = actor
        lock.unlock()
        
        await actor.store(info: info)
    }
}

方案2：用GCD串行队列实现

class DataStore {
    private var queues: [String: DispatchQueue] = [:]
    private let lock = NSLock()
    
    func store(info: Info, id: String) async {
        // 线程安全获取对应ID的串行队列
        lock.lock()
        let queue = queues[id] ?? DispatchQueue(label: "store.\(id)")
        queues[id] = queue
        lock.unlock()
        
        await queue.async {
            print("    store \(info)")
            let start = CACurrentMediaTime()
            while CACurrentMediaTime() - start < 2 { }
        }
    }
}

两种方案都能保证同一ID的任务串行执行，不同ID的任务并发处理。

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