Quarkus + Mutiny：阻塞代码执行与线程/上下文管理咨询

场景说明 #

应用启动阶段需依次执行3个一次性任务：

• Task 1：包含阻塞操作（如从GCP存储桶加载数据）
• Task 2：非阻塞数据转换
• Task 3：基于Hibernate Reactive Panache将数据写入PostgreSQL数据库

Q1：将阻塞任务放到工作线程执行的推荐方式是什么？使用runSubscriptionOn还是有更优方案？ #

最优方案分两种，按需选择：

1. 显式使用runSubscriptionOn（响应式流内细粒度控制首选）
   把阻塞逻辑封装在Uni中，通过runSubscriptionOn(Infrastructure.getDefaultWorkerPool())指定工作线程池执行阻塞操作，精准避免阻塞Vert.x事件循环。
   示例：

   public Uni<Void> task1WithBlocking() {
       return Uni.createFrom().voidItem()
               .invoke(this::blockingMethod) // 阻塞逻辑
               .runSubscriptionOn(Infrastructure.getDefaultWorkerPool()) // 指定工作线程
               .invoke(() -> logger.info("Log 4: Completed task1WithBlocking."))
               .replaceWithVoid();
   }
   

2. Quarkus @Blocking注解（独立阻塞方法管理首选）
   给阻塞方法添加@Blocking注解，Quarkus会自动将该方法调度到工作线程池执行，无需手动指定线程池，代码更简洁。
   示例：

   @Blocking
   public void blockingMethod() {
       try {
           Thread.sleep(2000);
       } catch (InterruptedException e) {
           e.printStackTrace();
       }
       logger.info("Log 3: Blocking method executed.");
   }
   

   注意：在响应式流中调用带@Blocking的方法时，建议配合Uni.createFrom().call(this::blockingMethod)确保上下文正确切换。

Q2：当阻塞任务在工作线程执行时，如何让Task 2和Task 3回到原线程（Vert.x事件循环）执行？ #

使用emitOn操作符切换回Vert.x事件循环线程即可。runSubscriptionOn仅影响订阅阶段的线程，而emitOn会改变后续所有操作的执行线程。在Task 1的Uni链后添加emitOn(Infrastructure.getDefaultExecutor())，就能让后续的Task 2、Task 3回到事件循环线程执行。

调整后的流水线示例：

public Uni<Void> executeDataPipeline() {
    logger.info("Log 2: Starting executeDataPipeline");

    return Uni.createFrom().voidItem()
            .transformToUni(unused -> task1WithBlocking())
            .emitOn(Infrastructure.getDefaultExecutor()) // 切换回事件循环线程
            .transformToUni(unused -> task2NonBlocking())
            .transformToUni(unused -> task3NonBlockingDB())
            .invoke(() -> logger.info("Log 7: Processing initialisation complete."));
}

Infrastructure.getDefaultExecutor()在Quarkus中默认指向Vert.x事件循环线程池，能确保后续非阻塞任务（尤其是依赖事件循环的Hibernate Reactive操作）在正确上下文执行。

Q3：@Observes StartupEvent运行在Quarkus主线程，导致Hibernate Reactive Panache报错，如何解决？ #

核心问题是Hibernate Reactive依赖Vert.x事件循环上下文，而StartupEvent观察者方法运行在Quarkus启动主线程，无事件循环上下文。有三种可靠解决办法：

方案1：通过Vertx API调度到事件循环 #

注入Vertx实例，使用runOnContext将流水线的订阅操作放到事件循环线程执行：

void onStart(@Observes StartupEvent ev, Vertx vertx) {
    logger.info("Log 1: QuarkusBlockingExample onStart is starting...");

    vertx.runOnContext(v -> {
        executeDataPipeline().subscribe().with(
                success -> logger.info("Log 5: executeDataPipeline completed."),
                failure -> logger.error("Log 5f: executeDataPipeline failed", failure)
        );
    });

    logger.info("Log 8: QuarkusBlockingExample onStart is finished.");
}

