来瞧一瞧 gRPC的拦截器

[TOC]

瞧一瞧 gRPC的拦截器

上一次说到gRPC的认证总共有4种，其中介绍了经常使用且重要的2种：

• 可使用openssl作认证证书，进行认证
• 客户端还能够将数据放到metadata中，服务器进行认证

但是朋友们，有没有想过，要是每个客户端与服务端通讯的接口都进行一次认证，那么这是否会很是多余呢，且每个接口的实现都要作一次认证，这真的太难受了

咱做为程序员，就应该要探索高效的方法来解决一些繁琐复杂冗余的事情。

今天咱们来分享一下gRPC的interceptor，即拦截器 ，相似于web框架里的中间件。

中间件是什么？

是一类提供系统软件和应用软件之间链接、便于软件各部件之间的沟通的计算机软件，它为软件应用程序提供操做系统之外的服务，被形象的描述为“软件胶水”

直白的说，中间件便是一个系统软件和应用软件之间的沟通桥梁。例如他能够记录响应时长、记录请求和响应数据日志等

中间件能够在拦截到发送给 handler 的请求，且能够拦截 handler 返回给客户端的响应

拦截器是什么？

拦截器是gRPC生态中的中间件

能够对RPC的请求和响应进行拦截处理，并且既能够在客户端进行拦截，也能够对服务器端进行拦截。

拦截器能作什么？

哈哈，他能作的可多了，最终要的一点是，拦截器能够作统一接口的认证工做，不再须要每个接口都作一次认证了，多个接口屡次访问，只须要在统一个地方认证便可

这是否是大大的提升了接口的使用和认证效率了呢，同时还能够减小代码的冗余度

拦截器有哪些分类呢？

根据不一样的侧重点，会有以下2种分类：

侧重点不一样，分类的拦截器也不一样，不过使用的方式都是大同小异的。

如何使用拦截器？

服务端会用到的方法

UnaryServerInterceptor提供了一个钩子来拦截服务器上单一RPC的执行，拦截器负责调用处理程序来完成RPC

其中参数中的UnaryHandler定义了由UnaryServerInterceptor调用的处理程序

客户端会用到的方法

type UnaryClientInterceptor func(
    ctx context.Context,         // 上下文
    method string,                 // RPC的名字，例如此处咱们使用的是gRPC
    req, reply interface{},      // 对应的请求和响应消息
    cc *ClientConn,                // cc是调用RPC的ClientConn
    invoker UnaryInvoker,         // invoker是完成RPC的处理程序，主要是调用它是拦截器
    opts ...CallOption) error    // opts包含全部适用的调用选项，包括来自ClientConn的默认值以及每一个调用选项

总体案例代码结构

代码结构与上2篇分享到的结构一致，本次拦截器，是统一作认证，把认证的地方统一放在同一个位置，而不是分散到每个接口

若须要具体的proto源码，能够查看个人上一期文章，以下为代码结构图示

开始书写案例

• 在原有代码基础上加入interceptor的功能，目前案例中注册一个拦截器
• gRPC + openssl + token + interceptor

server.go

• 主要加入UnaryServerInterceptor来对拦截器的应用

package main

import (
   "fmt"
   "google.golang.org/grpc/codes"
   "google.golang.org/grpc/metadata"
   "log"
   "net"

   pb "myserver/protoc/hi"

   "golang.org/x/net/context"
   "google.golang.org/grpc"
   "google.golang.org/grpc/credentials" // 引入grpc认证包
)

const (
   // Address gRPC服务地址
   Address = "127.0.0.1:9999"
)

// 定义helloService并实现约定的接口
type HiService struct{}

// HiService Hello服务
var HiSer = HiService{}

// SayHello 实现Hello服务接口
func (h HiService) SayHi(ctx context.Context, in *pb.HiRequest) (*pb.HiResponse, error) {

   // 解析metada中的信息并验证
   md, ok := metadata.FromIncomingContext(ctx)
   if !ok {
      return nil, grpc.Errorf(codes.Unauthenticated, "no token ")
   }

   var (
      appId  string
      appKey string
   )

   // md 是一个 map[string][]string 类型的
   if val, ok := md["appid"]; ok {
      appId = val[0]
   }

   if val, ok := md["appkey"]; ok {
      appKey = val[0]
   }

   if appId != "myappid" || appKey != "mykey" {
      return nil, grpc.Errorf(codes.Unauthenticated, "token invalide: appid=%s, appkey=%s", appId, appKey)
   }

   resp := new(pb.HiResponse)
   resp.Message = fmt.Sprintf("Hi %s.", in.Name)

   return resp, nil
}

// 认证token
func myAuth(ctx context.Context) error {
   md, ok := metadata.FromIncomingContext(ctx)
   if !ok {
      return grpc.Errorf(codes.Unauthenticated, "no token ")
   }

   log.Println("myAuth ...")

   var (
      appId  string
      appKey string
   )

   // md 是一个 map[string][]string 类型的
   if val, ok := md["appid"]; ok {
      appId = val[0]
   }

   if val, ok := md["appkey"]; ok {
      appKey = val[0]
   }

   if appId != "myappid" || appKey != "mykey" {
      return grpc.Errorf(codes.Unauthenticated, "token invalide: appid=%s, appkey=%s", appId, appKey)
   }

   return nil
}

// interceptor 拦截器
func interceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
   // 进行认证
   log.Println("interceptor...")
   err := myAuth(ctx)
   if err != nil {
      return nil, err
   }

   // 继续处理请求
   return handler(ctx, req)
}

