如何防止 PassthroughSubject 在并发上游期货完成之前杀死 .sink？

Question

我有一个 PassthroughSubject，它发送 30 个整数，后跟一条完成消息。

从对象那里收到这些数字后，我生成了一个休眠一秒钟的未来，并以输入数字 * 2 结束。

我使用 .receiveOn 来确保期货同时运行，但这意味着完成消息也会通过链同时传播 并在所有期货完成之前结束接收器。

那里的任何 RxSwift/Combine 向导都知道我可以如何做到这一点，因此完成消息的接收会因期货完成而延迟？

这是一个实现所描述行为的游乐场：

import Foundation
import Combine
import PlaygroundSupport

/// Setting up the playground
PlaygroundPage.current.needsIndefiniteExecution = true

/// Injects numbers 0-30 into combine message stream, and then sends a finish.
func publishNumbers(to subject: PassthroughSubject<Int, Error>) {
    (0..<30).forEach {
        subject.send($0)
    }
    subject.send(completion: .finished)
}
/// Delays for one secont, and completes the future by doubling the input.
func delayAndDoubleNumber(_ int: Int) -> Future<Int, Error> {
    return Future<Int, Error> { complete in
        sleep(1)
        complete(.success(int * 2))
    }
}

// Properties involved in Combine processing chain.
let numbersSubject = PassthroughSubject<Int, Error>()
let processingQueue = DispatchQueue.global(qos: .userInitiated)


// Combine processing chain
numbersSubject
    .receive(on: processingQueue) //Comment this line to observe that all futures finish, and are collected before the finish message kills the sink.
    .flatMap { number in
        return delayAndDoubleNumber(number)
    }
    .collect(4)
    .sink(receiveCompletion: { completion in
        print("Complete: \(completion)")
    }, receiveValue: { value in
        print("Received Value: \(value)")
    })

publishNumbers(to: numbersSubject)

Answer 1

从 Xcode 11 beta 3 开始，您不能将并发队列与 Combine 一起使用。 Xcode 11 GM 应该可以。

Philippe Hausler 是一名 Apple 工程师，在 Combine 工作。他在the official Swift forum上说了如下：

另外值得注意的是，用作调度程序的DispatchQueue 必须始终是串行的，以遵守联合运营商的合同。

Then later他这样说：

因此，在这里跟进，关于下游事件的传播方式有一些变化。即使 DispatchQueue 是并发的，或者 OperationQueue 不是 maxConcurrentOperations 为 1 的限制，或者任何有效的调度程序都是并发的，我们现在也能够满足 1.03 的约束；我们将始终在请求的调度程序上为.receive(on:) 发送序列化事件。我们稍微偏离规范的另一个警告是，我们世界中的上游事件（例如cancel() 和request(_:)）可以同时发生。话虽如此，我们确实以线程安全的方式处理它们。

您可以通过从Future 的闭包中分派到并发队列，然后返回主队列来使您的并发在 Xcode 11 beta 3 中工作：

import Foundation
import Combine
import PlaygroundSupport

PlaygroundPage.current.needsIndefiniteExecution = true

func delayAndDoubleNumber(_ int: Int) -> Future<Int, Never> {
    return Future<Int, Never> { complete in
        DispatchQueue.global(qos: .userInitiated).async {
            sleep(1)
            DispatchQueue.main.async {
                complete(.success(int * 2))
            }
        }
    }
}

let subject = PassthroughSubject<Int, Never>()

subject
    .flatMap { delayAndDoubleNumber($0) }
    .collect(4)
    .sink(
        receiveCompletion: { print("Complete: \($0)") },
        receiveValue: { print("Received Value: \($0)") })

let canceller = (0 ..< 30).publisher().subscribe(subject)

Answer 2

免责声明，这可能是对文档的错误解释，但我认为您应该使用 subscribe(on:) 运算符而不是 receive(on:)。

Apple Docs:

与影响下游消息的receive(on:options:)相比，subscribe(on:)改变了上游消息的执行上下文。

我对此的解释是，如果您希望在您的队列中发出来自numbersSubject 的事件，您可以使用subscribe(on:)，例如：

numbersSubject
    .flatMap { number in
        return delayAndDoubleNumber(number)
    }
    .collect(4)
    .subscribe(on: processingQueue)
    .receive(on: RunLoop.main)
    .sink(receiveCompletion: { completion in
        print("Complete: \(completion)")
    }, receiveValue: { value in
        print("Received Value: \(value)")
    })