方案2：使用Mutiny的runSubscriptionOn切换订阅线程 #

直接在流水线订阅前添加runSubscriptionOn，将整个流水线的订阅和执行切换到事件循环线程：

void onStart(@Observes StartupEvent ev) {
    logger.info("Log 1: QuarkusBlockingExample onStart is starting...");

    executeDataPipeline()
            .runSubscriptionOn(Infrastructure.getDefaultExecutor()) // 切换到事件循环线程订阅
            .subscribe().with(
                success -> logger.info("Log 5: executeDataPipeline completed."),
                failure -> logger.error("Log 5f: executeDataPipeline failed", failure)
            );

    logger.info("Log 8: QuarkusBlockingExample onStart is finished.");
}

方案3：使用QuarkusApplication替代StartupEvent #

实现QuarkusApplication接口，在run方法中直接执行响应式流水线，该方法默认运行在Vert.x事件循环上下文，适合将启动任务作为应用启动的必要步骤：

@ApplicationScoped
public class DataPipelineApp implements QuarkusApplication {
    private static final Logger logger = Logger.getLogger(DataPipelineApp.class);

    @Override
    public int run(String... args) throws Exception {
        logger.info("Log 1: Starting data pipeline...");
        executeDataPipeline().await().indefinitely();
        logger.info("Log 8: Data pipeline completed, application starting...");
        return 0;
    }

    // 流水线和任务方法同原有实现
}

优化后的完整代码示例 #

import io.quarkus.runtime.StartupEvent;
import io.smallrye.mutiny.Uni;
import io.smallrye.mutiny.infrastructure.Infrastructure;
import io.vertx.core.Vertx;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.enterprise.event.Observes;
import org.jboss.logging.Logger;

@ApplicationScoped
public class QuarkusBlockingExample {
    private static final Logger logger = Logger.getLogger(QuarkusBlockingExample.class);

    void onStart(@Observes StartupEvent ev, Vertx vertx) {
        logger.info("Log 1: QuarkusBlockingExample onStart is starting...");

        // 将流水线调度到Vert.x事件循环线程执行
        vertx.runOnContext(v -> {
            executeDataPipeline().subscribe().with(
                    success -> logger.info("Log 5: executeDataPipeline completed."),
                    failure -> logger.error("Log 5f: executeDataPipeline failed", failure)
            );
        });

        logger.info("Log 8: QuarkusBlockingExample onStart is finished.");
    }

    public Uni<Void> executeDataPipeline() {
        logger.info("Log 2: Starting executeDataPipeline");

        return Uni.createFrom().voidItem()
                .transformToUni(unused -> task1WithBlocking())
                .emitOn(Infrastructure.getDefaultExecutor()) // 切换回事件循环线程
                .transformToUni(unused -> task2NonBlocking())
                .transformToUni(unused -> task3NonBlockingDB())
                .invoke(() -> logger.info("Log 7: Processing initialisation complete."));
    }

    public Uni<Void> task1WithBlocking() {
        return Uni.createFrom().voidItem()
                .invoke(this::blockingMethod)
                .runSubscriptionOn(Infrastructure.getDefaultWorkerPool()) // 阻塞任务放工作线程
                .invoke(() -> logger.info("Log 4: Completed task1WithBlocking."))
                .replaceWithVoid();
    }

    public Uni<Void> task2NonBlocking() {
        return Uni.createFrom().voidItem()
                .invoke(() -> logger.info("Log 5: Completed task2NonBlocking."));
    }

    public Uni<Void> task3NonBlockingDB() {
        // 实际场景中替换为Hibernate Reactive Panache的操作
        return Uni.createFrom().voidItem()
                .invoke(() -> logger.info("Log 6: Completed DB load using Reactive Hibernate Panache."));
    }

    private void blockingMethod() {
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        logger.info("Log 3: Blocking method executed.");
    }
}

内容的提问来源于stack exchange，提问作者gunjan kumar