Reactor 之 subcribeOn vs publishOn
Reactor 之 subcribeOn vs publishOn
我们使用 subscribeOn 和 publishOn 操作符在响应链中切换执行上下文(Reactor 中叫 Scheduler)。
上一篇文章中，我们说到 Reactor 默认行为是执行订阅的同一线程将用于整个管道执行。如果要切换执行线程怎么办？可以使用 publishOn 和 SubscribeOn
让我们看个简单的例子：
class ReactiveJavaTutorial {
public static void main(String[] args) {
Flux<String> cities = Flux.just("New York", "London", "Paris", "Amsterdam")
.map(String::toUpperCase)
.filter(cityName -> cityName.length() <= 8)
.map(cityName -> cityName.concat(" City"))
.log();
cities.subscribe();
}输出：
17:39:41.693 [main] DEBUG reactor.util.Loggers - Using Slf4j logging framework
17:39:41.712 [main] INFO reactor.Flux.MapFuseable.1 - | onSubscribe([Fuseable] FluxMapFuseable.MapFuseableSubscriber)
17:39:41.714 [main] INFO reactor.Flux.MapFuseable.1 - | request(unbounded)
17:39:41.715 [main] INFO reactor.Flux.MapFuseable.1 - | onNext(NEW YORK City)
17:39:41.715 [main] INFO reactor.Flux.MapFuseable.1 - | onNext(LONDON City)
17:39:41.715 [main] INFO reactor.Flux.MapFuseable.1 - | onNext(PARIS City)
17:39:41.716 [main] INFO reactor.Flux.MapFuseable.1 - | onComplete()中括号中的就是线程名称，在这个例子中，都是 main。可以看到整个管道执行器中都是使用的 main 线程。
有的时候，我们可能想告诉 Reactor 别在整个管道中使用同一个线程。我可以使用 subscribeOn() 和 publishOn() 方法达到效果。
subscribeOn() 方法
subscribeOn() 方法适用于订阅过程。我们可以把它放在响应链条中的任意位置。它接收 Scheduler 参数，且在提供的线程池中选择线程执行。
在下面的例子中，我们使用有界弹性线程池（Schedulers.boundElastic()）。
@Test
public void testSubscribeThread() {
Flux<String> cities = Flux.just("New York", "London", "Paris", "Amsterdam")
.subscribeOn(Schedulers.boundedElastic())
.map(String::toUpperCase)
.filter(cityName -> cityName.length() <= 8)
.map(cityName -> cityName.concat(" City"))
.map(TestCase::concat)
.map(TestCase::stringToUpperCase)
.log();
// cities.subscribe();
System.out.println(cities.blockFirst());
}PS: 原文提供的case，没有输出，简单修改了一下。
输出：
20:07:53.517 [main] DEBUG reactor.util.Loggers - Using Slf4j logging framework
20:07:53.558 [main] INFO reactor.Flux.Map.1 - onSubscribe(FluxMap.MapSubscriber)
20:07:53.560 [main] INFO reactor.Flux.Map.1 - request(unbounded)
concat: boundedElastic-1
stringToUpperCase: boundedElastic-1
20:07:53.564 [boundedElastic-1] INFO reactor.Flux.Map.1 - onNext(NEW YORK CITY CITY)
20:07:53.565 [boundedElastic-1] INFO reactor.Flux.Map.1 - cancel()
NEW YORK CITY CITY可以看到 main 线程开始订阅，但是被切换成 boundedElastic-1 线程。我们提供了一个 Scheduler （Schedulers.boundedElastic()），然后这个线程池中的一个线程被选中来替换 main 线程。
publishOn() 方法
publishOn() 方法跟 subscribeOn() 很类似，但是有一个主要区别。
来看个例子：
class ReactiveJavaTutorial {
public static void main(String[] args) {
Flux.just("New York", "London", "Paris", "Amsterdam")
.map(ReactiveJavaTutorial::stringToUpperCase)
.publishOn(Schedulers.boundedElastic())
.map(ReactiveJavaTutorial::concat)
.subscribe();
}
private static String stringToUpperCase(String name) {
System.out.println("stringToUpperCase: " + Thread.currentThread().getName());
return name.toUpperCase();
}
private static String concat(String name) {
System.out.println("concat: " + Thread.currentThread().getName());
return name.concat(" City");
}
}这里，我们在两个 map 操作中放一个 publishOn()。我们来看输出：
stringToUpperCase: main
stringToUpperCase: main
stringToUpperCase: main
concat: boundedElastic-1
concat: boundedElastic-1
concat: boundedElastic-1可以看到，所有在 publishOn 操作之前的都是 main 线程执行，所有 publishOn 之后的都是 boundedElastic-1 执行。这是因为 publishOn 充当任何其他操作符。它从上游接收信号，并在关联的 Scheduler 上对一个 worker 执行回调时向下游重播。
这就是 publishOn 和 subscribeOn() 的主要区别。无论我们把 subscribeOn() 放在哪里，它提供的 Scheduler 都会应用到整条响应链。
腾讯云开发者
