Interceptor攔截器 -- gRPC生態裏的中間件

什麼是攔截器

gRPC的攔截器（interceptor）相似各類Web框架裏的請求中間件，請求中間件你們都知道是利用裝飾器模式對最終處理請求的handler程序進行裝飾，這樣中間件就能夠在處理請求前和完成處理後這兩個時機上，攔截到發送給 handler 的請求以及 handler 返回給客戶端的響應 。 中間件的最大的用處是能夠把一些 handler 的前置和後置操做從 handler 程序中解耦出來，好比最多見的記錄響應時長、記錄請求和響應數據日誌等操做每每是經過中間件程序實現的。 與 Web 框架的中間件同理，能夠對gRPC的請求和響應進行攔截處理，並且既能夠在客戶端進行攔截，也能夠對服務器端進行攔截。利用攔截器，能夠對gRPC進行很好的擴展，把一些業務邏輯外的冗餘操做從 handler 中抽離，提高項目的開發效率和擴展性。

怎麼使用攔截器

gRPC的服務器和客戶端都是分別能夠添加一個單向調用 (Unary) 的攔截器和流式調用 (Stream) 的攔截器。

這兩種調用方式的區別能夠理解爲HTTP和WebSocket的區別

對於客戶端的單向調用的攔截，只需定義一個 UnaryClientInterceptor 方法：

type UnaryClientInterceptor func(ctx context.Context, method string, req, reply interface{}, cc *ClientConn, invoker UnaryInvoker, opts ...CallOption) error
複製代碼

而客戶端流式調用的攔截，則須要定義一個 StreamClientInterceptor 方法：

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

同理，對於gRPC的服務端也有這兩種調用的攔截器方法，分別是 UnaryServerInterceptor 和 StreamServerInterceptor：

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
複製代碼

攔截器應用

下面簡單演示一下，怎麼用客戶端和服務端攔截器來實現gRPC客戶端調用日誌，和gRPC服務器訪問日誌的。

首先咱們定義一下客戶端單向調用的攔截器方法：

func UnaryClientInterceptor(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) (err error) {

	p := peer.Peer{}
	if opts == nil {
		opts = []grpc.CallOption{grpc.Peer(&p)}
	} else {
		opts = append(opts, grpc.Peer(&p))
	}

	start := time.Now()
	defer func() {
		in, _ := json.Marshal(req)
		out, _ := json.Marshal(reply)
		inStr, outStr := string(in), string(out)
		duration := int64(time.Since(start) / time.Millisecond)

		var remoteServer string
		if p.Addr != nil {
			remoteServer=p.Addr.String()
		}

		log.Println("grpc", method, "in", inStr, "out", outStr, "err", err, "duration/ms", duration, "remote_server", remoteServer)

	}()

	return invoker(ctx, method, req, reply, cc, opts...)
}
複製代碼

建立客戶端的時候應用上這個方法：

var client routeGuideClient

func init() {
	var err error
	client.cc, err = grpc.Dial(
		"127.0.0.1:12305",
		grpc.WithInsecure(),
		grpc.WithUnaryInterceptor(UnaryClientInterceptor),
	)
	if err != nil {
		panic(err)
	}

}
複製代碼

routeguide 這個服務名是本身起的，其實就是拿的 grpc 官方的示例稍微改動了一下作的試驗。

接下來定義一個服務器端的攔截器：

func UnaryServerInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
	remote, _ := peer.FromContext(ctx)
	remoteAddr := remote.Addr.String()

	in, _ := json.Marshal(req)
	inStr := string(in)
	log.Println("ip", remoteAddr, "access_start", info.FullMethod, "in", inStr)

	start := time.Now()
	defer func() {
		out, _ := json.Marshal(resp)
		outStr := string(out)
		duration := int64(time.Since(start) / time.Millisecond)
		if duration >= 500 {
			log.Println("ip", remoteAddr, "access_end", info.FullMethod, "in", inStr, "out", outStr, "err", err, "duration/ms", duration)
		} else {
			log.Println("ip", remoteAddr, "access_end", info.FullMethod, "in", inStr, "out", outStr, "err", err, "duration/ms", duration)
		}
	}()

	resp, err = handler(ctx, req)

	return
}
複製代碼

