30

在 gRPC 中，如何添加一个全局异常拦截器来拦截任何RuntimeException并向客户端传播有意义的信息？

例如，一个方法可能会divide抛出message 。在服务器端，我可以写：ArithmeticException/ by zero

@Override
public void divide(DivideRequest request, StreamObserver<DivideResponse> responseObserver) {
  int dom = request.getDenominator();
  int num = request.getNumerator();

  double result = num / dom;
  responseObserver.onNext(DivideResponse.newBuilder().setValue(result).build());
  responseObserver.onCompleted();
}

如果客户端通过 denominator = 0 ，它将得到：

Exception in thread "main" io.grpc.StatusRuntimeException: UNKNOWN

和服务器输出

Exception while executing runnable io.grpc.internal.ServerImpl$JumpToApplicationThreadServerStreamListener$2@62e95ade
java.lang.ArithmeticException: / by zero

客户不知道发生了什么。

如果我想将/ by zero消息传递给客户端，我必须将服务器修改为：（如本问题所述）

  try {
    double result = num / dom;
    responseObserver.onNext(DivideResponse.newBuilder().setValue(result).build());
    responseObserver.onCompleted();
  } catch (Exception e) {
    logger.error("onError : {}" , e.getMessage());
    responseObserver.onError(new StatusRuntimeException(Status.INTERNAL.withDescription(e.getMessage())));
  }

如果客户端发送 denominator = 0 ，它将得到：

Exception in thread "main" io.grpc.StatusRuntimeException: INTERNAL: / by zero

好，/ by zero传给客户。

但问题是，在真正的企业环境中，会有很多RuntimeExceptions ，如果我想把这些异常的消息传递给客户端，我必须尝试每一个方法来捕获，这非常麻烦。

是否有任何全局拦截器可以拦截每个方法，捕获RuntimeException并触发onError并将错误消息传播给客户端？这样我就不必RuntimeException在我的服务器代码中处理 s 了。

非常感谢 ！

笔记 ：

<grpc.version>1.0.1</grpc.version>
com.google.protobuf:proton:3.1.0
io.grpc:protoc-gen-grpc-java:1.0.1
4

9 回答 9

7

下面的代码将捕获所有运行时异常，另请参阅链接https://github.com/grpc/grpc-java/issues/1552

public class GlobalGrpcExceptionHandler implements ServerInterceptor {

   @Override
   public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> call,
         Metadata requestHeaders, ServerCallHandler<ReqT, RespT> next) {
      ServerCall.Listener<ReqT> delegate = next.startCall(call, requestHeaders);
      return new SimpleForwardingServerCallListener<ReqT>(delegate) {
         @Override
         public void onHalfClose() {
            try {
               super.onHalfClose();
            } catch (Exception e) {
               call.close(Status.INTERNAL
                .withCause (e)
                .withDescription("error message"), new Metadata());
            }
         }
      };
   }
}
于 2018-05-30T09:50:24.413 回答
3

TransmitStatusRuntimeExceptionInterceptor与您想要的非常相似，只是它只捕获StatusRuntimeException. 您可以分叉它并使其捕获所有异常。

要为服务器上的所有服务安装拦截器，您可以使用ServerBuilder.intercept()gRPC 1.5.0 中添加的

于 2018-05-30T18:44:17.757 回答
2

如果您想在所有 gRPC 端点（包括处理客户端流的端点）和拦截器中捕获异常，您可能需要类似于以下内容的内容：

import io.grpc.ForwardingServerCallListener;
import io.grpc.Metadata;
import io.grpc.ServerCall;
import io.grpc.ServerCallHandler;
import io.grpc.ServerInterceptor;
import io.grpc.Status;

public class GlobalGrpcExceptionHandler implements ServerInterceptor {

