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