多次快速调用时移除Combine的AnyCancellable致应用崩溃

问题背景 #

我维护了一个Combine管道，其中pipelinePublisher负责执行多步操作，部分步骤会向statePublisher发送状态更新。管道完成时，我尝试从cancellables集合中手动移除pipelinePublisher，但快速多次调用函数时，偶尔会在移除操作处触发崩溃。

核心代码如下：

func myFunction(_ request: MyRequest) -> AnyPublisher<State, Never> {
    let statePublisher = PassthroughSubject<State, Never>()
    let presentationSubject = CurrentValueSubject<MyRequest, Error>(request)

    var pipelinePublisher: AnyCancellable!

    pipelinePublisher = presentationSubject
      .eraseToAnyPublisher()
      .checkSomething(returningStateTo: statePublisher)
      // 其他操作符...
      .sink(
        receiveCompletion: { [weak self] _ in
          self?.cancellables.remove(pipelinePublisher) // 崩溃触发点
        },
        receiveValue: { _ in }
      )

    pipelinePublisher.store(in: &cancellables)

    return statePublisher
      .receive(on: RunLoop.main)
      .eraseToAnyPublisher()
}

崩溃日志 #

栈追踪1 #

2022-12-21 15:24:51.926131+0000 MyApp[23082:12933690] -[_NSCoreDataTaggedObjectID member:]: unrecognized selector sent to instance 0x8000000000000000
2022-12-21 15:24:51.931941+0000 MyApp[23082:12933690] *** Terminating app due to uncaught exception 'NSInvalidArgumentException', reason: '-[_NSCoreDataTaggedObjectID member:]: unrecognized selector sent to instance 0x8000000000000000'
*** First throw call stack:
(
    0   CoreFoundation                      0x000000018040e7c8 __exceptionPreprocess + 172
    1   libobjc.A.dylib                     0x0000000180051144 objc_exception_throw + 56
    2   CoreFoundation                      0x000000018041d47c +[NSObject(NSObject) instanceMethodSignatureForSelector:] + 0
    3   CoreFoundation                      0x00000001804126c8 ___forwarding___ + 1308
    4   CoreFoundation                      0x0000000180414b4c _CF_forwarding_prep_0 + 92
    5   libswiftCore.dylib                  0x000000018be6ee68 $sSh8_VariantV6removeyxSgxF + 160
    6   MyApp               0x00000001026c6080 $s12MyApp0A0C17myFunctiony7Combine12AnyPublisherVyAA12StateOs5NeverOGAA19MyRequestVFyAE11SubscribersO10CompletionOy_s5Error_pGcfU_ + 440
    7   Combine                             0x000000019baa2a70 $s7Combine11SubscribersO4SinkC7receive10completionyAC10CompletionOy_q_G_tF + 364
    8   Combine                             0x000000019baa2f28 $s7Combine11SubscribersO4SinkCy_xq_GAA10SubscriberA2aGP7receive10completionyAC10CompletionOy_7FailureQzG_tFTW + 20
    9   Combine                             0x000000019bb541cc $s7Combine10PublishersO7FlatMapV5Outer33_E91C3F00A6DFAAFEA2009FAF507AE039LLC7receive10completionyAA11SubscribersO10CompletionOy_7FailureQzG_tF + 1516
    10  Combine                             0x000000019bb55328 $s7Combine10PublishersO7FlatMapV5Outer33_E91C3F00A6DFAAFEA2009FAF507AE039LLCy_xq__qd__GAA10SubscriberA2aJP7receive10completionyAA11SubscribersO10CompletionOy_7FailureQzG_tFTW + 20
    11  Combine                             0x000000019bb53474 $s7Combine10PublishersO7FlatMapV5Outer33_E91C3F00A6DFAAFEA2009FAF507AE039LLC12receiveInner10completion_yAA11SubscribersO10CompletionOy_7FailureQzG_SitF + 1668
    12  Combine                             0x000000019bb52de4 $s7Combine10PublishersO7FlatMapV5Outer33_E91C3F00A6DFAAFEA2009FAF507AE039LLC4SideV7receive10completionyAA11SubscribersO10CompletionOy_7FailureQzG_tF + 20
    13  Combine                             0x000000019bac45ec $s7Combine6FutureC7Conduit33_3AE68DE9BADC00342FC052FEBC7D3BA6LLC7fulfillyys6ResultOyxq_GF + 1056
    14  Combine                             0x000000019bac4960 $s7Combine6FutureC7Conduit33_3AE68DE9BADC00342FC052FEBC7D3BA6LLC6finish10completionyAA11SubscribersO10CompletionOy_q_G_tF + 336
    15  Combine                             0x000000019bac2de4 $s7Combine6FutureC7promise33_3AE68DE9BADC00342FC052FEBC7D3BA6LLyys6ResultOyxq_GFyAA11ConduitBaseCyxq_GXEfU0_ + 156
    16  Combine                             0x000000019bac6b28 $s7Combine6FutureC7promise33_3AE68DE9BADC00342FC052FEBC7D3BA6LLyys6ResultOyxq_GFyAA11ConduitBaseCyxq_GXEfU0_TA + 16
    17  Combine                             0x000000019bae5140 $s7Combine11ConduitListO7forEachyyyAA0B4BaseCyxq_GKXEKF + 212
    18  Combine                             0x000000019bac2bfc $s7Combine6FutureC7promise33_3AE68DE9BADC00342FC052FEBC7D3BA6LLyys6ResultOyxq_GF + 716
    19  Combine                             0x000000019bac6b08 $s7Combine6FutureCyACyxq_Gyys6ResultOyxq_GcccfcyAGcfU_TA + 20
    20  MyApp               0x0000000102541dd8 $s7Combine6FutureC12MyApps5Error_pRs_rlE9operationACyxsAE_pGxyYaKc_tcfcyys6ResultOyxsAE_pGccfU_yyYaYbcfU_TY2_ + 212
    21  MyApp               0x0000000102542705 $s7Combine6FutureC12MyApps5Error_pRs_rlE9operationACyxsAE_pGxyYaKc_tcfcyys6ResultOyxsAE_pGccfU_yyYaYbcfU_TATQ0_ + 1
    22  MyApp               0x000000010242f1a1 $sxIeghHr_xs5Error_pIegHrzo_s8SendableRzs5NeverORs_r0_lTRTQ0_ + 1
    23  MyApp               0x000000010242f749 $sxIeghHr_xs5Error_pIegHrzo_s8SendableRzs5NeverORs_r0_lTRTA.24TQ0_ + 1
    24  libswift_Concurrency.dylib          0x00000001b03bedcd _ZL23completeTaskWithClosurePN5swift12AsyncContextEPNS_10SwiftErrorE + 1
)
libc++abi: terminating with uncaught exception of type NSException