在服務器啓動時應用上這個單向調用的攔截器：

var (
	port = flag.Int("p", 12305, "port")
)

type server struct {}

func (s server) Ping(ctx context.Context, request *routeguide.PingRequest) (reply *routeguide.PingReply, err error) {
	reply = &routeguide.PingReply{
		Reply:                "pong",
	}

	return
}

func main() {
	lis, err := net.Listen("tcp", fmt.Sprintf(":%d", *port))
	if err != nil {
		log.Fatalf("failed to listen: %v", err)
	}

	s := grpc.NewServer(grpc.UnaryInterceptor(UnaryServerInterceptor))
	routeguide.RegisterRouteGuideServer(s, &server{})
	s.Serve(lis)
}

複製代碼

啓動服務器，用客戶端調用後能夠看到在服務器和客戶端採集到的日誌：

// 服務器日誌
2021/04/24 17:24:44 ip 127.0.0.1:57258 access_start /routeguide.RouteGuide/Ping in {}
2021/04/24 17:24:44 ip 127.0.0.1:57258 access_end /routeguide.RouteGuide/Ping in {} out {"reply":"pong"} err <nil> duration/ms 0

// 客戶端日誌
2021/04/24 17:41:11 grpc /routeguide.RouteGuide/Ping in {} out {"reply":"pong"} err <nil> duration/ms 1 
複製代碼

與Web框架的中間件不一樣的是，Web框架能夠給每一個 handler 程序應用多箇中間件，可是gRPC的客戶端和服務器分別能夠添加一個單向調用類型的攔截器和流式調用類型的攔截器。不過gRPC社區裏Go gRPC Middleware 這個軟件包提供了攔截器的interceptor鏈式的功能，能夠將多個攔截器組合成一個攔截器鏈。

var client routeGuideClient

func init() {
	var err error
	client.cc, err = grpc.Dial(
		"127.0.0.1:12305",
		grpc.WithInsecure(),
		grpc.WithUnaryInterceptor(grpc_middleware.ChainUnaryClient(UnaryClientInterceptor)), // 參數裏能夠添加多個攔截器
	)
	if err != nil {
		panic(err)
	}

}
複製代碼

上面的示例程序爲了便於理解作了部分刪減，完整可運行的源碼能夠訪問 GitHub 連接得到。

社區裏那些實用的攔截器

利用攔截器，能夠對gRPC進行擴展，利用社區的力量將gRPC發展壯大，也可讓開發者更靈活地處理gRPC流程中的業務邏輯。下面列出了利用攔截器實現的一些功能框架：

1. Go gRPC Middleware:提供了攔截器的interceptor鏈式的功能，能夠將多個攔截器組合成一個攔截器鏈，固然它還提供了其它的功能，因此以gRPC中間件命名。
2. grpc-multi-interceptor: 是另外一個interceptor鏈式功能的庫，也能夠將單向的或者流式的攔截器組合。
3. grpc_auth: 身份驗證攔截器
4. grpc_ctxtags: 爲上下文增長Tag map對象
5. grpc_zap: 支持zap日誌框架
6. grpc_logrus: 支持logrus日誌框架
7. grpc_prometheus: 支持 prometheus
8. otgrpc: 支持opentracing/zipkin
9. grpc_opentracing:支持opentracing/zipkin
10. grpc_retry: 爲客戶端增長重試的功能
11. grpc_validator: 爲服務器端增長校驗的功能

今天的文章就到這裏啦，若是喜歡個人文章就幫我點個贊吧，我會每週經過技術文章分享個人所學所見和第一手實踐經驗，感謝你的支持。微信搜索關注公衆號「網管叨bi叨」每週教會你一個進階知識。