使用emitOn无法从工作线程池切换回Vert.x事件循环线程的问题咨询

你的代码与执行日志 #

首先来看你提供的资源方法代码：

@GET
@Path("/p1")
public Uni<Response> p1() {
    Log.info("Entry (event loop)");
    return Uni.createFrom()
            .item("input")
            .onItem().transform(input -> {
                Log.info("Light transform");
                return input.toUpperCase();
            })
            .emitOn(Infrastructure.getDefaultWorkerPool()) // ← Switch to worker
            .onItem().transform(data -> {
                Log.info("Heavy work (worker)");
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException ignored) {
                }
                return "Processed: " + data;
            })
            .emitOn(Infrastructure.getDefaultExecutor()) // doesn't work
            .onItem().transform(result -> {
                Log.info("Back on event loop");
                return Response.ok(result).build();
            });
}

对应的执行日志：

2025-08-11 04:13:58,404 INFO [com.exa.ExampleResource] (vert.x-eventloop-thread-1) Entry (event loop)
2025-08-11 04:13:58,405 INFO [com.exa.ExampleResource] (vert.x-eventloop-thread-1) Light transform
2025-08-11 04:13:58,405 INFO [com.exa.ExampleResource] (executor-thread-1) Heavy work (worker)
2025-08-11 04:13:59,407 INFO [com.exa.ExampleResource] (executor-thread-1) Back on event loop

你的问题 #

你尝试用emitOn(Infrastructure.getDefaultExecutor())切换回事件循环线程但失败了，想知道问题出在哪，以及如何在完成CPU密集型工作后正确回到事件循环。

问题根源 #

你用错了执行器！Infrastructure.getDefaultExecutor()指向的是Quarkus的通用线程池（也就是日志里的executor-thread-1这类线程），它根本不是Vert.x的事件循环线程池。所以你调用这个方法后，仍然停留在工作线程池中，不会切换到event loop线程。

正确的解决方法 #

要回到Vert.x的事件循环线程，你需要拿到Vert.x事件循环对应的执行器，然后传递给emitOn。具体步骤如下：

1. 注入Vertx实例：在你的资源类中注入Vertx对象（Quarkus会自动管理它的生命周期）。
2. 使用Vertx上下文的执行器：通过vertx.getOrCreateContext().executor()获取事件循环的执行器，作为emitOn的参数。

修改后的代码示例：

import io.quarkus.vertx.VertxInstance;
import io.vertx.core.Vertx;
import jakarta.inject.Inject;
import jakarta.ws.rs.GET;
import jakarta.ws.rs.Path;
import jakarta.ws.rs.core.Response;
import org.jboss.logging.Logger;
import io.smallrye.mutiny.Uni;
import io.quarkus.arc.Infrastructure;

@Path("/p1")
public class ExampleResource {

    private static final Logger LOG = Logger.getLogger(ExampleResource.class);

    @Inject
    Vertx vertx; // 注入Vertx实例

    @GET
    public Uni<Response> p1() {
        LOG.info("Entry (event loop)");
        return Uni.createFrom()
                .item("input")
                .onItem().transform(input -> {
                    LOG.info("Light transform");
                    return input.toUpperCase();
                })
                .emitOn(Infrastructure.getDefaultWorkerPool()) // 切换到工作线程做密集工作
                .onItem().transform(data -> {
                    LOG.info("Heavy work (worker)");
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException ignored) {
                    }
                    return "Processed: " + data;
                })
                .emitOn(vertx.getOrCreateContext().executor()) // 切换回事件循环线程
                .onItem().transform(result -> {
                    LOG.info("Back on event loop");
                    return Response.ok(result).build();
                });
    }
}

修改后的预期日志 #

执行后你会看到最后一步成功回到了event loop线程：

2025-08-11 04:13:58,404 INFO [com.exa.ExampleResource] (vert.x-eventloop-thread-1) Entry (event loop)
2025-08-11 04:13:58,405 INFO [com.exa.ExampleResource] (vert.x-eventloop-thread-1) Light transform
2025-08-11 04:13:58,405 INFO [com.exa.ExampleResource] (executor-thread-1) Heavy work (worker)
2025-08-11 04:13:59,407 INFO [com.exa.ExampleResource] (vert.x-eventloop-thread-1) Back on event loop

额外说明 #

• 为什么这个方法有效？vertx.getOrCreateContext().executor()返回的是当前请求对应的Vertx事件循环的执行器，它会把后续的操作提交到事件循环线程上执行。
• 如果你不想注入Vertx，也可以用Vertx.currentVertx()获取当前的Vertx实例，但注入的方式更符合Quarkus的依赖注入规范，也更可靠。

内容来源于stack exchange