Spring Reactor 线程模型

Question

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

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

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

考虑这个代码：

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

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

反应式宣言描述了反应式系统，并带来了异步通信和背压。除此之外，它还没有对技术细节做出任何假设。

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

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

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运算符来引入异步行为。

Answer 2

TL; 博士：

1. 这是一个 Project Reactor 的事情，Spring-Webflux 不决定哪个操作在哪个线程上运行。
2. Project Reactor 可以更轻松地判断您在何处跨越了线程边界。此外，没有（显式）同步正在进行；使得引入并发问题变得更加困难。

不，它不是具有不同语法的单线程模型。Project Reactor 尝试尽可能多地使用主线程来避免上下文切换。此外，它还提供特殊运算符，使您可以指定先前操作运行的线程。

例如，这个修改后的示例将在不同的线程上运行；assubscribeOn运算符定义整个链运行在哪个线程池上：

Mono.just("ONE")
    .map(item -> func(" A " + item))
    .map(item -> func(" B " + item))
    .map(item -> func(" C " + item))
    .subscribeOn(Schedulers.elastic())
    .subscribe(item -> {
        System.out.println(Thread.currentThread().getName() + " " + item);
    });

Mono.just("TWO")
    .map(item -> func(" A " + item))
    .map(item -> func(" B " + item))
    .map(item -> func(" C " + item))
    .subscribeOn(Schedulers.elastic())
    .subscribe(item -> {
        System.out.println(Thread.currentThread().getName() + " " + item);
    });

在这种情况下，两个操作都在一个线程上执行elastic-x；不阻塞主线程。每次运行的操作顺序可能会有所不同。