在服务器上调用onNext时，gRPC“呼叫已关闭”异常

Question

不幸的是，这种情况在生产中偶尔发生，但我无法可靠地复制。

gRPC服务器向少数客户端发送小而频繁的更新。每个客户端使用不同的参数对同一个调用发出多个请求。永远不会有来自服务器的数据流。

调用onNext时，出现以下错误：

Feb 20, 2019 10:13:03 AM io.grpc.internal.SerializingExecutor run
SEVERE: Exception while executing runnable concord.grpc.UncaughtExceptionServerInterceptor$SerializingServerCall$$Lambda$47/1901113624@2b4ca8e3
java.lang.IllegalStateException: call is closed
at com.google.common.base.Preconditions.checkState(Preconditions.java:174)
at io.grpc.internal.ServerCallImpl.sendMessage(ServerCallImpl.java:124)
at io.grpc.ForwardingServerCall.sendMessage(ForwardingServerCall.java:32)
at concord.grpc.UncaughtExceptionServerInterceptor$SerializingServerCall.access$1001(UncaughtExceptionServerInterceptor.java:142)
at concord.grpc.UncaughtExceptionServerInterceptor$SerializingServerCall.lambda$sendMessage$0(UncaughtExceptionServerInterceptor.java:158)
at io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:123)
at com.google.common.util.concurrent.MoreExecutors$DirectExecutor.execute(MoreExecutors.java:456)
at io.grpc.internal.SerializingExecutor.schedule(SerializingExecutor.java:93)
at io.grpc.internal.SerializingExecutor.execute(SerializingExecutor.java:86)
at concord.grpc.UncaughtExceptionServerInterceptor$SerializingServerCall.sendMessage(UncaughtExceptionServerInterceptor.java:158)
at io.grpc.stub.ServerCalls$ServerCallStreamObserverImpl.onNext(ServerCalls.java:341)
at myPackage$helper.lambda$calculateChangesAndNotifyObservers$1(myCode.java:229)

根据gRPC订阅数据的请求，保留观察员名单。多个客户端可能会请求完全相同的更新，因此每个进入的客户端都会被添加到相应数据的列表中。

如果他们取消，他们将通过以下方式从列表中删除：

if( responseObserver instanceof ServerCallStreamObserver<?> )
{
    ((ServerCallStreamObserver<?>) responseObserver).setOnCancelHandler( () ->
    {
        synchronized( _lastSnapshot )
        {
            _observers.remove( responseObserver );
        }                       
    } );
}

捕捉从onNext抛出的异常并简单地从列表中删除客户端是否明智？还是有更好的检测方法？还是我需要解决一个潜在的问题？

Answer 1

我今天也犯了这个错误。因为我在调用build（）后更改了生成器值。错误：

SyncAepUserResponse.Builder builder = SyncAepUserResponse.newBuilder();
builder.setStatus(Status.newBuilder().setCode(200).setMessage("Successful"))
                    .setMapId(entity.getId()).build();
builder.setCode("200");
responseObserver.onNext(response);
responseObserver.onCompleted();

正确的：

SyncAepUserResponse.Builder builder = SyncAepUserResponse.newBuilder();
builder.setCode("200");
builder.setStatus(Status.newBuilder().setCode(200).setMessage("Successful"))
                    .setMapId(entity.getId()).build();
responseObserver.onNext(response);
responseObserver.onCompleted();