grpc-go编解码器使用说明

[TOC]

参考资料:

1. gRPC 官方文档中文版 V1.0

1. 利用protobuf生成文件

1.1 service.proto

syntax = "proto3";
// The greeting service definition.
service Greeter {
 // Sends a greeting
 rpc SayHello (HelloRequest) returns (HelloReply) {}
}
// The request message containing the user's name.
message HelloRequest {
 string name = 1;
}
// The response message containing the greetings
message HelloReply {
 string message = 1;
}

利用命令生成相应的service.pb.go
protoc --go_out=./go/ ./proto/helloworld.proto
如果生成的某个proto文件需要import其他几个proto需要 -I指定搜索目录
protoc --go_out=./go/ -I proto ./proto/helloworld.proto
protoc --go_out=plugins=grpc:. helloworld.proto

1.2 grpc简单调用

go get "google.golang.org/grpc"

1.2.1 服务端

const (
 port = ":50051"
)
// server is used to implement helloworld.GreeterServer.
type server struct{}
// SayHello implements helloworld.GreeterServer
func (s *server) SayHello(ctx context.Context, in *pb.HelloRequest) (*pb.HelloReply, error) {
 return &pb.HelloReply{Message: "Hello " + in.Name}, nil
}
func Test_Server(t *testing.T) {
 lis, err := net.Listen("tcp", port)
 if err != nil {
 log.Fatalf("failed to listen: %v", err)
 }
 s := grpc.NewServer()
 pb.RegisterGreeterServer(s, &server{})
 // Register reflection service on gRPC server.
 reflection.Register(s)
 if err := s.Serve(lis); err != nil {
 log.Fatalf("failed to serve: %v", err)
 }
}

1.2.2 客户端

const (
 address = "localhost:50051"
 defaultName = "world"
)
 
func main() {
 conn, err := grpc.Dial(address, grpc.WithInsecure())
 if err != nil {
 log.Fatal("did not connect: %v", err)
 }
 defer conn.Close()
 c := pb.NewGreeterClient(conn)
 
 name := defaultName
 if len(os.Args) >1 {
 name = os.Args[1]
 }
 r, err := c.SayHello(context.Background(), &pb.HelloRequest{Name: name})
 if err != nil {
 log.Fatal("could not greet: %v", err)
 }
 log.Printf("Greeting: %s", r.Message)
}

2. 自定义编码器

2.1 服务端编解码

2.1.1 服务器处理数据 processUnaryRPC

func (s *Server) processUnaryRPC(t transport.ServerTransport, stream *transport.Stream, srv *service, md *MethodDesc, trInfo *traceInfo) (err error) {
 // 不看上面
 df := func(v interface{}) error {
 // 获取解码器解码
 if err := s.getCodec(stream.ContentSubtype()).Unmarshal(d, v); err != nil {
 return status.Errorf(codes.Internal, "grpc: error unmarshalling request: %v", err)
 }
 if sh != nil {
 sh.HandleRPC(stream.Context(), &stats.InPayload{
 RecvTime: time.Now(),
 Payload: v,
 Data: d,
 Length: len(d),
 })
 }
 if binlog != nil {
 binlog.Log(&binarylog.ClientMessage{
 Message: d,
 })
 }
 if trInfo != nil {
 trInfo.tr.LazyLog(&payload{sent: false, msg: v}, true)
 }
 return nil
 }
 // 不看下面
}

解码过程中主要通过调用s.getCodec(stream.ContentSubtype())获取编码器

2.1.2 服务端返回数据 sendResponse

func (s *Server) sendResponse(t transport.ServerTransport, stream *transport.Stream, msg interface{}, cp Compressor, opts *transport.Options, comp encoding.Compressor) error {
 // 对返回消息进行编码
 data, err := encode(s.getCodec(stream.ContentSubtype()), msg)
 if err != nil {
 grpclog.Errorln("grpc: server failed to encode response: ", err)
 return err
 }
 // 对数据进行压缩
 compData, err := compress(data, cp, comp)
 if err != nil {
 grpclog.Errorln("grpc: server failed to compress response: ", err)
 return err
 }
 // 添加消息头
 hdr, payload := msgHeader(data, compData)
 // TODO(dfawley): should we be checking len(data) instead?
 if len(payload) > s.opts.maxSendMessageSize {
 return status.Errorf(codes.ResourceExhausted, "grpc: trying to send message larger than max (%d vs. %d)", len(payload), s.opts.maxSendMessageSize)
 }
 // 发送消息
 err = t.Write(stream, hdr, payload, opts)
 if err == nil && s.opts.statsHandler != nil {
 s.opts.statsHandler.HandleRPC(stream.Context(), outPayload(false, msg, data, payload, time.Now()))
 }
 return err
}

服务端消息返回时,主要完成了以下几个过程

1. 编码
2. 压缩
3. 添加消息头
   编码过程中主要通过调用s.getCodec(stream.ContentSubtype())获取编码器

2.1.3 [关键]编解码器获取 getCodec

func (s *Server) getCodec(contentSubtype string) baseCodec {
 // 服务端指定编码器
 if s.opts.codec != nil {
 return s.opts.codec
 }
 // 请求头中获取使用编码器,并在本地编码表中查找是否存在
 // 从编码表中获取对应名称的编码表,或者返回默认的proto编码器
 if contentSubtype == "" {
 return encoding.GetCodec(proto.Name)
 }
 codec := encoding.GetCodec(contentSubtype)
 if codec == nil {
 return encoding.GetCodec(proto.Name)
 }
 return codec
}

