用于简单 RESTful 调用的反应式 R2DBC答案

【问题标题】：Reactive R2DBC for simple RESTful calls用于简单 RESTful 调用的反应式 R2DBC
【发布时间】：2021-09-19 14:15:07
【问题描述】：

由于 R2DBC 是反应式且非阻塞的，我想了解在简单的 RESTful CRUD 服务中使用 R2DBC 的好处

假设 Spring Boot 应用程序使用下面的存储库公开 RESTful 服务

public interface CustomerRepository extends ReactiveCrudRepository<Customer, Long> {

    @Query("SELECT * FROM customer WHERE last_name = :lastname")
    Flux<Customer> findByLastName(String lastName);

}

此存储库是从服务调用的，结果需要在服务中转换，然后返回到控制器。

Flux<Customer> customers = repository.findAll();

为了访问完整的客户列表，我需要在 Flux 上调用 blockLast()，这会使其阻塞并破坏使用响应式组件的目的

这是否意味着在这个简单的示例中使用 R2DBC 没有任何好处？我错过了什么吗？

flux 是否只能用于异步订阅，其中 Flux 集合的处理发生在不同的线程中？

【问题讨论】：

标签： project-reactor spring-data-r2dbc r2dbc

【解决方案1】：

为了访问完整的客户列表，我需要在 Flux 上调用 blockLast()，这会使其阻塞并破坏使用响应式组件的目的

如果您想实际获得对List<Customer> 的引用，您只需要调用blockLast() - 但正如您所说，如果您这样做，您将失去所有反应优势。（您真正应该考虑阻止恕我直言的唯一一次是，如果您正在迁移到反应式系统，并且将反应式库放置到位，但尚未准备好使整个系统反应式。）

如果您只想一次访问完整的客户列表，您可能明智地致电collectList() 以获取Mono<List<Customer>。这样你就可以停留在响应式上下文中，但在任何响应式运算符中，你都可以使用整个列表。

这里唯一需要注意的是内存占用 - 如果您只是按原样处理Flux<Customer>，那么您永远不需要将一组它们存储在内存中，所以实际上有多少并不重要是（甚至可以是无限的。）如果您先将其收集到一个列表中，那么所有这些客户都必须同时存储在内存中。

这是否是一个问题取决于您的用例。如果您谈论的是 10 个左右的客户，那完全没有问题。如果您谈论的是数十亿美元，那么这可能不是一个明智的解决方案。

【讨论】：

感谢迈克尔的详细说明。在上面的例子中使用 Mono> 有什么好处，除了代码可以保持在响应式上下文中吗？
同意。如果有百万条记录，那么使用 Flux 是正确的选择，因为记录可以在可用时进行处理，而不是等待建立庞大的列表。这将有助于有效地使用内存。但是我测试了那个例子，发现这种情况下的消费者正在同一个主线程中处理。这和阻塞场景一样好。我在这里发布了示例代码。 stackoverflow.com/questions/68320273/…如果我的理解有误请告诉我
@lives 我不确定我是否理解你的意思——它肯定不等同于阻塞解决方案，因为它使用一个线程。正在处理的线程无关紧要 - 反应式解决方案的要点是您可以在同一个线程上同时执行多个非阻塞操作，从而避免不同线程之间的上下文切换开销。
谢谢迈克。您能否分享任何解释“多个非阻塞操作在同一线程上同时执行”的链接？
@lives 这里的最佳答案将是一个好的开始——本质上这是单线程事件循环的概念：stackoverflow.com/questions/34855352/…

【解决方案2】：

由于该方法返回一个Flux，它返回一个可以在未来某个时间完成的承诺。
调用此方法的线程不应阻塞 - 这就是反应式的全部意义所在。如果你使用.blockLast()，你会让调用线程阻塞。正如迈克尔在这个答案中所说，只有在将反应式和非反应式代码集成在一起时才应该这样做。
这是一个解释异步如何实现的图表。计算是工作（反应，CompletableFuture 等）。希望这能消除您的疑虑；

正如您在图表中看到的，如果您执行.blockLast()，您将失去真正的非阻塞。理想情况下，调用线程应该立即释放，以便它可以做其他工作。

正确编写的响应式应用程序将是一个单链，订阅发生在最上层。整个应用程序应该是一个异步管道。您应该使用BlockHound 之类的应用程序来检查您的线程是否在任何地方阻塞。

“通量只能用于异步订阅，其中通量集合的处理发生在不同的线程中？” - Flux 只不过是一个承诺，它可以在未来一段时间内完成。如图所示，回调将在异步线程上执行。计算结束。您可以使用.publishOn() 切换此线程。

【讨论】：