Quarkus应用中如何保持Vert.x上下文而非切换到Project Reactor线程？

问题根源 #

出现的HR000069和HR000068错误核心原因：

• Hibernate Reactive的Reactive Session严格绑定在创建它的Vert.x EventLoop线程，不允许跨线程访问
• Azure异步Blob客户端基于Project Reactor实现，默认在reactor-http-nio线程池执行操作，线程切换后破坏了Hibernate的线程绑定上下文
• 你使用的runSubscriptionOn(Infrastructure.getDefaultWorkerPool())仅指定了订阅线程，但Reactor内部仍会切换到自身线程池，后续操作（或隐式的Session访问）会在错误线程触发异常

修复步骤 #

1. 调整Blob上传的线程切换逻辑 #

在将Reactor的Flux转换为Mutiny的Uni后，需要：

• 用runSubscriptionOn将Blob上传的IO操作放到Worker线程执行（避免阻塞EventLoop）
• 用emitOn强制操作完成后回到Vert.x EventLoop线程，保持Hibernate上下文的线程绑定
• 移除方法末尾不必要的runSubscriptionOn，避免整个合并操作脱离EventLoop

修改后的storeMultiple方法：

import io.vertx.core.Vertx;
import io.smallrye.mutiny.infrastructure.Infrastructure;

public Uni<Void> storeMultiple(List<BlobData> blobs) {
    List<Uni<Void>> uploadUnis = blobs.stream().map(blobData -> Uni
            .createFrom()
            .publisher(AdaptersToFlow.publisher(
                    blobContainerClient.getBlobAsyncClient(blobData.getBlobName())
                            .upload(BinaryData.fromBytes(blobData.getData()))
            ))
            // 将Blob上传的IO操作调度到Worker线程，不阻塞EventLoop
            .runSubscriptionOn(Infrastructure.getDefaultWorkerPool())
            // 强制回到Vert.x EventLoop线程，保持Hibernate上下文绑定
            .emitOn(Vertx.currentContext().owner().getEventLoop())
            .onItem().ignore().andContinueWithNull()
    ).toList();

    return Uni
            .join()
            .all(uploadUnis)
            .usingConcurrencyOf(blobConfig.uploadConcurrency())
            .andFailFast()
            .onItem().ignore().andContinueWithNull()
            .onFailure().recoverWithUni(exception -> {
                Log.error("Upload multiple blobs", "An error occurred while uploading one or more blobs.", exception);
                return Uni.createFrom().failure(new BlobProblem("An error occurred while uploading one or more blobs.", exception));
            });
}

2. 确保核心调用链的上下文一致性 #

如果你的storeEntity.store(entity)是在事务中执行的，显式声明Blob操作完成后回到EventLoop线程（进一步确保上下文一致性）：

storeEntity.store(entity)
        .onItem().transformToUni(ignored -> storeBlob.storeMultiple(blobs)
                // 显式确保Blob操作完成后回到EventLoop
                .emitOn(Vertx.currentContext().owner().getEventLoop())
        )
        .onFailure().recoverWithUni(exception -> {
            Log.error(entity.getId().toString(), "An error occurred while handling new entity.", exception);
            return Uni.createFrom().failure(exception);
        });

3. 可选：使用Quarkus的MutinyHelper适配Reactor上下文 #

如果你希望更稳妥地传递Vert.x上下文到Reactor操作中，可以替换AdaptersToFlow为MutinyHelper.subscribeInContext，确保Reactor操作在当前Vert.x上下文的线程中执行：

import io.quarkus.mutiny.runtime.MutinyHelper;

// 替换原有的Uni.createFrom.publisher部分
Uni.createFrom().publisher(MutinyHelper.subscribeInContext(
    blobContainerClient.getBlobAsyncClient(blobData.getBlobName())
            .upload(BinaryData.fromBytes(blobData.getData())),
    Vertx.currentContext()
))

关键原理 #

• EventLoop线程：负责处理Hibernate Reactive的Session/事务，必须保持上下文的线程绑定，不能被阻塞
• Worker线程：专门用于执行IO密集型或阻塞操作（如Blob上传），避免占用EventLoop影响性能
• 线程切换的边界：IO操作在Worker线程执行，完成后必须回到EventLoop线程，才能继续执行与Hibernate相关的操作

内容的提问来源于stack exchange，提问作者Ricardo de Vries