Quarkus异步执行中的上下文传播问题求助

问题背景 #

在Quarkus应用开发中，异步调用发生线程切换时，HTTP请求头等请求上下文信息会丢失，导致从上下文直接获取HttpServerRequest时抛出IllegalProductException。尝试过CompletableFuture、VirtualThreads等异步方案，问题始终存在。

复现代码 #

package org.context;

import io.quarkus.logging.Log;
import io.smallrye.mutiny.Uni;
import io.smallrye.mutiny.infrastructure.Infrastructure;
import io.smallrye.mutiny.subscription.Cancellable;
import io.vertx.core.MultiMap;
import io.vertx.core.http.HttpServerRequest;
import jakarta.enterprise.context.RequestScoped;
import jakarta.enterprise.inject.IllegalProductException;
import jakarta.ws.rs.core.Context;

import java.time.Duration;

@RequestScoped
public class GreetingService {

    @Context
    HttpServerRequest httpServerRequest;

    public MultiMap captureHeadersSync() {
        MultiMap headers = httpServerRequest.headers();
        Log.infof("Request Headers: %s", headers);
        return headers;
    }

    public void captureHeaderAsync() {
        Cancellable data = Uni
            .createFrom()
            .item(httpServerRequest.headers())
            .runSubscriptionOn(Infrastructure.getDefaultExecutor())
            .onItem()
            .delayIt()
            .by(Duration.ofSeconds(5))
            .subscribe()
            .with(
                item -> {
                    try {
                        Log.infof("Request Headers async sent by parameters: %s", item); // 正常工作
                        Log.infof("Request Headers async get from context: %s", httpServerRequest.headers()); // 抛出异常
                    } catch (IllegalProductException ignored) {
                        Log.info("捕获Headers失败。");
                    }
                },
                failure -> Log.info("请求失败。")
            );
    }
}

错误日志 #

jakarta.enterprise.inject.IllegalProductException: Normal scoped producer method may not return null: io.quarkus.vertx.http.runtime.CurrentVertxRequest.getCurrent()
        at io.quarkus.vertx.http.runtime.CurrentVertxRequest_ProducerMethod_getCurrent_bcI9FtU7pcNOHntvVCkP17muvXY_Bean.doCreate(Unknown Source)
        at io.quarkus.vertx.http.runtime.CurrentVertxRequest_ProducerMethod_getCurrent_bcI9FtU7pcNOHntvVCkP17muvXY_Bean.create(Unknown Source)
        at io.quarkus.vertx.http.runtime.CurrentVertxRequest_ProducerMethod_getCurrent_bcI9FtU7pcNOHntvVCkP17muvXY_Bean.create(Unknown Source)
        at io.quarkus.arc.impl.CurrentManagedContext$1.get(CurrentManagedContext.java:111)
        at io.quarkus.arc.impl.CurrentManagedContext$1.get(CurrentManagedContext.java:107)
        at io.quarkus.arc.generator.Default_jakarta_enterprise_context_RequestScoped_ContextInstances.c4(Unknown Source)
        at io.quarkus.arc.generator.Default_jakarta_enterprise_context_RequestScoped_ContextInstances.computeIfAbsent(Unknown Source)
        at io.quarkus.arc.impl.CurrentManagedContext.getIfActive(CurrentManagedContext.java:107)
        at io.quarkus.arc.impl.ClientProxies.getSingleContextDelegate(ClientProxies.java:30)
        at io.vertx.ext.web.CurrentVertxRequest_ProducerMethod_getCurrent_bcI9FtU7pcNOHntvVCkP17muvXY_ClientProxy.arc$delegate(Unknown Source)
        at io.vertx.ext.web.CurrentVertxRequest_ProducerMethod_getCurrent_bcI9FtU7pcNOHntvVCkP17muvXY_ClientProxy.request(Unknown Source)
        at io.quarkus.vertx.http.runtime.CurrentRequestProducer.getCurrentRequest(CurrentRequestProducer.java:22)
        at io.quarkus.vertx.http.runtime.CurrentRequestProducer_ProducerMethod_getCurrentRequest_BflQ6nq5HRIboLrFJbidYvMmGy0_Bean.doCreate(Unknown Source)
        at io.quarkus.vertx.http.runtime.CurrentRequestProducer_ProducerMethod_getCurrentRequest_BflQ6nq5HRIboLrFJbidYvMmGy0_Bean.create(Unknown Source)
        at io.quarkus.vertx.http.runtime.CurrentRequestProducer_ProducerMethod_getCurrentRequest_BflQ6nq5HRIboLrFJbidYvMmGy0_Bean.create(Unknown Source)
        at io.quarkus.arc.impl.CurrentManagedContext$1.get(CurrentManagedContext.java:111)
        at io.quarkus.arc.impl.CurrentManagedContext$1.get(CurrentManagedContext.java:107)
        at io.quarkus.arc.generator.Default_jakarta_enterprise_context_RequestScoped_ContextInstances.c3(Unknown Source)
        at io.quarkus.arc.generator.Default_jakarta_enterprise_context_RequestScoped_ContextInstances.computeIfAbsent(Unknown Source)
        at io.quarkus.arc.impl.CurrentManagedContext.getIfActive(CurrentManagedContext.java:107)
        at io.quarkus.arc.impl.ClientProxies.getSingleContextDelegate(ClientProxies.java:30)
        at io.vertx.core.http.CurrentRequestProducer_ProducerMethod_getCurrentRequest_BflQ6nq5HRIboLrFJbidYvMmGy0_ClientProxy.arc$delegate(Unknown Source)
        at io.vertx.core.http.CurrentRequestProducer_ProducerMethod_getCurrentRequest_BflQ6nq5HRIboLrFJbidYvMmGy0_ClientProxy.headers(Unknown Source)
        at org.context.GreetingService.lambda$captureHeaderAsync$0(GreetingService.java:40)
        at io.smallrye.context.impl.wrappers.SlowContextualConsumer.accept(SlowContextualConsumer.java:21)
        at io.smallrye.mutiny.helpers.UniCallbackSubscriber.onItem(UniCallbackSubscriber.java:73)
        at io.smallrye.mutiny.operators.uni.UniDelayOnItem$UniDelayOnItemProcessor.lambda$onItem$0(UniDelayOnItem.java:53)
        at org.jboss.threads.EnhancedQueueExecutor$RunnableScheduledFuture.performTask(EnhancedQueueExecutor.java:3033)
        at org.jboss.threads.EnhancedQueueExecutor$RunnableScheduledFuture.performTask(EnhancedQueueExecutor.java:3023)
        at org.jboss.threads.EnhancedQueueExecutor$AbstractScheduledFuture.run(EnhancedQueueExecutor.java:2878)
        at io.quarkus.vertx.core.runtime.VertxCoreRecorder$15.runWith(VertxCoreRecorder.java:637)
        at org.jboss.threads.EnhancedQueueExecutor$Task.doRunWith(EnhancedQueueExecutor.java:2651)
        at org.jboss.threads.EnhancedQueueExecutor$Task.run(EnhancedQueueExecutor.java:2630)
        at org.jboss.threads.EnhancedQueueExecutor.runThreadBody(EnhancedQueueExecutor.java:1622)
        at org.jboss.threads.EnhancedQueueExecutor$ThreadBody.run(EnhancedQueueExecutor.java:1589)
        at org.jboss.threads.DelegatingRunnable.run(DelegatingRunnable.java:11)
        at org.jboss.threads.ThreadLocalResettingRunnable.run(ThreadLocalResettingRunnable.java:11)
        at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
        at java.base/java.lang.Thread.run(Thread.java:1583)

