如何在Combine的sink内部取消订阅，避免回调重复触发

你遇到的回调触发两次的问题核心来自CurrentValueSubject的特性和订阅执行顺序：

• CurrentValueSubject存在默认初始值，当有新订阅产生时，会同步把当前最新值发送给订阅者，这个发送动作发生在.store(in:)方法执行之前
• 第一次sink闭包被触发时，订阅还未被存入cancellableSet，执行到.doingSomething分支调用logUserOut方法，同步发送.loggedOut状态，触发第二次sink回调
• 两次回调都执行完成后，订阅才会被存入cancellableSet，完全不符合你的预期。

方案一：队列异步调度（最简洁） #

在sink前添加receive(on:)操作符，让所有回调异步派发，保证订阅先存入集合再执行逻辑：

import Combine

enum State {
    case loggedOut
    case doingSomething
}

let aState = CurrentValueSubject<State, Never>(.doingSomething)
private var cancellableSet: Set<AnyCancellable> = []

func logUserOut() {
    cancellableSet.removeAll()
    aState.send(.loggedOut)
}

func doSomethingElse() { }

aState
    // 新增这行，所有回调派发到主队列异步执行，也可以替换为你自己的业务队列
    .receive(on: DispatchQueue.main)
    .sink { newState in
        print("numberOfSubscriptions is: \(cancellableSet.count)")
        switch newState {
        case .loggedOut:
            doSomethingElse()
        case .doingSomething:
            logUserOut()
        }
    }
    .store(in: &cancellableSet)

原理 #

添加receive(on:)后，事件回调会被放到队列的待执行列表里，当前线程会先走完订阅存储的逻辑，再执行回调，此时cancellableSet里已经存在对应的订阅，完全符合你的要求，且不需要用到take/drop操作符。

方案二：标记位控制（无队列依赖） #

如果不想依赖队列调度，可以用标记位控制逻辑执行时机：

import Combine

enum State {
    case loggedOut
    case doingSomething
}

let aState = CurrentValueSubject<State, Never>(.doingSomething)
private var cancellableSet: Set<AnyCancellable> = []
// 新增订阅就绪标记位
private var isSubscriptionReady = false

func logUserOut() {
    cancellableSet.removeAll()
    aState.send(.loggedOut)
}

func doSomethingElse() { }

// 把状态处理逻辑抽为单独方法
func handleState(_ newState: State) {
    print("numberOfSubscriptions is: \(cancellableSet.count)")
    switch newState {
    case .loggedOut:
        doSomethingElse()
    case .doingSomething:
        logUserOut()
    }
}

aState
    .sink { newState in
        // 仅当订阅已经存入集合后才执行业务逻辑
        guard isSubscriptionReady else { return }
        handleState(newState)
    }
    .store(in: &cancellableSet)

// 订阅存入完成后再标记为就绪
isSubscriptionReady = true
// 手动处理当前最新的状态值，补上被跳过的初始值回调
handleState(aState.value)

内容的提问来源于stack exchange，提问作者Dr. Mr. Uncle