Swift异步序列偶发丢失@Published属性更新问题排查与修复

问题背景 #

测试中出现偶发挂起失败：两个异步序列同时监听并修改同一个@Published属性fire，预期序列能接收所有属性更新以保证测试稳定，但运行约10万次内会出现测试无法完成的情况。即使给类添加@MainActor确保单线程执行，问题仍存在。

成功测试控制台输出：

createStream1(): true
createStream2(): true
createStream1(): false

失败测试控制台输出：

createStream1(): true
createStream2(): true
createStream1(): true

相关代码片段：

final class Tests: XCTestCase {
    @MainActor func test_SUTStream1() async {
        await SUT().toTest()
    }
}

@MainActor
class SUT {
    @Published private(set) var fire = false

    init() {}

    func toTest() async {
        let cancelable = Task {
            await createStream2()
        }
        await createStream1()
        cancelable.cancel()
    }

    private func createStream1() async {
        fire = true

        for await didFire in $fire.values {
            print("\(#function): \(didFire)")
            if didFire == false {
                break
            }
        }
    }

    private func createStream2() async {
        for await didFire in $fire.values {
            print("\(#function): \(didFire)")
            if didFire == true {
                fire = false
                break
            }
        }
    }
}

1. 异步序列偶发丢失@Published属性更新的原因 #

• 冷异步序列的订阅时序问题：$fire.values是冷序列，只有执行for await迭代时才会开始订阅属性变化。失败场景中，createStream1先设置fire = true再启动迭代；createStream2任务启动稍晚，开始订阅时fire已是true，因此收到该初始值并设置fire = false。但此时createStream1的迭代可能因@Published内部投递机制的时序问题，没接收到false的更新，反而重复收到旧的true值，导致循环无法退出。
• 单线程下的异步任务调度竞态：虽然@MainActor保证单线程执行，但异步任务的启动、订阅、事件投递的细微时序差异依然会引发问题。当createStream2设置fire = false时，createStream1的迭代可能还在处理之前的true值，导致后续的false更新被遗漏。
• 任务取消逻辑的依赖问题：toTest方法必须等createStream1完成后才会取消createStream2的任务，一旦createStream1因没收到false更新而挂起，整个测试就会卡死。

2. 问题修复方案 #

方案一：先完成订阅再触发属性变化 #

调整逻辑，确保两个异步序列都完成订阅后再修改fire，避免冷序列订阅滞后导致的更新丢失：

@MainActor
class SUT {
    @Published private(set) var fire = false

    init() {}

    func toTest() async {
        await withTaskGroup(of: Void.self) { group in
            // 同时启动两个序列的订阅
            group.addTask { await self.createStream1() }
            group.addTask { await self.createStream2() }
            
            // 等待一个主队列事件循环，确保订阅完成
            DispatchQueue.main.async {
                self.fire = true
            }
            
            await group.waitForAll()
        }
    }

    private func createStream1() async {
        for await didFire in $fire.values {
            print("\(#function): \(didFire)")
            if didFire == false {
                break
            }
        }
    }

    private func createStream2() async {
        for await didFire in $fire.values {
            print("\(#function): \(didFire)")
            if didFire == true {
                fire = false
                break
            }
        }
    }
}

方案二：用CurrentValueSubject替代@Published #

@Published基于CurrentValueSubject封装，但直接使用CurrentValueSubject能更精确控制事件投递，减少内部机制的不确定性：

import Combine

@MainActor
class SUT {
    private let _fire = CurrentValueSubject<Bool, Never>(false)
    var fire: Bool {
        get { _fire.value }
        set { _fire.send(newValue) }
    }

    init() {}

    func toTest() async {
        let cancelable = Task {
            await createStream2()
        }
        await createStream1()
        cancelable.cancel()
    }

    private func createStream1() async {
        fire = true

        for await didFire in _fire.values {
            print("\(#function): \(didFire)")
            if didFire == false {
                break
            }
        }
    }

    private func createStream2() async {
        for await didFire in _fire.values {
            print("\(#function): \(didFire)")
            if didFire == true {
                fire = false
                break
            }
        }
    }
}

方案三：添加订阅同步等待点 #

在createStream1设置fire = true后，添加短暂等待，确保createStream2有足够时间完成订阅：

private func createStream1() async {
    fire = true
    // 等待1ms，确保其他订阅者完成订阅（可根据实际调整时长）
    try? await Task.sleep(nanoseconds: 1_000_000)

    for await didFire in $fire.values {
        print("\(#function): \(didFire)")
        if didFire == false {
            break
        }
    }
}

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