func main() {
   log.SetFlags(log.Ltime | log.Llongfile)

   listen, err := net.Listen("tcp", Address)
   if err != nil {
      log.Panicf("Failed to listen: %v", err)
   }

   var opts []grpc.ServerOption

   // TLS认证
   creds, err := credentials.NewServerTLSFromFile("./keys/server.pem", "./keys/server.key")
   if err != nil {
      log.Panicf("Failed to generate credentials %v", err)
   }

   opts = append(opts, grpc.Creds(creds))

   // 注册一个拦截器
   opts = append(opts, grpc.UnaryInterceptor(interceptor))

   // 实例化grpc Server, 并开启TLS认证，其中还有拦截器
   s := grpc.NewServer(opts...)

   // 注册HelloService
   pb.RegisterHiServer(s, HiSer)

   log.Println("Listen on " + Address + " with TLS and interceptor")

   s.Serve(listen)
}

client.go

• 主要加入UnaryClientInterceptor来对拦截器的应用

package main

import (
   "log"
   pb "myclient/protoc/hi" // 引入proto包
   "time"

   "golang.org/x/net/context"
   "google.golang.org/grpc"
   "google.golang.org/grpc/credentials" // 引入grpc认证包
   "google.golang.org/grpc/grpclog"
)

const (
   // Address gRPC服务地址
   Address = "127.0.0.1:9999"
)

var IsTls = true

// myCredential 自定义认证
type myCredential struct{}

// GetRequestMetadata 实现自定义认证接口
func (c myCredential) GetRequestMetadata(ctx context.Context, uri ...string) (map[string]string, error) {
   return map[string]string{
      "appid":  "myappid",
      "appkey": "mykey",
   }, nil
}

// RequireTransportSecurity 自定义认证是否开启TLS
func (c myCredential) RequireTransportSecurity() bool {
   return IsTls
}

// 客户端拦截器
func Clientinterceptor(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
   start := time.Now()
   err := invoker(ctx, method, req, reply, cc, opts...)
   log.Printf("method == %s ; req == %v ; rep == %v ; duration == %s ; error == %v\n", method, req, reply, time.Since(start), err)
   return err
}

func main() {
   log.SetFlags(log.Ltime | log.Llongfile)
   // TLS链接  记得把xxx改为你写的服务器地址

   var err error
   var opts []grpc.DialOption

   if IsTls {
      //打开tls 走tls认证
      creds, err := credentials.NewClientTLSFromFile("./keys/server.pem", "www.eline.com")
      if err != nil {
         log.Panicf("Failed to create TLS mycredentials %v", err)
      }
      opts = append(opts, grpc.WithTransportCredentials(creds))
   } else {
      opts = append(opts, grpc.WithInsecure())
   }

   // 自定义认证，new(myCredential 的时候，因为咱们实现了上述2个接口，所以new的时候，程序会执行咱们实现的接口
   opts = append(opts, grpc.WithPerRPCCredentials(new(myCredential)))

   // 加上拦截器
   opts = append(opts, grpc.WithUnaryInterceptor(Clientinterceptor))

   conn, err := grpc.Dial(Address, opts...)
   if err != nil {
      grpclog.Fatalln(err)
   }

   defer conn.Close()

   // 初始化客户端
   c := pb.NewHiClient(conn)

   // 调用方法
   req := &pb.HiRequest{Name: "gRPC"}
   res, err := c.SayHi(context.Background(), req)
   if err != nil {
      log.Panicln(err)
   }
   log.Println(res.Message)

   // 故意再调用一次
   res, err = c.SayHi(context.Background(), req)
   if err != nil {
      log.Panicln(err)
   }

   log.Println(res.Message)
}

实际效果展现

注意，服务器只能配置一个 UnaryInterceptor和StreamClientInterceptor，不然会报错，客户端也是，虽然不会报错，可是只有最后一个才起做用。 若是你想配置多个，可使用拦截器链，如go-grpc-middleware，或者本身实现。

• 服务端的拦截器

  • UnaryServerInterceptor -- 单向调用的拦截器
  • StreamServerInterceptor -- stream调用的拦截器

• 客户端的拦截器

  • UnaryClientInterceptor
  • StreamClientInterceptor

上述拦截器不管是单向调用的拦截器 仍是 stream调用的拦截器 用法都大同小异

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

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

// 客户端
type UnaryClientInterceptor func(ctx context.Context, method string, req, reply interface{}, cc *ClientConn, invoker UnaryInvoker, opts ...CallOption) error

type StreamClientInterceptor func(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, streamer Streamer, opts ...CallOption) (ClientStream, error)

最后分享社区内用到的拦截器（还应该有更多...）

最后与你们分享几个社区内用到的拦截器

用于身份验证拦截器

• grpc_auth: https://github.com/grpc-ecosy...

interceptor链式功能的库，能够将单向的或者流式的拦截器组合

• grpc-multi-interceptor: https://github.com/kazegusuri...
• go-grpc-middleware: https://github.com/grpc-ecosy...

为上下文增长Tag map对象

• grpc_ctxtags: https://github.com/grpc-ecosy...

日志框架

• grpc_zap: https://github.com/grpc-ecosy...
• logrus：https://github.com/grpc-ecosy...

能够为客户端增长重试的功能

• grpc_retry: https://github.com/grpc-ecosy...

好了，本次就到这里，下一次分享 gRPC的请求追踪，

技术是开放的，咱们的心态，更应是开放的。拥抱变化，向阳而生，努力向前行。

我是小魔童哪吒，欢迎点赞关注收藏，下次见~