0

我正在使用 Quarkus，但命令式进程运行缓慢：

@ApplicationScoped
public class SlowApplication {

private static final Logger LOGGER = Logger.getLogger(SlowApplication.class.getName());

@Inject @Channel("texto") public Emitter<String> emitter;

public Integer slowProcess(int num) {
    for (int n=0;n<num;n++) {
        LOGGER.info("Start Num: "+n);
        try {
            Thread.sleep(250);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        emitter.send("Number is: "+n);
        LOGGER.info("End Num: "+n);
    }
    return num;
}

我想通过使用 REST API 从流程中获取更新：

@Path("/test")
@ApplicationScoped
public class MainWeb {

@Inject
SlowApplication app;

@Inject
Bridge bridge;

@Inject @Channel("texto-sent") public Multi<String> textos;

@GET
@Produces(MediaType.TEXT_HTML)
public String hello() {
    return "hello";
}

@GET
@Produces(MediaType.TEXT_HTML)
@Path("/custom")
public Uni<String> custom() {
    return Uni.createFrom().item(app.slowProcess(4)).onItem().transform(n -> String.format("Finished %s", n));
}

@GET
@Produces(MediaType.TEXT_HTML)
@Path("/stream")
public Multi<String> stream() {
    return textos;
}

}

@ApplicationScoped
public class Bridge {
private static final Logger LOGGER = Logger.getLogger(Bridge.class.getName());

@Outgoing("texto-sent")
@Incoming("texto")
public String receive(String str) {
    LOGGER.infof("Received " + str);
    return "Received " + str;
}

到达时出现以下错误：

emitter.send("Number is: "+n);

java.lang.IllegalStateException：SRMSG00027：找不到频道文本的订阅者

链接到Github 代码 （包括 pom 和依赖版本）

知道为什么@Incoming("texto")不被识别吗？

完全错误：

java.lang.IllegalStateException: SRMSG00027: No subscriber found for the channel texto
    at io.smallrye.reactive.messaging.extension.AbstractEmitter.verify(AbstractEmitter.java:157)
    at io.smallrye.reactive.messaging.extension.AbstractEmitter.emit(AbstractEmitter.java:139)
    at io.smallrye.reactive.messaging.extension.EmitterImpl.send(EmitterImpl.java:29)
    at org.jds.sandbox.createUni.SlowApplication.slowProcess(SlowApplication.java:25)
    at org.jds.sandbox.createUni.SlowApplication_Subclass.slowProcess$$superaccessor1(SlowApplication_Subclass.zig:204)
    at org.jds.sandbox.createUni.SlowApplication_Subclass$$function$$1.apply(SlowApplication_Subclass$$function$$1.zig:35)
    at io.quarkus.arc.impl.AroundInvokeInvocationContext.proceed(AroundInvokeInvocationContext.java:54)
    at io.quarkus.arc.runtime.devconsole.InvocationInterceptor.proceed(InvocationInterceptor.java:63)
    at io.quarkus.arc.runtime.devconsole.InvocationInterceptor.monitor(InvocationInterceptor.java:49)
    at io.quarkus.arc.runtime.devconsole.InvocationInterceptor_Bean.intercept(InvocationInterceptor_Bean.zig:521)
    at io.quarkus.arc.impl.InterceptorInvocation.invoke(InterceptorInvocation.java:41)
    at io.quarkus.arc.impl.AroundInvokeInvocationContext.perform(AroundInvokeInvocationContext.java:41)
    at io.quarkus.arc.impl.InvocationContexts.performAroundInvoke(InvocationContexts.java:32)
    at org.jds.sandbox.createUni.SlowApplication_Subclass.slowProcess(SlowApplication_Subclass.zig:161)
    at org.jds.sandbox.createUni.SlowApplication_ClientProxy.slowProcess(SlowApplication_ClientProxy.zig:128)
    at org.jds.sandbox.createUni.MainWeb.custom(MainWeb.java:37)
    at org.jds.sandbox.createUni.MainWeb_Subclass.custom$$superaccessor1(MainWeb_Subclass.zig:237)
    at org.jds.sandbox.createUni.MainWeb_Subclass$$function$$1.apply(MainWeb_Subclass$$function$$1.zig:29)
    at io.quarkus.arc.impl.AroundInvokeInvocationContext.proceed(AroundInvokeInvocationContext.java:54)
    at io.quarkus.arc.runtime.devconsole.InvocationInterceptor.proceed(InvocationInterceptor.java:63)
    at io.quarkus.arc.runtime.devconsole.InvocationInterceptor.monitor(InvocationInterceptor.java:49)
    at io.quarkus.arc.runtime.devconsole.InvocationInterceptor_Bean.intercept(InvocationInterceptor_Bean.zig:521)
    at io.quarkus.arc.impl.InterceptorInvocation.invoke(InterceptorInvocation.java:41)
    at io.quarkus.arc.impl.AroundInvokeInvocationContext.perform(AroundInvokeInvocationContext.java:41)
    at io.quarkus.arc.impl.InvocationContexts.performAroundInvoke(InvocationContexts.java:32)
    at org.jds.sandbox.createUni.MainWeb_Subclass.custom(MainWeb_Subclass.zig:195)
    at org.jds.sandbox.createUni.MainWeb_ClientProxy.custom(MainWeb_ClientProxy.zig:213)
    at org.jds.sandbox.createUni.MainWeb$quarkusrestinvoker$custom_0989e6157a2c7c5acc3973ff53d46eb621670726.invoke(MainWeb$quarkusrestinvoker$custom_0989e6157a2c7c5acc3973ff53d46eb621670726.zig:33)
    at org.jboss.resteasy.reactive.server.handlers.InvocationHandler.handle(InvocationHandler.java:29)
    at org.jboss.resteasy.reactive.server.handlers.InvocationHandler.handle(InvocationHandler.java:7)
    at org.jboss.resteasy.reactive.common.core.AbstractResteasyReactiveContext.run(AbstractResteasyReactiveContext.java:122)
    at org.jboss.resteasy.reactive.server.handlers.RestInitialHandler.beginProcessing(RestInitialHandler.java:47)
    at org.jboss.resteasy.reactive.server.vertx.ResteasyReactiveVertxHandler.handle(ResteasyReactiveVertxHandler.java:17)
    at org.jboss.resteasy.reactive.server.vertx.ResteasyReactiveVertxHandler.handle(ResteasyReactiveVertxHandler.java:7)
    at io.vertx.ext.web.impl.RouteState.handleContext(RouteState.java:1038)
    at io.vertx.ext.web.impl.RoutingContextImplBase.iterateNext(RoutingContextImplBase.java:137)
    at io.vertx.ext.web.impl.RoutingContextImpl.next(RoutingContextImpl.java:132)
    at io.quarkus.vertx.http.runtime.StaticResourcesRecorder.lambda$start$1(StaticResourcesRecorder.java:65)
    at io.vertx.ext.web.impl.RouteState.handleContext(RouteState.java:1038)
    at io.vertx.ext.web.impl.RoutingContextImplBase.iterateNext(RoutingContextImplBase.java:101)
    at io.vertx.ext.web.impl.RoutingContextImpl.next(RoutingContextImpl.java:132)
    at io.vertx.ext.web.handler.impl.StaticHandlerImpl.lambda$sendStatic$1(StaticHandlerImpl.java:206)
    at io.vertx.core.impl.ContextImpl.lambda$null$0(ContextImpl.java:327)
    at io.vertx.core.impl.ContextImpl.executeTask(ContextImpl.java:366)
    at io.vertx.core.impl.EventLoopContext.lambda$executeAsync$0(EventLoopContext.java:38)
    at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:164)
    at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:472)
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:500)
    at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
    at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
    at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
    at java.base/java.lang.Thread.run(Thread.java:835)
4

1 回答 1

1

解决问题的方法是：

  • 使用MutinyEmitter代替Emitter
  • 使用emitter.sendAndForget代替emitter.send

SlowApplication 的新代码：

    @ApplicationScoped
public class SlowApplication {
    
    private static final Logger LOGGER = Logger.getLogger(SlowApplication.class.getName());
    
@Inject @Channel("texto") public MutinyEmitter<String> emitter;

public Integer slowProcess(int num) {
    for (int n=0;n<num;n++) {
        LOGGER.info("Start Num: "+n);
        try {
            Thread.sleep(250);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        emitter.sendAndForget("Number is: "+n);
        LOGGER.info("End Num: "+n);
    }
    return num;
}

我还将 MainWeb 方法中的流更改为：

@GET
@Path("/stream")
@Produces(MediaType.SERVER_SENT_EVENTS)
@SseElementType("text/html")
public Multi<String> stream() {
    return textos;
}

注意：我按照 Clement 的建议更新到 1.13.7.Final 版本。

于 2021-06-16T11:58:22.083 回答