Uni<T> 如何在 Quarkus 中工作

问题描述

在查看了他们的官方教程 getting started with async guide 后，我试图了解 Quarkus 框架中 Uni 的行为。 在服务方法中我做了以下改动

package org.acme.getting.started.async;

import javax.enterprise.context.ApplicationScoped;

import io.smallrye.mutiny.Uni;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;


@ApplicationScoped
public class GreetingService {

    ExecutorService executor = Executors.newFixedThreadPool(10,r -> {
        return new Thread(r,"CUSTOM_TASK_EXECUTION_THREAD");
    });


    public Uni<String> greeting(String name) {
        System.out.println("greeting Executing on Thread "+Thread.currentThread().getName());
        return Uni.createFrom().item(ioSimulation(name))
                .emitOn(executor);//Infrastructure.getDefaultExecutor()
    }

    public String ioSimulation(String param){
        System.out.println("ioSimulation Executing on Thread "+Thread.currentThread().getName());
        try {
            Thread.sleep(8000);
        } catch (InterruptedException e) {
            e.printstacktrace();
        }
        return "hello "+param;

    }

}

然后我测试了 /greeting/{name} 资源，执行根本不是异步的，实际上它在同一线程内以同步方式执行所有相关方法。

那有什么区别

@GET
    @Produces(MediaType.TEXT_PLAIN)
    @Path("/greeting/{name}")
    public Uni<String> greeting(@PathParam String name) {
    }

和

@GET
    @Produces(MediaType.TEXT_PLAIN)
    @Path("/greeting/{name}")
    public String greeting(@PathParam String name) {
    }

它是如何异步的？请帮助我理解它。

解决方法

问题出在Uni.createFrom().item(ioSimulation(name))。在创建 ioSimulation 之前同步调用 Uni。

您至少应该使用 Supplier 变体：Uni.createFrom().item(() -> ioSimulation(name))。那应该会有所帮助。

,

除了@Ladicek 的回答，我相信您想使用 runSubscriptionOn 而不是 emitOn。请参阅 https://smallrye.io/smallrye-mutiny/guides/emit-subscription 作为参考。

,

以上两个答案都是正确的。我对服务类进行了一些更改，ioSimulation(String param) 方法已卸载到自定义线程池。最终解决方案如下

public class GreetingService {

    ThreadFactory threadFactory = new NameableThreadFactory("CUSTOM_TASK_EXECUTION_THREAD");
    ExecutorService executor = Executors.newFixedThreadPool(10,threadFactory);

    public Uni<String> greeting(String name) {
        System.out.println("greeting Executing on Thread "+Thread.currentThread().getName());
        return Uni.createFrom().item(()->ioSimulation(name))
                .runSubscriptionOn(executor);//Infrastructure.getDefaultExecutor()
    }


    public String ioSimulation(String param){
        System.out.println("ioSimulation Executing on Thread "+Thread.currentThread().getName());
        try {
            Thread.sleep(8000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "hello "+param;

    }
}

--更新--

如果您在 emitOn(Executor executor) 方法中传递 supplier，

item 也将异步工作，但方法略有不同，我认为在试验 Uni 时，{{3 }} 是我的仓库，如果有人感兴趣的话。