java - 在返回通量数据库实体之前运行异步任务

Question

我有Flux<URL> .如何为每个 URL 发出多个并发无效请求(例如 myWebClient.refresh(URL) )，然后(完成所有请求后)从数据库读取数据并返回 Flux<MyAnyEntity> (例如 repo.findAll() )？

Answer 1

您可以使用 Flux/Mono 运算符实现:

// get the URIs from somewhere
Flux<URI> uris = //...

Flux<MyAnyEntity> entities = uris
                // map each URI to a HTTP client call and do nothing with the response
                .flatMap(uri -> webClient.get().uri(uri).exchange().then())
                // chain that call with a call on your repository
                .thenMany(repo.findAll());

更新:

这段代码自然是异步的、非阻塞的，所以 flatMap 运算符中的所有操作将根据消费者传达的需求并发执行(这就是我们正在谈论的背压) .

如果 Reactive Streams Subscriber request(N) 元素，那么 N 请求可能会并发执行。我不认为这不是您想要直接处理的事情，尽管您可以使用窗口操作符进行微浴操作来影响事情。

在这种情况下使用 .subscribeOn(Schedulers.parallel()) 不会提高并发性 - as stated in the reference documentation你应该只将它用于 CPU 密集型工作。