[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或者其他进行编解码过程，只需要自定义一个对应的编解码器进行替换
网友评论