栈追踪2 #

*** Terminating app due to uncaught exception 'NSInvalidArgumentException', reason: '-[__NSCFNumber member:]: unrecognized selector sent to instance 0x8000000000000000'

补充代码（调用链路） #

myFunction通过foo方法调用，相关代码如下：

static func publisher(forParam: String) -> AnyPublisher<State, Never> {
    return Future {
      // 处理逻辑
      return objects
    }
    .flatMap { objects in
      let request = MyRequest(objects)
      return shared.myFunction(request)
    }
    .eraseToAnyPublisher()
}

static func foo(
    param: String,
    handler: ((State) -> Void)? = nil
) {
    var cancellable: AnyCancellable!
    cancellable = publisher(forParam: param)
    .sink(
      receiveCompletion: { _ in
        self.shared.fooItems.cancellables.remove(cancellable) // 此处也会触发相同崩溃
      }, receiveValue: { state in
        handler?(state)
      }
    )

    cancellable.store(in: &shared.fooItems.cancellables)
}

关键约束 #

• foo方法必须保留完成闭包作为API的一部分
• publisher(forParam:)和myFunction(_:)仅由foo调用，但foo会被多处高频调用
• myFunction的管道会因错误触发取消，同时向foo发送状态和完成事件
• 状态可能在无后续完成事件的情况下传递给foo

问题分析 #

崩溃的核心原因是**Set<AnyCancellable>不是线程安全的容器**：

1. 管道的异步操作可能在后台线程触发receiveCompletion回调，此时调用remove操作会直接修改Set
2. 主线程同时执行store(in:)插入新的AnyCancellable，多线程并发修改会导致Set内部哈希表损坏
3. 损坏后的Set会返回错误的内存地址（被CoreData对象ID、NSNumber等其他实例占用），调用member:方法时触发未识别选择器崩溃

修复方案 #

方案1：统一线程访问容器 #

确保所有对cancellables的操作都在同一个线程（推荐主线程）执行：

修改myFunction代码

func myFunction(_ request: MyRequest) -> AnyPublisher<State, Never> {
    let statePublisher = PassthroughSubject<State, Never>()
    let presentationSubject = CurrentValueSubject<MyRequest, Error>(request)

    // 替换隐式解包为let，避免悬垂引用
    let cancellable = presentationSubject
      .eraseToAnyPublisher()
      .checkSomething(returningStateTo: statePublisher)
      // 其他操作符...
      .sink(
        receiveCompletion: { [weak self, cancellable] _ in
          // 切换到主线程执行移除操作
          DispatchQueue.main.async {
            self?.cancellables.remove(cancellable)
          }
          // 可选：给状态发布者发送完成事件
          statePublisher.send(completion: .finished)
        },
        receiveValue: { _ in }
      )

    // 确保插入操作也在主线程执行（如果当前函数可能在后台调用）
    DispatchQueue.main.async {
        self.cancellables.insert(cancellable)
    }

    return statePublisher
      .receive(on: RunLoop.main)
      .eraseToAnyPublisher()
}

修改foo代码

static func foo(
    param: String,
    handler: ((State) -> Void)? = nil
) {
    let cancellable = publisher(forParam: param)
    .sink(
      receiveCompletion: { [weak cancellable] _ in
        DispatchQueue.main.async {
            guard let cancellable = cancellable else { return }
            self.shared.fooItems.cancellables.remove(cancellable)
        }
      }, receiveValue: { state in
        handler?(state)
      }
    )

    DispatchQueue.main.async {
        self.shared.fooItems.cancellables.insert(cancellable)
    }
}

方案2：自定义线程安全容器 #

如果必须在多线程环境下操作，可以封装一个线程安全的Set包装类：

final class ThreadSafeCancellableSet {
    private let lock = NSLock()
    private var cancellables = Set<AnyCancellable>()

    func insert(_ cancellable: AnyCancellable) {
        lock.lock()
        defer { lock.unlock() }
        cancellables.insert(cancellable)
    }

    func remove(_ cancellable: AnyCancellable) {
        lock.lock()
        defer { lock.unlock() }
        cancellables.remove(cancellable)
    }

    func cancelAll() {
        lock.lock()
        defer { lock.unlock() }
        cancellables.forEach { $0.cancel() }
        cancellables.removeAll()
    }
}

之后将原有的Set<AnyCancellable>替换为ThreadSafeCancellableSet，调用insert和remove时无需手动切换线程。

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