10

每当我同时使用 subscribeOn 和 publishOn 时，都不会打印任何内容。如果我只使用一个，它将打印。如果我使用 subscribeOn(Schedulers.immediate()) 或弹性它可以工作。知道为什么吗？

据我了解，publishOn 会影响它发布的线程并订阅订阅者运行的线程。你能指出我正确的方向吗？

fun test() {
        val testPublisher = EmitterProcessor.create<String>().connect()
        testPublisher
                .publishOn(Schedulers.elastic())
                .map { it ->
                    println("map on ${Thread.currentThread().name}")
                    it
                }
                .subscribeOn(Schedulers.parallel())  
                .subscribe { println("subscribe on ${Thread.currentThread().name}") }
        testPublisher.onNext("a")
        testPublisher.onNext("b")
        testPublisher.onNext("c")
        Thread.sleep(5000)
        println("---")
    }
4

1 回答 1

27

subscribeOn而是影响订阅发生的位置。即触发源发射元素的初始事件。另一方面，Subscriber的钩子受到链条中最近的影响（很像您的）。onNextpublishOnmap

但是EmitterProcessor，和大多数Processors 一样，它更先进，可以做一些偷窃工作。我不确定为什么您的案例中没有打印任何内容（您的示例转换为 Java 可以在我的机器上运行），但我敢打赌它与那个处理器有关。

此代码将更好地演示subscribeOnvs publishOn：

Flux.just("a", "b", "c") //this is where subscription triggers data production
        //this is influenced by subscribeOn
        .doOnNext(v -> System.out.println("before publishOn: " + Thread.currentThread().getName()))
        .publishOn(Schedulers.elastic())
        //the rest is influenced by publishOn
        .doOnNext(v -> System.out.println("after publishOn: " + Thread.currentThread().getName()))
        .subscribeOn(Schedulers.parallel())
        .subscribe(v -> System.out.println("received " + v + " on " + Thread.currentThread().getName()));
    Thread.sleep(5000);

这打印出来：

before publishOn: parallel-1
before publishOn: parallel-1
before publishOn: parallel-1
after publishOn: elastic-2
received a on elastic-2
after publishOn: elastic-2
received b on elastic-2
after publishOn: elastic-2
received c on elastic-2
于 2017-01-30T17:10:56.643 回答