我们使用 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
都会应用到整条响应链。
subscribeOn and publishOn operators in Project Reactor