Quarkus 下的 Panache 为数据库更新提供超时答案

【问题标题】：Panache under Quarkus giving timeout for db updateQuarkus 下的 Panache 为数据库更新提供超时
【发布时间】：2021-11-27 11:21:01
【问题描述】：

我的 Quarkus 应用程序中有以下 api：

  @PATCH
  @Path("/{id}")
  @Transactional
  public Uni<Response> updateCustomer(@RestPath Long id, MyPatchObject myBody) {
    return myService.updateCustomer(id, myBody);
  }

在我的服务中：

  public Uni<Response> updateCustomer(final long id, final MyPatchObject myBody) {
    return myRepo.findById(id) 
// some validation here
        .replaceWith(myRepo.update(id, myBody)
                           .chain(() -> myRepo.flush())
// reload to get the latest changes
                           .replaceWith(myRepo.findById(id)) 
// some other handling that persits to another repository
        .replaceWith(Response.noContent().build());
  }

还有我的存储库：

  public Uni<Integer> update(Long id, MyPatchObject params) {
    // Some handling which results in these constants:
    String query = "black_listed = :blacklisted where id = :id"
    Parameters params = ... // contains: [blacklisted-true, id-1]
    return this.update(query,params);
  }

但是，当我尝试通过测试调用此 api 时，我得到了这个超时异常

2021-10-07 12:40:28,911 ERROR [org.jbo.res.rea.com.cor.AbstractResteasyReactiveContext] (executor-thread-0) Request failed: java.lang.RuntimeException: java.util.concurrent.TimeoutException
    at io.quarkus.hibernate.reactive.panache.common.runtime.AbstractJpaOperations.executeInVertxEventLoop(AbstractJpaOperations.java:52)
    at io.smallrye.mutiny.operators.uni.UniRunSubscribeOn.subscribe(UniRunSubscribeOn.java:25)
    at io.smallrye.mutiny.operators.AbstractUni.subscribe(AbstractUni.java:36)
    at io.smallrye.mutiny.operators.uni.UniOnItemTransformToUni.subscribe(UniOnItemTransformToUni.java:25)
    at io.smallrye.mutiny.operators.AbstractUni.subscribe(AbstractUni.java:36)
    at io.smallrye.mutiny.operators.uni.UniOnItemTransformToUni.subscribe(UniOnItemTransformToUni.java:25)
    at io.smallrye.mutiny.operators.AbstractUni.subscribe(AbstractUni.java:36)
    at io.smallrye.mutiny.operators.uni.UniOnItemTransformToUni.subscribe(UniOnItemTransformToUni.java:25)
    at io.smallrye.mutiny.operators.AbstractUni.subscribe(AbstractUni.java:36)
    at io.smallrye.mutiny.operators.uni.UniOnItemTransformToUni$UniOnItemTransformToUniProcessor.performInnerSubscription(UniOnItemTransformToUni.java:81)
    at io.smallrye.mutiny.operators.uni.UniOnItemTransformToUni$UniOnItemTransformToUniProcessor.onItem(UniOnItemTransformToUni.java:57)
    at io.smallrye.mutiny.operators.uni.UniOnItemTransformToUni$UniOnItemTransformToUniProcessor.onItem(UniOnItemTransformToUni.java:60)
    at io.smallrye.mutiny.operators.uni.builders.DefaultUniEmitter.complete(DefaultUniEmitter.java:36)
    at io.smallrye.mutiny.groups.UniOnNull.lambda$failWith$1(UniOnNull.java:43)
    at io.smallrye.context.impl.wrappers.SlowContextualBiConsumer.accept(SlowContextualBiConsumer.java:21)
    at io.smallrye.mutiny.groups.UniOnItem.lambda$transformToUni$4(UniOnItem.java:169)
    at io.smallrye.context.impl.wrappers.SlowContextualConsumer.accept(SlowContextualConsumer.java:21)
    at io.smallrye.mutiny.operators.uni.builders.UniCreateWithEmitter.subscribe(UniCreateWithEmitter.java:22)
    at io.smallrye.mutiny.operators.AbstractUni.subscribe(AbstractUni.java:36)
    at io.smallrye.mutiny.operators.uni.UniOnItemTransformToUni$UniOnItemTransformToUniProcessor.performInnerSubscription(UniOnItemTransformToUni.java:81)
    at io.smallrye.mutiny.operators.uni.UniOnItemTransformToUni$UniOnItemTransformToUniProcessor.onItem(UniOnItemTransformToUni.java:57)
    at io.smallrye.mutiny.operators.uni.UniOnItemTransformToUni$UniOnItemTransformToUniProcessor.onItem(UniOnItemTransformToUni.java:60)
    at io.smallrye.mutiny.operators.uni.UniOperatorProcessor.onItem(UniOperatorProcessor.java:36)
    at io.smallrye.mutiny.operators.uni.builders.UniCreateFromCompletionStage$CompletionStageUniSubscription.forwardResult(UniCreateFromCompletionStage.java:63)
    at java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:859)
    at java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:837)
    at java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506)
    at java.base/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2143)
    at io.vertx.core.Future.lambda$toCompletionStage$2(Future.java:360)
    at io.vertx.core.impl.future.FutureImpl$3.onSuccess(FutureImpl.java:141)
    at io.vertx.core.impl.future.FutureBase.emitSuccess(FutureBase.java:60)
    at io.vertx.core.impl.future.FutureImpl.tryComplete(FutureImpl.java:211)
    at io.vertx.core.impl.future.PromiseImpl.tryComplete(PromiseImpl.java:23)
    at io.vertx.sqlclient.impl.QueryResultBuilder.tryComplete(QueryResultBuilder.java:102)
    at io.vertx.sqlclient.impl.QueryResultBuilder.tryComplete(QueryResultBuilder.java:35)
    at io.vertx.core.Promise.complete(Promise.java:66)
    at io.vertx.core.Promise.handle(Promise.java:51)
    at io.vertx.core.Promise.handle(Promise.java:29)
    at io.vertx.core.impl.future.FutureImpl$3.onSuccess(FutureImpl.java:141)
    at io.vertx.core.impl.future.FutureBase.lambda$emitSuccess$0(FutureBase.java:54)
    at io.vertx.core.impl.EventLoopContext.execute(EventLoopContext.java:81)
    at io.vertx.core.impl.DuplicatedContext.execute(DuplicatedContext.java:173)
    at io.vertx.core.impl.future.FutureBase.emitSuccess(FutureBase.java:51)
    at io.vertx.core.impl.future.FutureImpl.tryComplete(FutureImpl.java:211)
    at io.vertx.core.impl.future.PromiseImpl.tryComplete(PromiseImpl.java:23)
    at io.vertx.core.impl.future.PromiseImpl.onSuccess(PromiseImpl.java:49)
    at io.vertx.core.impl.future.PromiseImpl.handle(PromiseImpl.java:41)
    at io.vertx.core.impl.future.PromiseImpl.handle(PromiseImpl.java:23)
    at io.vertx.sqlclient.impl.command.CommandResponse.fire(CommandResponse.java:46)
    at io.vertx.sqlclient.impl.SocketConnectionBase.handleMessage(SocketConnectionBase.java:287)
    at io.vertx.pgclient.impl.PgSocketConnection.handleMessage(PgSocketConnection.java:96)
    at io.vertx.sqlclient.impl.SocketConnectionBase.lambda$init$0(SocketConnectionBase.java:99)
    at io.vertx.core.net.impl.NetSocketImpl.lambda$new$1(NetSocketImpl.java:97)
    at io.vertx.core.streams.impl.InboundBuffer.handleEvent(InboundBuffer.java:240)
    at io.vertx.core.streams.impl.InboundBuffer.write(InboundBuffer.java:130)
    at io.vertx.core.net.impl.NetSocketImpl.lambda$handleMessage$9(NetSocketImpl.java:390)
    at io.vertx.core.impl.EventLoopContext.emit(EventLoopContext.java:50)
    at io.vertx.core.impl.ContextImpl.emit(ContextImpl.java:274)
    at io.vertx.core.impl.EventLoopContext.emit(EventLoopContext.java:22)
    at io.vertx.core.net.impl.NetSocketImpl.handleMessage(NetSocketImpl.java:389)
    at io.vertx.core.net.impl.ConnectionBase.read(ConnectionBase.java:155)
    at io.vertx.core.net.impl.VertxHandler.channelRead(VertxHandler.java:154)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    at io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelRead(CombinedChannelDuplexHandler.java:436)
    at io.vertx.pgclient.impl.codec.PgEncoder.lambda$write$0(PgEncoder.java:87)
    at io.vertx.pgclient.impl.codec.PgCommandCodec.handleReadyForQuery(PgCommandCodec.java:139)
    at io.vertx.pgclient.impl.codec.PgDecoder.decodeReadyForQuery(PgDecoder.java:237)
    at io.vertx.pgclient.impl.codec.PgDecoder.channelRead(PgDecoder.java:96)
    at io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:251)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
    at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
    at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:719)
    at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:655)
    at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:581)
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:493)
    at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:986)
    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:831)
