使自定义发布者在 Swift Combine 上的不同 DispatchQueue 上运行

Question

我创建了一个函数，该函数使用以下代码在 Swift Combine 中返回一个自定义发布者：

func customPubliher() -> AnyPublisher<Bool, Never> {
    return Future<Bool, Never> { promise in
        promise(.success(true))
    }.eraseToAnyPublisher()
}

然后我使用以下代码订阅了这个发布者：

customPublisher()
    .subscribe(on: DispatchQueue.global())
    .map { _ in
        print(Thread.isMainThread)
    }
    .sink(receiveCompletion: { _ in }, receiveValue: { value in
        // Do something with the value received
    }).store(in: &disposables)

但是，即使我说行.subscribe(on: DispatchQueue.global())，当我做了申购，代码不是在不同的队列（已执行print的.map产出如此）。

但是，例如，如果我将自定义发布者替换为一个内置的 Combine 发布者Just()（见下文），则相同的代码将在不同的队列上正常执行：

Just(true)
    .subscribe(on: DispatchQueue.global())
    .map { _ in
        print(Thread.isMainThread)
    }
    .sink(receiveCompletion: { _ in }, receiveValue: { value in
        // Do something with the value received
    }).store(in: &disposables)

在.map上面输出虚假的代码。

使用自定义发布器时，我做错了什么？我希望它在不同的队列上运行，就像Just()发布者一样。

Answer 1

在我对你的代码的测试中，我得到了false. 实际上DispatchQueue与某些特定线程没有一对一的关系，它是一个执行队列，并指定DispatchQueue.global()您要求系统选择一些空闲队列以默认优先级执行您的任务。因此，由系统决定在哪个队列和哪个线程中执行您的任务。

如果您有意将其强制进入后台，请使用

.subscribe(on: DispatchQueue.global(qos: .background))

Answer 2

你想要receive(on:)，没有subscribe(on:)。从文档：

您可以使用 receive(on:options:) 运算符来接收特定调度程序上的结果，例如在主运行循环上执行 UI 工作。与 subscribe(on:options:) 影响上游消息相反，receive(on:options:) 改变下游消息的执行上下文。

容易混淆和忘记receive，因为subscribe 听起来像你想要的，但通常不是。（subscribe是关于订阅/请求/取消的机制；我认为我还没有遇到需要它的情况。）

let sub = Just(true)
    .receive(on: DispatchQueue.global(qos: .background))
    .map { _ in
        print("map: \(Thread.current.qualityOfService == .background)")
    }
    // Back to the main thread (without this, we're still on the background thread):
    .receive(on: DispatchQueue.main)
    .sink { _ in
        print("sink: \(Thread.isMainThread)")
    }