内容简介:[TOC]参考资料:
[TOC]
参考资料:
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 }
服务端消息返回时,主要完成了以下几个过程
- 编码
- 压缩
-
添加消息头
编码过程中主要通过调用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 }
编解码器的获取有两种方式
- 由服务端在初始化Server时,指定使用的codec
-
在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 }
客户端发送数据过程与服务端一样
- 编码
- 压缩
-
添加请求头
不同之处在于客户端编解码器直接由客户端提供即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或者其他进行编解码过程,只需要自定义一个对应的编解码器进行替换
以上就是本文的全部内容,希望本文的内容对大家的学习或者工作能带来一定的帮助,也希望大家多多支持 码农网
猜你喜欢:- Netty-解码器架构与常用解码器
- 解码器 与 编码器
- 基于数据驱动的三维几何解码器
- 用Pytorch构建一个自动解码器
- 开源编解码器 SOLO 源码解读(一):带宽扩展
- 更深的编码器+更浅的解码器=更快的自回归模型
本站部分资源来源于网络,本站转载出于传递更多信息之目的,版权归原作者或者来源机构所有,如转载稿涉及版权问题,请联系我们。