在使用 Quarkus Hibernate Reactive 和 quarkus-reactive-client 时如何使用一些阻塞方法？

Question

我正在尝试调用反应式休息，但在执行此操作时，我必须在其中调用另一个请求并创建几个文件夹，但 quarkus-hibernate-reactive 显示以下错误，最佳实践是什么？

Error:

Caused by: java.lang.IllegalStateException: The current thread cannot be blocked: vert.x-eventloop-thread-2
    at io.smallrye.mutiny.operators.UniBlockingAwait.await(UniBlockingAwait.java:29)

我的代码如下：

@POST
@javax.ws.rs.Path("/add") 
@Produces({MediaType.APPLICATION_JSON, MediaType.TEXT_PLAIN})
@Consumes({ MediaType.APPLICATION_JSON, MediaType.TEXT_PLAIN })
public Uni<Response> addUser(@HeaderParam("userName") String addedBy) throws Exception{
     //Here I have one if condition which is checking from database if user exist!

     if (checkUser(addedBy).await().indefinitely().intValue() > 0) {
        //then creatingfolders

        if (createFolders(addedBy) > 0) {
             UserBean u = new UserBean(addedBy);
             return Uni.createFrom().item(userRepository.addUser(u)).onItem().transform(f -> f != null ? Response.ok(f) : Response.ok(null))
                 .onItem().transform(ResponseBuilder::build);
        }
     }
}

public Uni<Long> addUser(UserBean userIN) {
   return Panache.withTransaction(userIN::persist)
                .replaceWith(userIN.getId());
}

如何让它等待创建文件夹完成并检查用户是否存在？我尝试过使用@Blocking，但没有成功。请帮忙

Answer 1

您应该切换到常规 Hibernate ORM 而不是 Hibernate Reactive，或者以反应方式将操作链接在一起。

而不是做

checkUser(addedBy).await().indefinitely().intValue()

做类似的事情：

checkUser(addedBy).onItem().transformToUni(count -> {
   ....
})

Answer 2

问题的答案是

runSubscriptionOn

（或

emitOn

） - vertx 事件循环线程不允许您运行阻塞代码以便不阻塞事件处理。但是，只有当您使用

await()

或类似的组合时，vertx 才能真正知道它被阻塞，而不是在运行其他阻塞或昂贵的操作（如阻塞 IO 操作、互斥锁等）时，才会发生这种情况，无论是哪种情况你应该在不同的工作线程/池上运行：

import io.smallrye.mutiny.infrastructure.Infrastructure;

Uni<String> uni = Uni.createFrom()
        .item(this::invokeRemoteServiceUsingBlockingIO)
        .runSubscriptionOn(Infrastructure.getDefaultWorkerPool());
//                                        ''''''''''''''''''''

请注意，您显然需要使用惰性项目（供应商等），否则您仍然会阻塞事件线程。

另请参阅：

https://smallrye.io/smallrye-mutiny/latest/guides/imperative-to-reactive/#running-blocking-code-on-subscription

此外，vertx 有一个文件系统模块（您可以在 Bean 中注入 Vertx），它可以让您被动地处理文件，所以在这种情况下您甚至不需要它：

@Inject
io.vertx.mutiny.core.Vertx vertx;
//       ''''''
Uni<Void> uni = vertx.fileSystem().mkdir("/foo");

注意

：自从写这篇文章以来，我更喜欢emitOn而不是

runSubscriptionOn

，它们达到了相同的结果，但是当

runSubscriptionOn

影响方法之前的链时，emitOn影响方法之后

之后的链，这有事实证明更容易合作 - 但是 YMMV.

您提到您已经尝试过 Blocking 注释，但是您使用了正确的注释吗？我相信有两个，而且只有一个作品。

Answer 3