Grpc Stream 在写入发生之前关闭

Question

我有一个应用程序可以根据客户的请求通过 GRPC 从数据库中传输数据。

当客户端在流式传输完成之前关闭时。错误在日志中抛出，但我不知道如何捕获它以关闭底层数据库连接，从而导致数据库连接泄漏。

我正在使用 Java 和 grpc-java 版本 1.23.0.

  try {
... events is a stream retrieved from DB
            StreamObservers.copyWithFlowControl(response, new StreamObserverWithCallbacks<>(
                    responseObserver,
                    count -> {
                        LOGGER.info("Streaming finished);
                        events.close();
                    },
                    ex -> {
                        LOGGER.error("Error streaming",  ex);
                        events.close();
                    }
            ));
        } catch (Exception e) {
            LOGGER.error("Any other exception", e);
            events.close();
        }

这是我关闭客户端时日志中的异常

2019-08-27 18:39:34,155 WARN  [grpc-nio-worker-ELG-3-1] i.g.n.s.i.g.n.NettyServerHandler: Stream Error
io.grpc.netty.shaded.io.netty.handler.codec.http2.Http2Exception$StreamException: Stream closed before write could take place
    at io.grpc.netty.shaded.io.netty.handler.codec.http2.Http2Exception.streamError(Http2Exception.java:167)
    at io.grpc.netty.shaded.io.netty.handler.codec.http2.DefaultHttp2RemoteFlowController$FlowState.cancel(DefaultHttp2RemoteFlowController.java:481)
    at io.grpc.netty.shaded.io.netty.handler.codec.http2.DefaultHttp2RemoteFlowController.onStreamClosed(DefaultHttp2RemoteFlowController.java:105)
    at io.grpc.netty.shaded.io.netty.handler.codec.http2.DefaultHttp2Connection.notifyClosed(DefaultHttp2Connection.java:356)
    at io.grpc.netty.shaded.io.netty.handler.codec.http2.DefaultHttp2Connection$ActiveStreams.removeFromActiveStreams(DefaultHttp2Connection.java:1000)
    at io.grpc.netty.shaded.io.netty.handler.codec.http2.DefaultHttp2Connection$ActiveStreams.deactivate(DefaultHttp2Connection.java:956)
    at io.grpc.netty.shaded.io.netty.handler.codec.http2.DefaultHttp2Connection$DefaultStream.close(DefaultHttp2Connection.java:512)
    at io.grpc.netty.shaded.io.netty.handler.codec.http2.DefaultHttp2Connection.close(DefaultHttp2Connection.java:152)
    at io.grpc.netty.shaded.io.netty.handler.codec.http2.Http2ConnectionHandler$BaseDecoder.channelInactive(Http2ConnectionHandler.java:209)
    at io.grpc.netty.shaded.io.netty.handler.codec.http2.Http2ConnectionHandler.channelInactive(Http2ConnectionHandler.java:417)
    at io.grpc.netty.shaded.io.grpc.netty.NettyServerHandler.channelInactive(NettyServerHandler.java:586)
    at io.grpc.netty.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:257)
    at io.grpc.netty.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:243)
    at io.grpc.netty.shaded.io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:236)
    at io.grpc.netty.shaded.io.netty.channel.DefaultChannelPipeline$HeadContext.channelInactive(DefaultChannelPipeline.java:1416)
    at io.grpc.netty.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:257)
    at io.grpc.netty.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:243)
    at io.grpc.netty.shaded.io.netty.channel.DefaultChannelPipeline.fireChannelInactive(DefaultChannelPipeline.java:912)
    at io.grpc.netty.shaded.io.netty.channel.AbstractChannel$AbstractUnsafe.run(AbstractChannel.java:816)
    at io.grpc.netty.shaded.io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:163)
    at io.grpc.netty.shaded.io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:416)
    at io.grpc.netty.shaded.io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:515)
    at io.grpc.netty.shaded.io.netty.util.concurrent.SingleThreadEventExecutor.run(SingleThreadEventExecutor.java:918)
    at io.grpc.netty.shaded.io.netty.util.internal.ThreadExecutorMap.run(ThreadExecutorMap.java:74)
    at io.grpc.netty.shaded.io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
    at java.base/java.lang.Thread.run(Thread.java:834)

如您所见，none 个异常已在 try/catch 块中或在流式传输错误时被捕获。

我是不是漏掉了什么？有什么我可以做的吗？

Answer 1

你看到的异常其实没问题；它的出现是因为一些 gRPC 和 Netty 内部原因。涉及取消时会发生这种情况。

您目前缺少的是观察取消本身。使用观察取消。

Answer 2

StreamObserverWithCallbacks 是我们自己的 class，它包裹了一个 StreamObserver 委托，所以我们只需要做 set

this.delegate.setOnCancelHandler(() -> onError.accept(new RuntimeException("Stream got cancelled")));

Grpc Stream 在写入发生之前关闭

Grpc Stream closed before write could take place

java

stream

grpc

grpc-java