Project Reactor 3中的发布与订阅

Question

我使用publishOn和subscribeOn的流量相同，如下所示：

    System.out.println("*********Calling Concurrency************");
    List<Integer> elements = new ArrayList<>();
    Flux.just(1, 2, 3, 4)
      .map(i -> i * 2)
      .log()
      .publishOn(Schedulers.elastic())
      .subscribeOn(Schedulers.parallel())
      .subscribe(elements::add);
    System.out.println("-------------------------------------");

虽然，当我使用两者时，日志中不会打印任何内容。但是当我只使用publishOn时，我得到了以下信息日志：

*********Calling Concurrency************
[info] | onSubscribe([Synchronous Fuseable] FluxArray.ArraySubscription)
[info] | request(256)
[info] | onNext(1)
[info] | onNext(2)
[info] | onNext(3)
[info] | onNext(4)
[info] | onComplete()
-------------------------------------

publishOn比Subscribeon更受推荐吗？或者它比subscribeon有更多的偏好？两者之间的区别是什么，什么时候使用哪个？

Answer 1

我花了一些时间来理解它，可能是因为publishon通常是在subscribeon之前解释的，这里有一个更简单的外行解释。

subscribeOn意味着在指定的调度器工作线程（其他线程）上运行初始源发射，例如subscribe()、onSubscribe()和request()，对于任何后续操作也是如此，例如onnext/onerror/oncomplete、map等，无论subscribeOn（）的位置如何，都会发生这种行为

如果在fluent调用中没有执行任何publishon，那么就这样，所有内容都将在这样的线程上运行。

但是，如果在中间调用publishon()，那么任何后续的操作符调用都将在提供的调度器工作程序上运行到这样的publishon()。

这里有一个例子

java prettyprint-override">Consumer<Integer> consumer = s -> System.out.println(s + " : " + Thread.currentThread().getName());

Flux.range(1, 5)
        .doOnNext(consumer)
        .map(i -> {
          System.out.println("Inside map the thread is " + Thread.currentThread().getName());
          return i * 10;
        })
        .publishOn(Schedulers.newElastic("First_PublishOn()_thread"))
        .doOnNext(consumer)
        .publishOn(Schedulers.newElastic("Second_PublishOn()_thread"))
        .doOnNext(consumer)
        .subscribeOn(Schedulers.newElastic("subscribeOn_thread"))
        .subscribe();

结果会是

1 : subscribeOn_thread-4
Inside map the thread is subscribeOn_thread-4
2 : subscribeOn_thread-4
Inside map the thread is subscribeOn_thread-4
10 : First_PublishOn()_thread-6
3 : subscribeOn_thread-4
Inside map the thread is subscribeOn_thread-4
20 : First_PublishOn()_thread-6
4 : subscribeOn_thread-4
10 : Second_PublishOn()_thread-5
30 : First_PublishOn()_thread-6
20 : Second_PublishOn()_thread-5
Inside map the thread is subscribeOn_thread-4
30 : Second_PublishOn()_thread-5
5 : subscribeOn_thread-4
40 : First_PublishOn()_thread-6
Inside map the thread is subscribeOn_thread-4
40 : Second_PublishOn()_thread-5
50 : First_PublishOn()_thread-6
50 : Second_PublishOn()_thread-5

正如您所看到的，第一个doonnext()和下面的map()在名为subscribeon_thread的线程上运行，直到调用任何publishon()，然后任何后续调用都将在为该publishon()提供的调度程序上运行，并且任何后续调用都将再次发生，直到任何人调用另一个publishon()。