java - webflux和reactor的正确使用方法是什么

Question

我正在学习 webflux 和 Reactor。得到以下三种测试方法。 “documentOperations.findById”和“documentOperations.delete”是两个数据库操作。我知道 test1 很糟糕，因为两个数据库操作被放置在一个异步方法中。我的问题是:
test2 和 test3 对系统性能的影响相同吗？或者换句话说，哪一个更好？

private Mono<ServerResponse> test1(ServerRequest request, Contexts contexts) {
    return request.body(bodyDocumentExtractor)
            .flatMap(doc -> {
                Document document = documentOperations.findById(doc.getId());
                documentOperations.delete(document.getId());
                return ServerResponse.noContent().build();
            });
}

private Mono<ServerResponse> test2(ServerRequest request, Contexts contexts) {
    return request.body(bodyDocumentExtractor)
            .flatMap(doc -> {
                return Mono.just(documentOperations.findById(doc.getId()))
                        .flatMap(document -> {
                            documentOperations.delete(document.getId());
                            return ServerResponse.noContent().build();
                        });
            });
}

private Mono<ServerResponse> test3(ServerRequest request, Contexts contexts) {
    return request.body(bodyDocumentExtractor)
            .flatMap(doc -> {
                return Mono.just(documentOperations.findById(doc.getId()));
            }).flatMap(document -> {
                documentOperations.delete(document.getId());
                return ServerResponse.noContent().build();
            });
}

Answer 1

上面的例子都不好。所有数据库调用都会返回具体类型，这意味着它们都是阻塞调用。

// returns the concrete type
// thread does the call, needs to wait until we get the value (document)
Document document = documentOperations.findById("1");

如果它是非阻塞的，则返回 Mono<T>或Flux<T> .

// Returns a mono, so we know it's not blocking. 
// We can chain on actions with for example flatMap etc.
Mono<Document> document = documentOperations.findById("1");

如果您必须使用阻塞数据库，例如 Oracle 数据库等。您需要将其放置在其整个线程上，以便它不会阻塞任何主工作线程。这可以通过调度程序来完成。因此，在此示例中，当客户端订阅时，它将被放置在单独的线程上。

Mono<Document> document = Mono.fromCallable(() -> documentOperations.findById("1"))
    .subscribeOn(Schedulers.boundedElastic());;

举个例子:

private Mono<ServerResponse> test3(ServerRequest request, Contexts contexts) {
    return request.body(bodyDocumentExtractor)
        .flatMap(doc -> Mono.fromCallable(() -> documentOperations.findById(doc.getId()))
        .flatMap(document -> Mono.fromCallable(() -> documentOperations.delete(document.getId()))
                .then(ServerResponse.noContent().build());
        ).subscribeOn(Schedulers.boundedElastic());
}

Reactor documentation - Wrap blocking calls