java - Spring Reactor 线程模型

Question

Spring Webflux (v 2.0.1.RELEASE) 的新手警报。

我想将 Spring Webflux 用于后端(Webless)应用程序，以处理来自 JMS 监听器的大量数据。

我的理解是 Spring Webflux 提供了一个]非阻塞/异步并发模型。但是，我遇到了一个需要帮助的基本问题。作为免责声明，响应式(Reactive)编程的整个概念对我来说是非常新的，我仍然处于这种范式转变的过程中。

考虑这个代码:

Mono.just("ONE")
.map(item -> func(" A " + item))
.map(item -> func(" B " + item))
.map(item -> func(" C " + item))
.subscribe(System.out::println);

Mono.just("TWO")
.map(item -> func(" A " + item))
.map(item -> func(" B " + item))
.map(item -> func(" C " + item))
.subscribe(System.out::println);

我从文档中了解到，在调用“订阅”函数之前，事件处理链不会发生任何事情。

但是在内部，spring 是否使用(如果它愿意)为“map”函数内的每个函数异步使用单独的线程？如果 spring 为这些链使用“单一”线程，那么这里的真正目的是什么？它不是基于不同语法的阻塞和单线程模型吗？

我观察到代码总是按顺序运行并使用相同的线程。 spring webflux的线程模型是什么？

Answer 1

响应式编程是一种编程范式，因此它不会对技术实现做出任何假设。

The reactive manifesto描述了响应式(Reactive)系统并带来了异步通信和背压。除此之外，它还没有对技术细节做出任何假设。

Spring Reactor 是 Webflux 的基础，它是一个库，可让您轻松构建响应式(Reactive)系统并遵循响应式(Reactive)编程范式。

流使用的线程取决于发布者。默认是使用当前线程。如果发布者是同步的，则在没有任何干预的情况下，流不能是异步的。如果发布者阻止，则流将被阻止。但请看下面的例子:

Flux.interval(Duration.ofMillis(100))
    .take(2)
    .subscribe(i -> System.out.println(Thread.currentThread().getName()));

Flux.interval在另一个线程上发布，因此链在另一个线程中异步运行。

让我们再看一个例子:

Scheduler scheduler = Schedulers.newElastic("foo");

Flux<Integer> flux = Flux.just(1, 2)
    .subscribeOn(scheduler);

flux.subscribe(i -> System.out.println(Thread.currentThread().getName()));
flux.subscribe(i -> System.out.println(Thread.currentThread().getName()));

您会注意到每个订阅者都在自己的线程上运行(尽管来自同一个线程池)。 publishOn运算符类似。

如果您订阅了发布者，您可以使用相同的编程范式，无论它是同步的还是异步的。并且您始终可以通过添加 subscribeOn 来引入异步行为。或 publishOn运算符(operator)。