    @Override
    public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> serverCall,
                                                                 Metadata requestHeaders,
                                                                 ServerCallHandler<ReqT, RespT> serverCallHandler) {
        try {
            ServerCall.Listener<ReqT> delegate = serverCallHandler.startCall(serverCall, requestHeaders);
            return new ForwardingServerCallListener.SimpleForwardingServerCallListener<ReqT>(delegate) {
                @Override
                public void onMessage(ReqT message) {
                    try {
                        super.onMessage(message); // Here onNext is called (in case of client streams)
                    } catch (Throwable e) {
                        handleEndpointException(e, serverCall);
                    }
                }

                @Override
                public void onHalfClose() {
                    try {
                        super.onHalfClose(); // Here onCompleted is called (in case of client streams)
                    } catch (Throwable e) {
                        handleEndpointException(e, serverCall);
                    }
                }
            };
        } catch (Throwable t) {
            return handleInterceptorException(t, serverCall);
        }
    }

    private <ReqT, RespT> void handleEndpointException(Throwable t, ServerCall<ReqT, RespT> serverCall) {
        serverCall.close(Status.INTERNAL
                         .withCause(t)
                         .withDescription("An exception occurred in the endpoint implementation"), new Metadata());
    }

    private <ReqT, RespT> ServerCall.Listener<ReqT> handleInterceptorException(Throwable t, ServerCall<ReqT, RespT> serverCall) {
        serverCall.close(Status.INTERNAL
                         .withCause(t)
                         .withDescription("An exception occurred in a **subsequent** interceptor"), new Metadata());

        return new ServerCall.Listener<ReqT>() {
            // no-op
        };
    }
}

免责声明：我通过检查实现收集了这个，我没有在文档中阅读它，我不确定它是否会改变。作为参考，我指的是io.grpcversion 1.30。

于 2021-02-16T15:14:11.463 回答
2

在 Kotlin 中，您必须以不同的方式构建 ServerInterceptor。我在 Micronaut 中使用了 grpc-kotlin，异常从未出现在SimpleForwardingServerCallListener onHalfClose或其他处理程序中。

于 2020-11-13T14:13:58.897 回答
1

你读过拦截器的 grpc java例子吗？

所以在我的例子中，我们使用代码和消息作为标准来定义服务器发送到客户端的错误类型。

示例：服务器发送响应，如

{
  code: 409,
  message: 'Id xxx aldready exist'
}

因此，在客户端中，您可以设置客户端拦截器以使用反射获取该代码和响应。仅供参考，我们将Lognet Spring Boot 启动器用于 grpc作为服务器，将 Spring Boot 用于客户端。

于 2017-09-18T02:36:25.800 回答
0 
public class GrpcExceptionHandler implements ServerInterceptor {

private final Logger logger = LoggerFactory.getLogger (GrpcExceptionHandler.class);

@Override
public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall (ServerCall<ReqT, RespT> call,
                                                              Metadata headers,
                                                              ServerCallHandler<ReqT, RespT> next) {
    logger.info ("GRPC call at: {}", Instant.now ());
    ServerCall.Listener<ReqT> listener;

    try {
        listener = next.startCall (call, headers);
    } catch (Throwable ex) {
        logger.error ("Uncaught exception from grpc service");
        call.close (Status.INTERNAL
                .withCause (ex)
                .withDescription ("Uncaught exception from grpc service"), null);
        return new ServerCall.Listener<ReqT>() {};
    }

    return listener;
}

}

上面的示例拦截器。

当然，你需要先引导它，然后再期待它的任何东西。

serverBuilder.addService (ServerInterceptors.intercept (bindableService, interceptor));

更新

public interface ServerCallHandler<RequestT, ResponseT> {
  /**
   * Produce a non-{@code null} listener for the incoming call. Implementations are free to call
   * methods on {@code call} before this method has returned.
   *
   * <p>If the implementation throws an exception, {@code call} will be closed with an error.
   * Implementations must not throw an exception if they started processing that may use {@code
   * call} on another thread.
   *
   * @param call object for responding to the remote client.
   * @return listener for processing incoming request messages for {@code call}
   */
  ServerCall.Listener<RequestT> startCall(
      ServerCall<RequestT, ResponseT> call,
      Metadata headers);
}

可悲的是，不同的线程上下文意味着没有异常处理范围，所以我的答案不是您正在寻找的解决方案..

于 2016-10-04T08:54:49.193 回答
0

