Using both publishOn and subscribeOn on a flux res

问题:

Whenever i use both subscribeOn and publishOn nothing is printed. If I use only one it will print. If I use subscribeOn(Schedulers.immediate()) or elastic it works. Any ideea why that is?

It was my understanding that publishOn affects on what thread it gets published and subscribe on on what thread the subscriber runs. Could you please point me in the right direction?

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("---")
    }

回答1:

subscribeOn rather influence where the subscription occurs. That is, the initial event that triggers the source to emit elements. The Subscriber 's onNext hook on the other hand is influenced by the closest publishOn up in the chain (much like your map).

But EmitterProcessor, like most Processors, is more advanced and can do some work stealing. I'm unsure why you don't get anything printed in your case (your sample converted to Java works on my machine), but I bet it has something to do with that Processor.

This code would better demonstrate subscribeOn vs 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);

This prints out:

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