Caused by: java.util.concurrent.TimeoutException
    at java.base/java.util.concurrent.CompletableFuture.timedGet(CompletableFuture.java:1956)
    at java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2091)
    at io.quarkus.hibernate.reactive.panache.common.runtime.AbstractJpaOperations.executeInVertxEventLoop(AbstractJpaOperations.java:50)
    ... 86 more

我确实在 io.quarkus.hibernate.reactive.panache.common.runtime.AbstractJpaOperations 中找到了一些代码

    // FIXME: make it configurable?
    static final long TIMEOUT_MS = 5000;

这似乎相关但不确定是否是问题的根本原因。

我的更新查询正在运行以更新单个字段，按条目的 ID 过滤，在一个只有一条记录的表中，所以这肯定不是数据库优化的问题。什么可能导致错误？

【问题讨论】：

您是否知道在第一个 findById 完成之前正在执行更新？
更新返回一个uni，所以在此之前不应该执行？
返回 Uni 的方法在其他线程中执行一些（但不一定是全部）工作。 Unis 是懒惰的，只有在被订阅并且也许事情在你的代码中以正确的顺序被执行之后才会做他们的工作，但是你不能相信在异步代码之前可能有一些同步代码的 libs 实现。您应该将 replaceWith 用于同步代码，使用 call 和 chain 等方法用于异步代码。

标签： java timeout quarkus quarkus-panache

【解决方案1】：

我不能确定这是导致您的问题的原因，但您应该使用 Uni#replaceWith 和 Uni#invoke 进行同步，使用 Uni#call 和 Uni#chain 进行异步代码，如下所示：

public Uni<Customer> updateCustomer(final long id, final MyPatchObject myBody) {
    return myRepo.findById(id)
        .invoke(customer -> myValidationMethod(customer))
        .call(() -> myRepo.update(id, myBody)) // .call propagates the incoming Uni
        .call(() -> myRepo.flush())
        .chain(() -> myRepo.findById(id)); // .chain propagates the returned Uni
}

为了挑剔，不要从您的服务返回响应 :)

@PATCH
@Path("/{id}")
@Transactional
public Uni<Response> updateCustomer(@RestPath Long id, MyPatchObject myBody) {
    return myService.updateCustomer(id, myBody)
        .replaceWith(Response.noContent().build());
}

毕竟，Mutiny 有一个非常混乱的命名法......

【讨论】：