Quarkus 3.5.0中如何用PanacheRepository异步调用存储过程？

问题根源 #

你之前的代码中，Uni.createFrom().emitter()的回调逻辑仍运行在IO线程上，而EntityManager的JDBC操作是阻塞性的，违反了Quarkus的IO线程非阻塞约束，因此触发BlockingOperationNotAllowedException。核心解决思路是将阻塞的数据库操作转移到工作线程执行。

可行解决方案 #

方法1：使用@Blocking注解自动调度到工作线程 #

在PanacheRepository的方法上添加@Blocking注解，Quarkus会自动将方法执行调度到工作线程，同时保持反应式的Uni返回类型。

import io.quarkus.hibernate.orm.panache.PanacheRepository;
import io.smallrye.mutiny.Uni;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.persistence.EntityManager;
import jakarta.persistence.PersistenceContext;
import io.quarkus.vertx.Blocking;

@ApplicationScoped
public class YourEntityRepository implements PanacheRepository<YourEntity> {

    @PersistenceContext
    EntityManager entityManager;

    @Blocking
    public Uni<Integer> executeStoredProcedure() {
        // 执行更新类存储过程，返回受影响行数
        int affectedRows = entityManager.createNativeQuery("call procName(1, 2)", Integer.class)
                .executeUpdate();
        return Uni.createFrom().item(affectedRows);
    }
}

方法2：手动指定工作线程执行阻塞逻辑 #

如果不想依赖注解，可通过Uni的API手动将阻塞操作切换到工作线程，使用runSubscriptionOn绑定默认工作线程池：

import io.quarkus.hibernate.orm.panache.PanacheRepository;
import io.smallrye.mutiny.Uni;
import io.smallrye.mutiny.infrastructure.Infrastructure;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.persistence.EntityManager;
import jakarta.persistence.PersistenceContext;

@ApplicationScoped
public class YourEntityRepository implements PanacheRepository<YourEntity> {

    @PersistenceContext
    EntityManager entityManager;

    public Uni<Integer> executeStoredProcedure() {
        return Uni.createFrom().item(() -> {
            // 此逻辑将在工作线程执行
            return entityManager.createNativeQuery("call procName(1, 2)", Integer.class)
                    .executeUpdate();
        }).runSubscriptionOn(Infrastructure.getDefaultWorkerPool());
    }
}

方法3：结合事务执行存储过程 #

若存储过程需要在事务上下文运行，可结合Panache的withTransaction方法，并指定工作线程：

public Uni<Integer> executeStoredProcedureInTransaction() {
    return withTransaction(() -> {
        int affectedRows = entityManager.createNativeQuery("call procName(1, 2)", Integer.class)
                .executeUpdate();
        return Uni.createFrom().item(affectedRows);
    }).runSubscriptionOn(Infrastructure.getDefaultWorkerPool());
}

处理返回结果集的存储过程 #

如果存储过程返回查询结果集（而非更新计数），将executeUpdate()替换为getResultList()即可：

@Blocking
public Uni<List<Integer>> queryStoredProcedureResults() {
    List<Integer> results = entityManager.createNativeQuery("call procName(1, 2)", Integer.class)
            .getResultList();
    return Uni.createFrom().item(results);
}

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