Unary and Stream interceptor

前言

我想在每个 RPC 方法的前或后做某些事情,怎么做?

本章节将要介绍的拦截器(interceptor),就能帮你在合适的地方实现这些功能。

有几种方法

在 gRPC 中,大类可分为两种 RPC 方法,与拦截器的对应关系是:

  • 普通方法:一元拦截器(grpc.UnaryInterceptor)
  • 流方法:流拦截器(grpc.StreamInterceptor)

看一看

grpc.UnaryInterceptor

  1. func UnaryInterceptor(i UnaryServerInterceptor) ServerOption {
  2. return func(o *options) {
  3. if o.unaryInt != nil {
  4. panic("The unary server interceptor was already set and may not be reset.")
  5. }
  6. o.unaryInt = i
  7. }
  8. }

函数原型:

  1. type UnaryServerInterceptor func(ctx context.Context, req interface{}, info *UnaryServerInfo, handler UnaryHandler) (resp interface{}, err error)

通过查看源码可得知,要完成一个拦截器需要实现 UnaryServerInterceptor 方法。形参如下:

  • ctx context.Context:请求上下文
  • req interface{}:RPC 方法的请求参数
  • info *UnaryServerInfo:RPC 方法的所有信息
  • handler UnaryHandler:RPC 方法本身

grpc.StreamInterceptor

  1. func StreamInterceptor(i StreamServerInterceptor) ServerOption

函数原型:

  1. type StreamServerInterceptor func(srv interface{}, ss ServerStream, info *StreamServerInfo, handler StreamHandler) error

StreamServerInterceptor 与 UnaryServerInterceptor 形参的意义是一样,不再赘述

如何实现多个拦截器

另外,可以发现 gRPC 本身居然只能设置一个拦截器,难道所有的逻辑都只能写在一起?

关于这一点,你可以放心。采用开源项目 go-grpc-middleware 就可以解决这个问题,本章也会使用它。

  1. import "github.com/grpc-ecosystem/go-grpc-middleware"
  2. myServer := grpc.NewServer(
  3. grpc.StreamInterceptor(grpc_middleware.ChainStreamServer(
  4. ...
  5. )),
  6. grpc.UnaryInterceptor(grpc_middleware.ChainUnaryServer(
  7. ...
  8. )),
  9. )

gRPC

从本节开始编写 gRPC interceptor 的代码,我们会将实现以下拦截器:

  • logging:RPC 方法的入参出参的日志输出
  • recover:RPC 方法的异常保护和日志输出

实现 interceptor

logging

  1. func LoggingInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
  2. log.Printf("gRPC method: %s, %v", info.FullMethod, req)
  3. resp, err := handler(ctx, req)
  4. log.Printf("gRPC method: %s, %v", info.FullMethod, resp)
  5. return resp, err
  6. }

recover

  1. func RecoveryInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
  2. defer func() {
  3. if e := recover(); e != nil {
  4. debug.PrintStack()
  5. err = status.Errorf(codes.Internal, "Panic err: %v", e)
  6. }
  7. }()
  8. return handler(ctx, req)
  9. }

Server

  1. import (
  2. "context"
  3. "crypto/tls"
  4. "crypto/x509"
  5. "errors"
  6. "io/ioutil"
  7. "log"
  8. "net"
  9. "runtime/debug"
  10. "google.golang.org/grpc"
  11. "google.golang.org/grpc/credentials"
  12. "google.golang.org/grpc/status"
  13. "google.golang.org/grpc/codes"
  14. "github.com/grpc-ecosystem/go-grpc-middleware"
  15. pb "github.com/EDDYCJY/go-grpc-example/proto"
  16. )
  17. ...
  18. func main() {
  19. c, err := GetTLSCredentialsByCA()
  20. if err != nil {
  21. log.Fatalf("GetTLSCredentialsByCA err: %v", err)
  22. }
  23. opts := []grpc.ServerOption{
  24. grpc.Creds(c),
  25. grpc_middleware.WithUnaryServerChain(
  26. RecoveryInterceptor,
  27. LoggingInterceptor,
  28. ),
  29. }
  30. server := grpc.NewServer(opts...)
  31. pb.RegisterSearchServiceServer(server, &SearchService{})
  32. lis, err := net.Listen("tcp", ":"+PORT)
  33. if err != nil {
  34. log.Fatalf("net.Listen err: %v", err)
  35. }
  36. server.Serve(lis)
  37. }

验证

logging

启动 simple_server/server.go,执行 simple_client/client.go 发起请求,得到结果:

  1. $ go run server.go
  2. 2018/10/02 13:46:35 gRPC method: /proto.SearchService/Search, request:"gRPC"
  3. 2018/10/02 13:46:35 gRPC method: /proto.SearchService/Search, response:"gRPC Server"

recover

在 RPC 方法中人为地制造运行时错误,再重复启动 server/client.go,得到结果:

client

  1. $ go run client.go
  2. 2018/10/02 13:19:03 client.Search err: rpc error: code = Internal desc = Panic err: assignment to entry in nil map
  3. exit status 1

server

  1. $ go run server.go
  2. goroutine 23 [running]:
  3. runtime/debug.Stack(0xc420223588, 0x1033da9, 0xc420001980)
  4. /usr/local/Cellar/go/1.10.1/libexec/src/runtime/debug/stack.go:24 +0xa7
  5. runtime/debug.PrintStack()
  6. /usr/local/Cellar/go/1.10.1/libexec/src/runtime/debug/stack.go:16 +0x22
  7. main.RecoveryInterceptor.func1(0xc420223a10)
  8. ...

检查服务是否仍然运行,即可知道 Recovery 是否成功生效

总结

通过本章节,你可以学会最常见的拦截器使用方法。接下来其它“新”需求只要举一反三即可。

参考

本系列示例代码