一.服务端拦截器server端原理
serverOptions配置中的Interceptor,其中unary为一元拦截器,stream为流式拦截器。
本文只看一元式拦截器,即最常见的客户端向服务器发送单个请求并返回单个响应。
创建一个新的grpc server时,这个方法将拦截器进行了组合封装
chainUnaryServerInterceptors具体代码,现将unaryInt和数组chainUnaryInts进行了合并,拦截器数量小于等于1的逻辑很好理解,重点看一下len>1的情况,chainUnaryInterceptors这个函数将拦截器数据递归封装为一个总的拦截器,这个总的拦截器包含数组中所有拦截器的逻辑。
下面是封装过程,chainUnaryInterceptors函数输入为拦截器数组,输出为封装好的总拦截器。具体逻辑返回新封装的拦截器,新封装的拦截器的结果为执行数组中第一个拦截器的结果。第一个拦截器的结果实际上是执行handler返回的。而这里的handler又要通过getChainUnaryHandler来获取。getChainUnaryHandler获取handler为一个递归过程,根据curr的位置调用下一个拦截器,直到执行最后一个拦截器的handler即finalHandler,这个finalHandler就是是我们grpc的具体方法。这样形成一个函数的嵌套,保障执行完所有的拦截器的逻辑,最后调用grpc的具体方法。
以上只是组装了拦截器,我们看具体调用逻辑
Serve里代码太长,直接看最重要的,监听tcp返回一个连接,然后开个gorutine处理客户端请求
创建http2 transport后再次开启goruotine处理流
一路跟下来,根据客户端的method匹配的服务注册的具体grpc方法,并执行。具体执行的代码写在创建的pb文件中。
我的proto文件
syntax = "proto3";
package protocol_demo;
option go_package = "lib/proto/protocol_demo";
message HelloWorldReq {
int64 Id = 1;
}
message HelloWorldRsp {
string Message = 2;
}
service HelloWorld {
rpc GetHelloWorld (HelloWorldReq) returns (HelloWorldRsp) {
}
}
生成的bp文件,注册服务的时候将方法名和对应的方法添加到Methods中,本例中对应的方法为_HelloWorld_GetHelloWorld_Handler。上面的md.Handler就是执行的这个方法,看到如果没有拦截器直接执行具体方法。如果有拦截器,则执行拦截器即:return interceptor
这个intercepor就是上文中封装的总拦截器。
以上就是grpc server端拦截器实现的源码逻辑。
二. 实现一个拦截器
package main
import (
"context"
"fmt"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
"gopractice/lib/proto/protocol_demo"
"net"
"runtime/debug"
"time"
)
func main() {
lis, err := net.Listen("tcp", ":50052")
if err != nil {
fmt.Printf("failed to listen: %s \n", err)
return
}
opts := []grpc.ServerOption{
grpc.ChainUnaryInterceptor(RecoveryInterceptor, LoggerInterceptor),
}
s := grpc.NewServer(opts...)
protocol_demo.RegisterHelloWorldServer(s, &server{})
fmt.Println("success")
err = s.Serve(lis)
if err != nil {
fmt.Printf("failed to start grpc server: %s \n", err)
return
}
}
type server struct {
}
func (s *server) GetHelloWorld(ctx context.Context, req *protocol_demo.HelloWorldReq) (rsp *protocol_demo.HelloWorldRsp, err error) {
rsp = &protocol_demo.HelloWorldRsp{}
rsp.Message = fmt.Sprintf("hello user: %d", req.Id)
fmt.Printf("hello method\n")
return rsp, nil
}
func LoggerInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
t := time.Now()
fmt.Printf("gRpc begin method: method: %s | req: %v | time: %s", info.FullMethod, req, t.Format("2006-01-02 15:04:05.000000"))
fmt.Println()
resp, err = handler(ctx, req)
fmt.Printf("gRpc finish method: %s | rsp: %v | time: %s | durations: %s", info.FullMethod, resp, req, time.Since(t))
fmt.Println()
return
}
func RecoveryInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
defer func() {
if e := recover(); e != nil {
debug.PrintStack()
err = status.Errorf(codes.Internal, "Panic err: %v | %s", e, string(debug.Stack()))
fmt.Println()
}
}()
fmt.Printf("RecoveryInterceptor in\n")
resp, err = handler(ctx, req)
fmt.Printf("RecoveryInterceptor out\n")
return
}
三.客户端拦截器
客户端与服务端类似,直接贴相关代码
组装拦截器的逻辑是一样的,只不过handler变为了invoker
执行调用是有拦截器执行拦截器逻辑,没有直接invoke
创建一个streaming RPC直接发送接收相关数据完成调用
四. 客户端代码实现
package main
import (
"context"
"fmt"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/connectivity"
"google.golang.org/grpc/credentials/insecure"
"google.golang.org/grpc/status"
"gopractice/lib/proto/protocol_demo"
"runtime/debug"
"time"
)
func main() {
conn, err := grpc.Dial("localhost:50052", grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithChainUnaryInterceptor(LoggerInterceptor, RecoveryInterceptor))
if err != nil {
fmt.Printf("new grpc client failed: %s \n", err)
return
}
defer conn.Close()
c := protocol_demo.NewHelloWorldClient(conn)
r, err := c.GetHelloWorld(context.Background(), &protocol_demo.HelloWorldReq{Id: 100})
if err != nil {
fmt.Printf("request GetHelloWorld faild: %s \n", err)
return
}
fmt.Printf("success, message is : %s \n", r.Message)
}
func LoggerInterceptor(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
t := time.Now()
fmt.Printf("gRpc begin method: method: %s | req: %v | time: %s", method, req, t.Format("2006-01-02 15:04:05.000000"))
fmt.Println()
err := invoker(ctx, method, req, reply, cc, opts...)
fmt.Printf("gRpc finish method: %s | rsp: %v | time: %s | durations: %s", method, reply, t.Format("2006-01-02 15:04:05.000000"), time.Since(t))
fmt.Println()
return err
}
func RecoveryInterceptor(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) (err error) {
defer func() {
if e := recover(); e != nil {
debug.PrintStack()
err = status.Errorf(codes.Internal, "Panic err: %v | %s", e, string(debug.Stack()))
fmt.Println()
}
}()
fmt.Printf("RecoveryInterceptor in\n")
err = invoker(ctx, method, req, reply, cc, opts...)
fmt.Printf("RecoveryInterceptor out\n")
return
}