解决方案 #

1. 提前捕获上下文数据传递到异步流程 #

在同步阶段获取需要的上下文信息（如请求头），直接传递到异步回调中，避免在异步线程中访问已失效的请求作用域代理：

public void captureHeaderAsync() {
    // 同步阶段捕获请求头
    MultiMap headers = httpServerRequest.headers();
    
    Uni.createFrom().item(headers)
        .runSubscriptionOn(Infrastructure.getDefaultExecutor())
        .onItem().delayIt().by(Duration.ofSeconds(5))
        .subscribe().with(
            item -> Log.infof("Request Headers async: %s", item),
            failure -> Log.info("请求失败。")
        );
}

2. 手动管理请求上下文（进阶场景） #

如果需要在异步流程中完整保留请求上下文，可使用RequestContextController手动激活/关闭上下文：

import io.quarkus.arc.RequestContextController;
import jakarta.inject.Inject;

@RequestScoped
public class GreetingService {

    @Context
    HttpServerRequest httpServerRequest;
    
    @Inject
    RequestContextController requestContextController;

    public void captureHeaderAsync() {
        // 激活当前请求上下文
        requestContextController.activate();
        try {
            Uni.createFrom().item(() -> httpServerRequest.headers())
                .runSubscriptionOn(Infrastructure.getDefaultExecutor())
                .onItem().delayIt().by(Duration.ofSeconds(5))
                .subscribe().with(
                    item -> {
                        try {
                            Log.infof("Request Headers async from context: %s", item);
                        } finally {
                            // 异步完成后关闭上下文
                            requestContextController.deactivate();
                        }
                    },
                    failure -> {
                        Log.info("请求失败。");
                        requestContextController.deactivate();
                    }
                );
        } catch (Exception e) {
            requestContextController.deactivate();
            throw e;
        }
    }
}

3. 避免不必要的线程切换 #

如果业务逻辑不需要强制切换线程，移除runSubscriptionOn调用，让Mutiny在原请求线程的事件循环中处理，自然保留上下文。

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