Quarkus中注入Emitter失败：SRMSG00019通道连接异常求助

问题背景 #

基于Java 21的Quarkus 3.17应用中，使用Quartz的定时任务类注入AMQP的Emitter时失败，抛出错误jakarta.enterprise.inject.spi.DefinitionException: SRMSG00019: Unable to connect an emitter。但其他类注入同名Channel的Emitter完全正常，且该定时类中的ConfigMapping注入无异常。

问题类代码 #

package myproject.scheduling;

import java.util.List;

import org.eclipse.microprofile.reactive.messaging.Channel;
import org.eclipse.microprofile.reactive.messaging.Emitter;
import io.quarkus.scheduler.Scheduled;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.inject.Inject;

import myproject.config.ConnectionChecks;
import myproject.config.Constants.AmqpQueues.Producers;
import myproject.config.Constants.Jobs;
import myproject.models.measurements.ConnectionStatus;
import myproject.scheduling.jobs.ConnectionCheckRunner;

@ApplicationScoped
public class Scheduler {

    @Inject
    @Channel(Producers.MONITORING)
    Emitter<String> emitter;

    @Inject
    ConnectionChecks checks;

    @Scheduled(every = "1s", identity = Jobs.CONNECTION_CHECK)
    void scheduleInternetCheck() {
        List<ConnectionStatus> result = new ConnectionCheckRunner(checks).run();
        result.forEach((entry) -> System.out.println(entry.toLineProtocol()));
        result.forEach((entry) -> emitter.send(entry.toLineProtocol()));
    }
}

错误日志 #

2024-12-01 22:13:47,731 ERROR [io.qua.sch.com.run.StatusEmitterInvoker] (executor-thread-1) Error occurred while executing task for trigger io.quarkus.quartz.runtime.QuartzSchedulerImpl$QuartzTrigger@7f01115e: java.util.concurrent.CompletionException: java.lang.RuntimeException: Error injecting org.eclipse.microprofile.reactive.messaging.Emitter<java.lang.String> myproject.scheduling.Scheduler.emitter
    at java.base/java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:332)
    at java.base/java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:347)
    at java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:874)
    at java.base/java.util.concurrent.CompletableFuture.uniWhenCompleteStage(CompletableFuture.java:887)
    at java.base/java.util.concurrent.CompletableFuture.whenComplete(CompletableFuture.java:2357)
    at java.base/java.util.concurrent.CompletableFuture$MinimalStage.whenComplete(CompletableFuture.java:2948)
    at io.quarkus.scheduler.common.runtime.DefaultInvoker.invoke(DefaultInvoker.java:25)
    at io.quarkus.scheduler.common.runtime.DelegateInvoker.invokeDelegate(DelegateInvoker.java:29)
    at io.quarkus.scheduler.common.runtime.StatusEmitterInvoker.invoke(StatusEmitterInvoker.java:35)
    at io.quarkus.scheduler.common.runtime.DelegateInvoker.invokeDelegate(DelegateInvoker.java:29)
    at io.quarkus.scheduler.common.runtime.DelegateInvoker.invokeComplete(DelegateInvoker.java:36)
    at io.quarkus.scheduler.common.runtime.OffloadingInvoker$2.call(OffloadingInvoker.java:54)
    at io.quarkus.scheduler.common.runtime.OffloadingInvoker$2.call(OffloadingInvoker.java:51)
    at io.vertx.core.impl.ContextImpl.lambda$executeBlocking$4(ContextImpl.java:192)
    at io.vertx.core.impl.ContextInternal.dispatch(ContextInternal.java:270)
    at io.vertx.core.impl.ContextImpl$1.execute(ContextImpl.java:221)
    at io.vertx.core.impl.WorkerTask.run(WorkerTask.java:56)
    at io.quarkus.vertx.core.runtime.VertxCoreRecorder$15.runWith(VertxCoreRecorder.java:642)
    at org.jboss.threads.EnhancedQueueExecutor$Task.doRunWith(EnhancedQueueExecutor.java:2675)
    at org.jboss.threads.EnhancedQueueExecutor$Task.run(EnhancedQueueExecutor.java:2654)
    at org.jboss.threads.EnhancedQueueExecutor.runThreadBody(EnhancedQueueExecutor.java:1627)
    at org.jboss.threads.EnhancedQueueExecutor$ThreadBody.run(EnhancedQueueExecutor.java:1594)
    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)
