使用Combine在并发队列receive on时为何无法接收全部值？

问题复现代码 #

let queue = DispatchQueue.global() // Or OperationQueue()
let subscription = (1...10).publisher
    .receive(on: queue)
    .sink { value in
        print("Received \(value)")
    }

现象 #

调试控制台输出每次都不相同，且无法获取全部10个值，例如：

Received 1
Received 2
Received 3
Received 7
Received 6

原因 #

核心问题是订阅对象subscription的生命周期过短。

当这段代码在局部作用域（比如某个函数内部）执行时，代码执行完毕后subscription会立即被销毁。Combine的订阅机制中，一旦AnyCancellable类型的订阅对象被释放，订阅会立刻被取消，publisher会停止发送剩余事件。

虽然(1...10).publisher会同步发送所有10个值，但receive(on: queue)会把每个值的处理任务调度到全局并发队列。由于并发队列的调度存在延迟，当subscription被销毁时，还有部分任务尚未被执行，这部分对应的数值就不会被打印出来——这也是每次丢失的数值不一样的原因，因为并发队列的任务调度顺序和时机是不确定的。

解决方案 #

• 如果是在UI项目（比如iOS的ViewController）中，把subscription声明为类的实例属性，延长其生命周期，确保所有事件处理完成前订阅不会被取消：

  class ViewController: UIViewController {
      private var subscription: AnyCancellable?
      
      override func viewDidLoad() {
          super.viewDidLoad()
          let queue = DispatchQueue.global()
          subscription = (1...10).publisher
              .receive(on: queue)
              .sink { value in
                  print("Received \(value)")
              }
      }
  }
  

• 如果是命令行工具，需要让程序保持运行，等待所有事件处理完成，比如添加RunLoop运行代码：

  let queue = DispatchQueue.global()
  let subscription = (1...10).publisher
      .receive(on: queue)
      .sink(receiveCompletion: { _ in
          RunLoop.main.stop()
      }, receiveValue: { value in
          print("Received \(value)")
      })
  RunLoop.main.run()
  

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