编解码器的获取有两种方式

1. 由服务端在初始化Server时,指定使用的codec
2. 在encoding中注册请求头中对应的编码器
   对于方式二,由于请求头contenSubtype参数未知,暂且不考虑

2.2 客户端编解码

2.2.1 客户端发送数据 SendMsg

func (cs *clientStream) SendMsg(m interface{}) (err error) {
 defer func() {
 if err != nil && err != io.EOF {
 cs.finish(err)
 }
 }()
 if cs.sentLast {
 return status.Errorf(codes.Internal, "SendMsg called after CloseSend")
 }
 if !cs.desc.ClientStreams {
 cs.sentLast = true
 }
 // 编码
 data, err := encode(cs.codec, m)
 if err != nil {
 return err
 }
 // 压缩
 compData, err := compress(data, cs.cp, cs.comp)
 if err != nil {
 return err
 }
 // 添加请求头
 hdr, payload := msgHeader(data, compData)
 if len(payload) > *cs.callInfo.maxSendMessageSize {
 return status.Errorf(codes.ResourceExhausted, "trying to send message larger than max (%d vs. %d)", len(payload), *cs.callInfo.maxSendMessageSize)
 }
 msgBytes := data 
 op := func(a *csAttempt) error {
 err := a.sendMsg(m, hdr, payload, data)
 m, data = nil, nil
 return err
 }
 err = cs.withRetry(op, func() { cs.bufferForRetryLocked(len(hdr)+len(payload), op) })
 if cs.binlog != nil && err == nil {
 cs.binlog.Log(&binarylog.ClientMessage{
 OnClientSide: true,
 Message: msgBytes,
 })
 }
 return
}

客户端发送数据过程与服务端一样

1. 编码
2. 压缩
3. 添加请求头
   不同之处在于客户端编解码器直接由客户端提供即cs.codec

2.2.2 客户端接收数据 recvMsg

func (a *csAttempt) recvMsg(m interface{}, payInfo *payloadInfo) (err error) {
 // 不看上面
 err = recv(a.p, cs.codec, a.s, a.dc, m, *cs.callInfo.maxReceiveMessageSize, payInfo, a.decomp)
 // 不看下面
}

客户端接收到数据后,也是使用cs.codec进行解码

2.3 grpc.Codec接口

type Codec interface {
 // Marshal returns the wire format of v.
 Marshal(v interface{}) ([]byte, error)
 // Unmarshal parses the wire format into v.
 Unmarshal(data []byte, v interface{}) error
 // String returns the name of the Codec implementation. This is unused by
 // gRPC.
 String() string
}

编解码器主要需要实现以上三个接口

2.3.1 自定义一个JSON的编解码器

type JSONCoder struct {}
func (j *JSONCoder) Marshal(v interface{}) ([]byte, error) {
 log.Println("JSONCoder Marshal")
 return json.Marshal(v)
}
func (j *JSONCoder) Unmarshal(data []byte, v interface{}) error{
 log.Println("JSONCoder UnMarshal")
 return json.Unmarshal(data, v)
}
func (j *JSONCoder) String() string {
 log.Println("JSONCoder String")
 return "JSONCoder"
}

2.4 自定义编解码器使用

2.4.1 服务端使用

func Test_Server(t *testing.T) {
 lis, err := net.Listen("tcp", port)
 if err != nil {
 log.Fatalf("failed to listen: %v", err)
 }
 // 与正常的差别,是在新建服务时指定需要的自定义编解码器
 s := grpc.NewServer(grpc.CustomCodec(&JSONCoder{}))
 pb.RegisterGreeterServer(s, &server{})
 reflection.Register(s)
 if err := s.Serve(lis); err != nil {
 log.Fatalf("failed to serve: %v", err)
 }
}

2.4.2 客户端使用

func main() {
 conn, err := grpc.Dial(address, grpc.WithInsecure())
 if err != nil {
 log.Fatal("did not connect: %v", err)
 }
 defer conn.Close()
 c := pb.NewGreeterClient(conn)
 
 name := defaultName
 if len(os.Args) >1 {
 name = os.Args[1]
 }
 // 与正常差别,在调用rpc接口时,指定自定义编解码
 r, err := c.SayHello(context.Background(), &pb.HelloRequest{Name: name}, grpc.CallCustomCodec(&JSONCoder{}))
 if err != nil {
 log.Fatal("could not greet: %v", err)
 }
 log.Printf("Greeting: %s", r.Message)
}

2.5 使用Thirft作为编解码器

由以上分析可知,若是需要使用Thirft或者其他进行编解码过程,只需要自定义一个对应的编解码器进行替换

thrifter项目地址

GO灵活编解码Thirft

有疑问加站长微信联系(非本文作者)

• 请尽量让自己的回复能够对别人有帮助
• 支持 Markdown 格式, **粗体**、~~删除线~~、`单行代码`
• 支持 @ 本站用户;支持表情(输入 : 提示),见 Emoji cheat sheet
• 图片支持拖拽、截图粘贴等方式上传