gRPC 未来回调出现异常：StatusRuntimeException：CANCELED

Question

io.grpc.StatusRuntimeException: CANCELLED
    at io.grpc.Status.asRuntimeException(Status.java:539)
    at io.grpc.stub.ClientCalls$UnaryStreamToFuture.onClose(ClientCalls.java:439)
    at io.grpc.internal.ClientCallImpl.closeObserver(ClientCallImpl.java:422)
    at io.grpc.internal.ClientCallImpl.access$100(ClientCallImpl.java:74)
    at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl.close(ClientCallImpl.java:508)
    at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl.access$700(ClientCallImpl.java:425)
    at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInContext(ClientCallImpl.java:540)
    at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:52)
    at io.grpc.internal.SerializeReentrantCallsDirectExecutor.execute(SerializeReentrantCallsDirectExecutor.java:64)
    at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl.closed(ClientCallImpl.java:544)
    at io.grpc.internal.AbstractClientStream2$TransportState.closeListener(AbstractClientStream2.java:307)
    at io.grpc.internal.AbstractClientStream2$TransportState.transportReportStatus(AbstractClientStream2.java:287)
    at io.grpc.netty.NettyClientHandler.cancelStream(NettyClientHandler.java:455)
    at io.grpc.netty.NettyClientHandler.write(NettyClientHandler.java:231)
    at io.netty.channel.AbstractChannelHandlerContext.invokeWrite0(AbstractChannelHandlerContext.java:739)
    at io.netty.channel.AbstractChannelHandlerContext.invokeWrite(AbstractChannelHandlerContext.java:731)
    at io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:817)
    at io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:724)
    at io.netty.channel.DefaultChannelPipeline.write(DefaultChannelPipeline.java:1022)
    at io.netty.channel.AbstractChannel.write(AbstractChannel.java:291)
    at io.grpc.netty.WriteQueue.flush(WriteQueue.java:124)
    at io.grpc.netty.WriteQueue.access$000(WriteQueue.java:46)
    at io.grpc.netty.WriteQueue$1.run(WriteQueue.java:58)
    at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:163)
    at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:403)
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:445)
    at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:858)
    at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:144)
    at java.lang.Thread.run(Thread.java:745)

当我在带有未来和回调的 grpc 服务器句柄进程中调用另一个 grpc 服务时，我遇到了上述异常。

我的代码是这样的：

handle(){

....

***FutureClient futureClient = ...
ListenableFuture<***> future = futureClient.call(...);
Futures.addCallback(future, new FutureCallback<>() {...});

...

}

进程handle()完成后访问被取消。

我该如何解决这个问题？

Answer 1

来自服务器的传出 RPC 通过io.grpc.Context 隐式绑定到触发它们的传入请求。如果您正在执行一个即使在传入 RPC 完成后仍继续进行的 RPC，您应该使用context.fork()：

Context forked = Context.current().fork();
Context old = forked.attach();
try {
  // RPCs at this point can continue after the incoming RPC completes
  futureClient.call(...);
} finally {
  forked.detach(old);
}

Context.run(Runnable) 也是直接使用附加/分离的便捷替代方案。