Caused by: java.lang.RuntimeException: Error injecting org.eclipse.microprofile.reactive.messaging.Emitter<java.lang.String> myproject.scheduling.Scheduler.emitter
    at myproject.scheduling.Scheduler_Bean.doCreate(Unknown Source)
    at myproject.scheduling.Scheduler_Bean.create(Unknown Source)
    at myproject.scheduling.Scheduler_Bean.create(Unknown Source)
    at io.quarkus.arc.impl.AbstractSharedContext.createInstanceHandle(AbstractSharedContext.java:119)
    at io.quarkus.arc.impl.AbstractSharedContext$1.get(AbstractSharedContext.java:38)
    at io.quarkus.arc.impl.AbstractSharedContext$1.get(AbstractSharedContext.java:35)
    at io.quarkus.arc.generator.Default_jakarta_enterprise_context_ApplicationScoped_ContextInstances.c3(Unknown Source)
    at io.quarkus.arc.generator.Default_jakarta_enterprise_context_ApplicationScoped_ContextInstances.computeIfAbsent(Unknown Source)
    at io.quarkus.arc.impl.AbstractSharedContext.get(AbstractSharedContext.java:35)
    at io.quarkus.arc.impl.ClientProxies.getApplicationScopedDelegate(ClientProxies.java:21)
    at myproject.scheduling.Scheduler_ClientProxy.arc$delegate(Unknown Source)
    at myproject.scheduling.Scheduler_ClientProxy.scheduleInternetCheck(Unknown Source)
    at myproject.scheduling.Scheduler_ScheduledInvoker_scheduleInternetCheck_db9533208f2a6decf3d622882f85ce81424bd3a1.invokeBean(Unknown Source)
    ... 20 more
Caused by: jakarta.enterprise.inject.spi.DefinitionException: SRMSG00019: Unable to connect an emitter with the channel `monitoring-producer`
    at io.smallrye.reactive.messaging.providers.extension.ChannelProducer.getEmitter(ChannelProducer.java:225)
    at io.smallrye.reactive.messaging.providers.extension.ChannelProducer.produceEmitter(ChannelProducer.java:173)
    at io.smallrye.reactive.messaging.providers.extension.ChannelProducer_ProducerMethod_produceEmitter_jRfYFqSs8A2Ams_fdlkwttyVQ3w_Bean.doCreate(Unknown Source)
    at io.smallrye.reactive.messaging.providers.extension.ChannelProducer_ProducerMethod_produceEmitter_jRfYFqSs8A2Ams_fdlkwttyVQ3w_Bean.create(Unknown Source)
    at io.smallrye.reactive.messaging.providers.extension.ChannelProducer_ProducerMethod_produceEmitter_jRfYFqSs8A2Ams_fdlkwttyVQ3w_Bean.get(Unknown Source)
    at io.smallrye.reactive.messaging.providers.extension.ChannelProducer_ProducerMethod_produceEmitter_jRfYFqSs8A2Ams_fdlkwttyVQ3w_Bean.get(Unknown Source)
    at io.quarkus.arc.impl.CurrentInjectionPointProvider.get(CurrentInjectionPointProvider.java:48)
    ... 33 more

正常工作的类示例 #

package myproject.rest;

import java.util.Map;

import org.eclipse.microprofile.reactive.messaging.Channel;
import org.eclipse.microprofile.reactive.messaging.Emitter;

import myproject.config.Constants.AmqpQueues.Producers;
import myproject.models.measurements.metrics.JobMetric;
import myproject.models.measurements.Data;
import io.smallrye.common.annotation.RunOnVirtualThread;
import jakarta.inject.Inject;
import jakarta.ws.rs.GET;
import jakarta.ws.rs.POST;
import jakarta.ws.rs.Path;
import jakarta.ws.rs.PathParam;

@Path("data")
@RunOnVirtualThread
public class DataResource extends Resource {

    @Inject
    @Channel(Producers.DATA)
    Emitter<String> emitter;

    @Inject
    @Channel(Producers.MONITORING)
    Emitter<String> metricEmitter;

    @POST
    public Map<String, String> postData(Data data) {
        JobMetric metric = startMetric("api_post_data");
        emitter.send(data.toLineProtocol());
        finishMetric(metric, metricEmitter);
        return getOkResponse("Uploaded data sucessfully");
    }

}

解决方案 #

方法1：添加虚拟线程注解 #

Quarkus默认用Vert.x Worker线程执行定时任务，而Reactive Messaging的Emitter依赖正确的CDI和Vert.x上下文。给定时任务类添加@RunOnVirtualThread注解，让任务在虚拟线程中执行（和正常工作的REST类一致），即可继承正确上下文完成注入：

@ApplicationScoped
@RunOnVirtualThread // 新增该注解
public class Scheduler {
    // ... 原有代码不变
}

方法2：手动获取Emitter #

如果不想使用虚拟线程，可以在定时方法内通过CDI上下文手动获取Emitter，避免注入时机的上下文问题：

@ApplicationScoped
public class Scheduler {

    @Inject
    ConnectionChecks checks;

    @Scheduled(every = "1s", identity = Jobs.CONNECTION_CHECK)
    void scheduleInternetCheck() {
        // 手动获取Emitter
        Emitter<String> emitter = CDI.current()
                .select(Emitter.class, new Channel.Literal(Producers.MONITORING))
                .get();
        
        List<ConnectionStatus> result = new ConnectionCheckRunner(checks).run();
        result.forEach((entry) -> System.out.println(entry.toLineProtocol()));
        result.forEach((entry) -> emitter.send(entry.toLineProtocol()));
    }
}

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