如何构建异步静止endpoint，在工作线程中调用阻塞操作并立即回复（Quarkus）

Question

我查看了文档和stackoverflow，但没有找到合适的方法。E、 g.这篇文章似乎非常接近：在Quarkus/Mutiny的反应式REST GETendpoint中调度阻塞服务。然而，我不希望在我的服务中有这么多不必要的样板代码，最多也不要更改任何服务代码。

我通常只想调用一个使用实体管理器的服务方法，因此是一个阻塞操作，但是，我想立即向调用者返回一个字符串，就像“查询开始”之类的。我不需要回调对象，它只是一种火了就忘记的方法。

我试过这样的

@NonBlocking
@POST
@Produces(MediaType.TEXT_PLAIN)
@Path("/query")
public Uni<String> triggerQuery() {
    return Uni.createFrom()
    .item("query started")
    .call(() -> service.startLongRunningQuery());
}

但它不起作用-

You have attempted to perform a blocking operation on a IO thread. This is not allowed, as blocking the IO thread will cause major performance issues with your application. If you want to perform blocking EntityManager operations make sure you are doing it from a worker thread.",

我实际上期望qukus注意相应地分配任务，即对io线程的rest调用和对工作线程的阻塞实体管理器操作。所以我一定是用错了。

更新：

还尝试了我在https://github.com/quarkusio/quarkus/issues/11535将方法正文更改为

return Uni.createFrom()
        .item("query started")
        .emitOn(Infrastructure.getDefaultWorkerPool())
        .invoke(()-> service.startLongRunningQuery());

现在我没有得到一个错误，但service.startLongRunningQuery（）没有被调用，因此没有日志和查询实际上发送到数据库。

与（如何使用Mutiny反应式编程调用长时间运行的阻塞无效返回方法？）：

return Uni.createFrom()
            .item(() ->service.startLongRunningQuery()) 
            .runSubscriptionOn(Infrastructure.getDefaultWorkerPool())

与（如何在另一个线程上运行阻塞代码并立即返回超文本传输协议请求）相同：

ExecutorService executor = Executors.newFixedThreadPool(10, r -> new Thread(r, "CUSTOM_THREAD"));

return Uni.createFrom()
                .item(() -> service.startLongRunningQuery())
                .runSubscriptionOn(executor);

知道为什么要服务吗。假设rest调用通过IO线程处理，服务调用由工作线程处理，则根本不调用startongrunningquery（），如何实现触发和遗忘行为？

Answer 1

使用EventBus进行https://quarkus.io/guides/reactive-event-bus

发送和忘记是一条路要走。

Answer 2

这取决于您是要立即返回（在有效执行startongrunningquery操作之前），还是要等待操作完成。

如果是第一种情况，请使用以下内容：

@Inject EventBus bus;

@NonBlocking
@POST
@Produces(MediaType.TEXT_PLAIN)
@Path("/query")
public void triggerQuery() {
    bus.send("some-address", "my payload");
}

@Blocking // Will be called on a worker thread
@ConsumeEvent("some-address")
public void executeQuery(String payload) {
    service.startLongRunningQuery();
}

在第二种情况下，您需要在工作线程上执行查询。

@POST
@Produces(MediaType.TEXT_PLAIN)
@Path("/query")
public Uni<String> triggerQuery() {
   return Uni.createFrom(() -> service.startLongRunningQuery())
      .runSubscriptionOn(Infrastructure.getDefaultWorkerPool());
}

请注意，您需要RESTEasy反应式才能工作（而不是经典的RESTEasy）。如果您使用经典的RESTEasy，您将需要quarkus RESTEasy mutiny扩展（但我建议使用RESTEasy Responsive，它将更加有效）。