如果您可以使用yidongnan/grpc-spring-boot-starter将您的 gRPC 服务器应用程序转换为 Spring Boot，那么您可以编写 @GrpcAdvice，类似于 Spring Boot @ControllerAdvice 为

    @GrpcAdvice
    public class ExceptionHandler {
    
      @GrpcExceptionHandler(ValidationErrorException.class)
      public StatusRuntimeException handleValidationError(ValidationErrorException cause) {
    
         Status.INVALID_ARGUMENT.withDescription("Invalid Argument")
            .asRuntimeException()
      }
    }
于 2021-05-30T16:16:56.850 回答
0

在 Kotlin 上，在侦听器的方法上添加 try/catch 不起作用，由于某种原因，异常被吞没了。

在@markficket 发布的链接之后，我实现了一个解决方案，创建了一个SimpleForwardingServerCall.

class ErrorHandlerServerInterceptor : ServerInterceptor {

    private inner class StatusExceptionHandlingServerCall<ReqT, RespT>(delegate: ServerCall<ReqT, RespT>) 
        : ForwardingServerCall.SimpleForwardingServerCall<ReqT, RespT>(delegate) {

        override fun close(status: Status, trailers: Metadata) {
            status.run {
                when {
                    isOk -> status
                    cause is MyException -> myExceptionHandler(cause as MyException)
                    else -> defaultExceptionHandler(cause)
                }
            }
                .let { super.close(it, trailers) }
        }

        private fun myExceptionHandler(cause: MyException): Status = cause.run { ... }

        private fun defaultExceptionHandler(cause: Throwable?): Status = cause?.run { ... }

    }

    override fun <ReqT : Any, RespT : Any> interceptCall(
        call: ServerCall<ReqT, RespT>,
        metadata: Metadata,
        next: ServerCallHandler<ReqT, RespT>
    ): ServerCall.Listener<ReqT> =
        next.startCall(StatusExceptionHandlingServerCall(call), metadata)

}

然后当然需要在服务器创建时添加拦截器

ServerBuilder
  .forPort(port)
  .executor(Dispatchers.IO.asExecutor())
  .addService(service)
  .intercept(ErrorHandlerServerInterceptor())
  .build()

然后你可以简单地在你的 gRPC 方法上抛出异常

override suspend fun someGrpcCall(request: Request): Response {
  ... code ...
  throw NotFoundMyException("Cannot find entity")
}
于 2021-07-19T08:51:10.873 回答
-1

我已经使用 AOP 来处理全局的 rpc 错误，我觉得它很方便。我在guice中使用AOP，在spring中使用的方式应该是类似的

  1. 定义一个方法拦截器

```

public class ServerExceptionInterceptor implements MethodInterceptor {
    final static Logger logger = LoggerFactory.getLogger(ServerExceptionInterceptor.class);

    public Object invoke(MethodInvocation invocation) throws Throwable {
        try {
            return  invocation.proceed();
        } catch (Exception ex) {
            String stackTrace = Throwables.getStackTraceAsString(ex);
            logger.error("##grpc server side error, {}", stackTrace);
            Object[] args = invocation.getArguments();
            StreamObserver<?> responseObserver = (StreamObserver<?>)args[1];
            responseObserver.onError(Status.INTERNAL
                    .withDescription(stackTrace)
                    .withCause(ex)
                    .asRuntimeException());

            return null;
        }
    }

    @Target({ElementType.METHOD, ElementType.TYPE}) @Retention(RUNTIME)
    public @interface WrapError {
        String value() default "";
    }

}

```

  1. 将 @WrapError 添加到所有 rpc 方法 @Override @WrapError public void sayHello(HelloRequest req, StreamObserver<HelloReply> responseObserver) { HelloReply reply = HelloReply.newBuilder().setMessage("Hello " + req.getName()).build(); logger.info("#rpc server, sayHello, planId: {}", req.getName()); if(true) throw new RuntimeException("testing-rpc-error"); //simulate an exception responseObserver.onNext(reply); responseObserver.onCompleted(); }

    1. 在 guice 模块中绑定拦截器

ServerExceptionInterceptor interceptor = new ServerExceptionInterceptor(); requestInjection(interceptor); bindInterceptor(Matchers.any(), Matchers.annotatedWith(WrapError.class), interceptor);

4.测试

于 2018-05-07T05:36:38.010 回答