一.服务端拦截器server端原理
serverOptions配置中的Interceptor,其中unary为一元拦截器,stream为流式拦截器。
本文只看一元式拦截器,即最常见的客户端向服务器发送单个请求并返回单个响应。
![](https://img-blog.csdnimg.cn/img_convert/88eeae905f0e6263d1e36eb39c4eba50.png)
创建一个新的grpc server时,这个方法将拦截器进行了组合封装
![](https://img-blog.csdnimg.cn/img_convert/8b9f6b4112f2da0d91f749d878514eea.png)
chainUnaryServerInterceptors具体代码,现将unaryInt和数组chainUnaryInts进行了合并,拦截器数量小于等于1的逻辑很好理解,重点看一下len>1的情况,chainUnaryInterceptors这个函数将拦截器数据递归封装为一个总的拦截器,这个总的拦截器包含数组中所有拦截器的逻辑。
![](https://img-blog.csdnimg.cn/img_convert/005f52a9df1fb7a3b93e62577126fdbb.png)
下面是封装过程,chainUnaryInterceptors函数输入为拦截器数组,输出为封装好的总拦截器。具体逻辑返回新封装的拦截器,新封装的拦截器的结果为执行数组中第一个拦截器的结果。第一个拦截器的结果实际上是执行handler返回的。而这里的handler又要通过getChainUnaryHandler来获取。getChainUnaryHandler获取handler为一个递归过程,根据curr的位置调用下一个拦截器,直到执行最后一个拦截器的handler即finalHandler,这个finalHandler就是是我们grpc的具体方法。这样形成一个函数的嵌套,保障执行完所有的拦截器的逻辑,最后调用grpc的具体方法。
![](https://img-blog.csdnimg.cn/img_convert/e07dde1634440f9d10236b0bcb302ee2.png)
![](https://img-blog.csdnimg.cn/img_convert/a6683b0e966e7ff601deea389e2eeba0.png)
以上只是组装了拦截器,我们看具体调用逻辑
![](https://img-blog.csdnimg.cn/img_convert/c23619aa22a3691ec17204579b164a9a.png)
Serve里代码太长,直接看最重要的,监听tcp返回一个连接,然后开个gorutine处理客户端请求
![](https://img-blog.csdnimg.cn/img_convert/4b4d5422ec59b0169d776aec42e31977.png)
![](https://img-blog.csdnimg.cn/img_convert/30e67b6b9b1853751a4e61dad740d309.png)
创建http2 transport后再次开启goruotine处理流
![](https://img-blog.csdnimg.cn/img_convert/22bccef5bac0740f113f262e0cc417f8.png)
![](https://img-blog.csdnimg.cn/img_convert/60793378ae59b7109462eccce2aea0e3.png)
一路跟下来,根据客户端的method匹配的服务注册的具体grpc方法,并执行。具体执行的代码写在创建的pb文件中。
![](https://img-blog.csdnimg.cn/img_convert/571c5cbe8dd00f05c32fc0bfe9490ce5.png)
![](https://img-blog.csdnimg.cn/img_convert/75f4f227f6c7fc25f572438fcf93fb43.png)
我的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就是上文中封装的总拦截器。
![](https://img-blog.csdnimg.cn/img_convert/fe072e8d2549871730b1714ae1e645e5.png)
以上就是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
}
![](https://img-blog.csdnimg.cn/img_convert/1c44c466584d6d174373485ae1fffde5.png)
三.客户端拦截器
客户端与服务端类似,直接贴相关代码
![](https://img-blog.csdnimg.cn/img_convert/d67d2ea9c06d9e113c8a79e130052875.png)
组装拦截器的逻辑是一样的,只不过handler变为了invoker
![](https://img-blog.csdnimg.cn/img_convert/eea81c66fd153060fb3f4c3f99f0c75d.png)
执行调用是有拦截器执行拦截器逻辑,没有直接invoke
![](https://img-blog.csdnimg.cn/img_convert/4fdc3d901f1a6de58ec227734352cc7e.png)
创建一个streaming RPC直接发送接收相关数据完成调用
![](https://img-blog.csdnimg.cn/img_convert/ce16c5a76535980ee8b399312bfa16c2.png)
四. 客户端代码实现
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
}
![](https://img-blog.csdnimg.cn/img_convert/73eff4b63705780a7facfe2e66786f9a.png)