Reactive Quarkus应用Java/原生模式运行行为不一致问题

问题背景 #

集成hibernate-panache-reactive的响应式Quarkus应用，存在Java模式、原生镜像模式运行行为不一致的问题，业务执行流程分为4个步骤：

• 通过hibernate-panache-reactive从MySQL数据库加载大量数据
• 基于已加载的数据构建关系图
• 在关系图上执行耗时较长的计算算法
• 根据步骤3返回的结果从数据库加载更多数据

初始实现代码如下：

GraphProcessor graphProcessor = createInitialProcessor();
return Uni.createFrom().item(graphProcessor)
// 1) loading of initial data
        .onItem().transformToUni(this::loadDataViaPanaceReactive1)
        .onItem().transformToUni(this::loadDataViaPanaceReactive2)
        .onItem().transformToUni(this::loadDataViaPanaceReactive3)
// 2) building of graph               
        .onItem().transform(graphProcessor::processLoadedData)
        .onItem().invoke(graphProcessor::loadingComplete) //sync
// 3) running time consuming algorithm on graph
        .onItem().transformToMulti(this::runTimeConsumingTask)
        .onItem().invoke(this::prepareDBQueries)
// 4) load more data from DB
        .onItem().transformToUniAndConcatenate(this::loadMoreData1)
        .onItem().transformToUniAndConcatenate(this::loadMoreData2)
        .onItem().transformToUniAndConcatenate(this::transformToPublicForm)
        .onFailure().invoke(log::error);

问题迭代过程 #

1. 初始代码在Java模式下运行正常，但原生模式运行时抛出线程阻塞错误：步骤2、3的计算任务耗时过长，阻塞了事件循环调用线程。
2. 第一次修复：在步骤1和步骤2之间添加.emitOn(Infrastructure.getDefaultWorkerPool())将耗时任务调度到工作线程池执行，解决了线程阻塞问题，但触发新的会话线程不匹配错误：

java.lang.IllegalStateException: HR000069: Detected use of the reactive Session from a different Thread than the one which was used to open the reactive Session - this suggests an invalid integration; original thread: 'vert.x-eventloop-thread-0' current Thread: 'vert.x-eventloop-thread-1'

3. 第二次修复：在步骤3和步骤4之间插入.emitOn(Infrastructure.getDefaultExecutor())，尝试将执行线程切回响应式会话所属的事件循环线程，修改后代码如下：

GraphProcessor graphProcessor = createInitialProcessor();
return Uni.createFrom().item(graphProcessor)
// 1) loading of initial data
        .onItem().transformToUni(this::loadDataViaPanaceReactive1)
        .onItem().transformToUni(this::loadDataViaPanaceReactive2)
        .onItem().transformToUni(this::loadDataViaPanaceReactive3)
// 2) building of graph               
        .emitOn(Infrastructure.getDefaultWorkerPool()) // Required for native mode
        .onItem().transform(graphProcessor::processLoadedData)
        .onItem().invoke(graphProcessor::loadingComplete)
// 3) running time consuming algorithm on graph
        .onItem().transformToMulti(this::runTimeConsumingTask)
        .onItem().invoke(this::prepareDBQueries)
        .emitOn(Infrastructure.getDefaultExecutor()) // Required for native mode
// 4) load more data from DB
        .onItem().transformToUniAndConcatenate(this::loadMoreData1)
        .onItem().transformToUniAndConcatenate(this::loadMoreData2)
        .onItem().transformToUniAndConcatenate(this::transformToPublicForm)
        .onFailure().invoke(log::error);

该版本在原生模式下可正常运行，但切回Java模式运行时，会偶发抛出相同的HR000069响应式会话线程不匹配异常。

现有写法的核心问题 #

• 对线程切换API的能力存在错误假设：Infrastructure.getDefaultExecutor()返回的是Vert.x全部事件循环线程组成的线程池，调度时会随机分配空闲线程，和最初打开Reactive Session的特定事件循环线程没有绑定关系，无法保证精准回到原线程，必然出现偶发的线程不匹配问题。
• 线程切换逻辑未做Session上下文隔离：直接将持有Panache实体引用的逻辑调度到工作线程，跨线程访问时会直接触发Hibernate Reactive的线程校验规则。
• 跨模式表现不一致的根源是JVM模式和原生模式的线程调度时序、负载分布存在差异：原生镜像启动速度更快、事件循环线程负载更均匀，因此错误的切线程逻辑在原生模式下刚好大概率命中原线程，在JVM模式下调度随机性更强，就会出现偶发异常。

响应式链路混合耗时任务+Panache操作的最佳实践 #

• 严格拆分任务边界：将所有不访问Reactive Session、不触发Panache实体懒加载、不执行DB操作的CPU密集/阻塞耗时逻辑，和Panache数据库操作完全拆分。耗时逻辑统一通过runSubscriptionOn(Infrastructure.getDefaultWorkerPool())调度到工作线程池执行，不要用emitOn做跨线程调度。
• 禁止手动切线程匹配Session：不需要手动调用API切回事件循环线程，Hibernate Reactive Panache自带请求上下文传播能力，工作线程执行完耗时任务返回结果后，后续链式调用的Panache操作会自动路由到当前请求绑定的Session所属事件循环线程，手动切线程反而会破坏上下文关联。
• 跨线程传递时剥离Session关联：耗时任务调度到工作线程前，提前在事件循环线程上将Panache实体中需要用到的字段提取为普通POJO，不要将未脱管的实体引用传递到工作线程，避免工作线程中意外触发Session访问。
• 不要在工作线程中执行任何Panache相关操作：所有DB操作全部放在响应式链路的非工作线程调度段执行，工作线程只处理纯内存计算逻辑。

修正后参考代码 #

GraphProcessor graphProcessor = createInitialProcessor();
return Uni.createFrom().item(graphProcessor)
// 1) 初始数据加载：在Session绑定的事件循环线程执行
        .onItem().transformToUni(this::loadDataViaPanaceReactive1)
        .onItem().transformToUni(this::loadDataViaPanaceReactive2)
        .onItem().transformToUni(this::loadDataViaPanaceReactive3)
        // 提前提取纯内存计算数据，剥离Session关联
        .onItem().transform(this::extractPlainCalcData)
// 2+3) 图构建+耗时计算：调度到工作线程执行，全程不碰Panache/Session
        .chain(plainData -> Uni.createFrom().item(() -> {
            Graph processedGraph = graphProcessor.processLoadedData(plainData);
            graphProcessor.loadingComplete();
            return runTimeConsumingTaskAndCollectResult(processedGraph);
        }).runSubscriptionOn(Infrastructure.getDefaultWorkerPool()))
// 4) 后续DB查询：直接调用Panache方法，框架自动路由到正确Session线程
        .onItem().transformToMulti(calcResult -> Multi.createFrom().iterable(calcResult.getQueryBatches()))
        .onItem().invoke(this::prepareDBQueries)
        .onItem().transformToUniAndConcatenate(this::loadMoreData1)
        .onItem().transformToUniAndConcatenate(this::loadMoreData2)
        .onItem().transformToUniAndConcatenate(this::transformToPublicForm)
        .onFailure().invoke(log::error);

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