如何在 gRPC 服务器中添加全局异常拦截器？

Question

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

例如，一个 divide 方法可能会抛出 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();
}

如果客户端传递的分母为0，它将得到：

If the client passes denominator = 0 , it will get :

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())));
  }

如果客户端发送的分母为0，则会收到以下信息：

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

好的，客户端传递了/ by zero，但问题是，在真正的企业环境中，会有很多RuntimeException，如果我想把这些异常的信息传递给客户端，我就必须尝试捕获每个方法，这非常繁琐。

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

非常感谢！

注意：

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

- smallufo

Answer 1

以下代码将捕获所有运行时异常，还请参考链接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());
            }
         }
      };
   }
}

- mahesh

Answer 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.grpc 版本 1.30。

- burubum

Answer 3

TransmitStatusRuntimeExceptionInterceptor非常类似于您想要的，但它只捕获StatusRuntimeException。您可以fork它并使其捕获所有异常。

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

- Kun Zhang

Answer 4

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

- markfickett

Answer 5

您阅读过grpc java的拦截器示例吗？

在我的情况下，我们使用代码和消息作为定义服务器向客户端发送了什么类型的错误的标准。

例如：服务器发送响应如下：

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

在客户端中，您可以设置客户端拦截器来获取该代码并使用反射响应。值得注意的是，我们使用Lognet Spring Boot starter for grpc作为服务器，使用Spring Boot作为客户端。

- Reza Adha Hamonangan

Answer 6

在 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")
}

- Maximiliano De Lorenzo

Answer 7

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);
}

遗憾的是，不同的线程上下文意味着没有异常处理范围，所以我的答案不是你要找的解决方案。

- Gladmir

---

抱歉，它不起作用。这行代码 logger.error ("Uncaught exception from grpc service"); 没有被执行到！我也感到很奇怪。 - smallufo

拦截确实会发生，但文档指出 next.startCall(call, headers) 立即返回并在另一个线程中执行，最终我们会失去异常的堆栈信息。很遗憾我不知道目前是否有任何解决方法。 - Gladmir

Answer 8

如果您可以使用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()
      }
    }

- Pankaj

Answer 9

我已经使用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 "";
    }

}

```

2. 在所有rpc方法中添加@WrapError注解 @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"); //模拟异常 responseObserver.onNext(reply); responseObserver.onCompleted(); }

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

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

4.testing

- Iceberg