【Go】五、Grpc 的入门使用

grpc 与 protobuf

grpc 使用的是 protobuf 协议，其是一个通用的 rpc 框架，基本支持主流的所有语言、其底层使用 http/2 进行网络通信，具有较高的效率

protobuf 是一种序列化格式，这种格式具有 序列化以及解码速度快（对比json、xml 速度快 2 - 100 倍），压缩率高等优点，是一个性能炸弹

基础环境配置

我们使用之前，要先安装 protobuf 的相关环境：

在 github 上下载 protoc 并配置好环境变量（环境变量配置到 bin 这一级）（用来生成源码）

记得在编译器中安装 protobuf support 插件

protobuf尝试

对 protobuf 进行尝试：创建一个.proto 文件：

syntax = "proto3";message HelloRequest {string name = 1;  // 注意这里不是赋值，而是指定编号
}

之后使用 protoc 生成源码

protoc -I . --go_out=. --go-grpc_out=require_unimplemented_servers=false:. ./helloworld.proto

测试proto的使用：

func main() {req := HelloWorld.HelloRequest{Name:    "Chen",Age:     16,Courses: []string{"C", "Go"},}rsp, _ := proto.Marshal(&req)       // 编码newReq := HelloWorld.HelloRequest{} // 创建一个空的proto_ = proto.Unmarshal(rsp, &newReq)   // 解码，解码存储在 newReq 中fmt.Println(newReq.Name, newReq.Age, newReq.Courses)
}

实际开发尝试

创建目录结构：

grpc_test

server

server.go

client

proto

helloworld.proto

helloworld.pb.go（自动生成）

helloworld_grpc.pb.go（自动生成）

首先自己编写 helloworld.proto：

syntax = "proto3";
// 生成的包名
option go_package = ".;proto";// 标注生成的方法接口
service Greeter {rpc SayHello (HelloRequest) returns (HelloReply);
}// 生成的结构体
message HelloRequest {string name = 1;
}message HelloReply {string message = 1;
}

之后使用命令：

protoc -I . --go_out=. --go-grpc_out=require_unimplemented_servers=false:. ./helloworld.proto

生成helloworld.pb.go 文件：

// Code generated by protoc-gen-go. DO NOT EDIT.
// versions:
// 	protoc-gen-go v1.32.0
// 	protoc        v4.25.3
// source: helloworld.protopackage protoimport (protoreflect "google.golang.org/protobuf/reflect/protoreflect"protoimpl "google.golang.org/protobuf/runtime/protoimpl"reflect "reflect"sync "sync"
)const (// Verify that this generated code is sufficiently up-to-date._ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)// Verify that runtime/protoimpl is sufficiently up-to-date._ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
)type HelloRequest struct {state         protoimpl.MessageStatesizeCache     protoimpl.SizeCacheunknownFields protoimpl.UnknownFieldsName string `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"`
}func (x *HelloRequest) Reset() {*x = HelloRequest{}if protoimpl.UnsafeEnabled {mi := &file_helloworld_proto_msgTypes[0]ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))ms.StoreMessageInfo(mi)}
}func (x *HelloRequest) String() string {return protoimpl.X.MessageStringOf(x)
}func (*HelloRequest) ProtoMessage() {}func (x *HelloRequest) ProtoReflect() protoreflect.Message {mi := &file_helloworld_proto_msgTypes[0]if protoimpl.UnsafeEnabled && x != nil {ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))if ms.LoadMessageInfo() == nil {ms.StoreMessageInfo(mi)}return ms}return mi.MessageOf(x)
}// Deprecated: Use HelloRequest.ProtoReflect.Descriptor instead.
func (*HelloRequest) Descriptor() ([]byte, []int) {return file_helloworld_proto_rawDescGZIP(), []int{0}
}func (x *HelloRequest) GetName() string {if x != nil {return x.Name}return ""
}type HelloReply struct {state         protoimpl.MessageStatesizeCache     protoimpl.SizeCacheunknownFields protoimpl.UnknownFieldsMessage string `protobuf:"bytes,1,opt,name=message,proto3" json:"message,omitempty"`
}func (x *HelloReply) Reset() {*x = HelloReply{}if protoimpl.UnsafeEnabled {mi := &file_helloworld_proto_msgTypes[1]ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))ms.StoreMessageInfo(mi)}
}func (x *HelloReply) String() string {return protoimpl.X.MessageStringOf(x)
}func (*HelloReply) ProtoMessage() {}func (x *HelloReply) ProtoReflect() protoreflect.Message {mi := &file_helloworld_proto_msgTypes[1]if protoimpl.UnsafeEnabled && x != nil {ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))if ms.LoadMessageInfo() == nil {ms.StoreMessageInfo(mi)}return ms}return mi.MessageOf(x)
}// Deprecated: Use HelloReply.ProtoReflect.Descriptor instead.
func (*HelloReply) Descriptor() ([]byte, []int) {return file_helloworld_proto_rawDescGZIP(), []int{1}
}func (x *HelloReply) GetMessage() string {if x != nil {return x.Message}return ""
}var File_helloworld_proto protoreflect.FileDescriptorvar file_helloworld_proto_rawDesc = []byte{0x0a, 0x10, 0x68, 0x65, 0x6c, 0x6c, 0x6f, 0x77, 0x6f, 0x72, 0x6c, 0x64, 0x2e, 0x70, 0x72, 0x6f,0x74, 0x6f, 0x22, 0x22, 0x0a, 0x0c, 0x48, 0x65, 0x6c, 0x6c, 0x6f, 0x52, 0x65, 0x71, 0x75, 0x65,0x73, 0x74, 0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09,0x52, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x22, 0x26, 0x0a, 0x0a, 0x48, 0x65, 0x6c, 0x6c, 0x6f, 0x52,0x65, 0x70, 0x6c, 0x79, 0x12, 0x18, 0x0a, 0x07, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x18,0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x32, 0x31,0x0a, 0x07, 0x47, 0x72, 0x65, 0x65, 0x74, 0x65, 0x72, 0x12, 0x26, 0x0a, 0x08, 0x53, 0x61, 0x79,0x48, 0x65, 0x6c, 0x6c, 0x6f, 0x12, 0x0d, 0x2e, 0x48, 0x65, 0x6c, 0x6c, 0x6f, 0x52, 0x65, 0x71,0x75, 0x65, 0x73, 0x74, 0x1a, 0x0b, 0x2e, 0x48, 0x65, 0x6c, 0x6c, 0x6f, 0x52, 0x65, 0x70, 0x6c,0x79, 0x42, 0x09, 0x5a, 0x07, 0x2e, 0x3b, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x06, 0x70, 0x72,0x6f, 0x74, 0x6f, 0x33,
}var (file_helloworld_proto_rawDescOnce sync.Oncefile_helloworld_proto_rawDescData = file_helloworld_proto_rawDesc
)func file_helloworld_proto_rawDescGZIP() []byte {file_helloworld_proto_rawDescOnce.Do(func() {file_helloworld_proto_rawDescData = protoimpl.X.CompressGZIP(file_helloworld_proto_rawDescData)})return file_helloworld_proto_rawDescData
}var file_helloworld_proto_msgTypes = make([]protoimpl.MessageInfo, 2)
var file_helloworld_proto_goTypes = []interface{}{(*HelloRequest)(nil), // 0: HelloRequest(*HelloReply)(nil),   // 1: HelloReply
}
var file_helloworld_proto_depIdxs = []int32{0, // 0: Greeter.SayHello:input_type -> HelloRequest1, // 1: Greeter.SayHello:output_type -> HelloReply1, // [1:2] is the sub-list for method output_type0, // [0:1] is the sub-list for method input_type0, // [0:0] is the sub-list for extension type_name0, // [0:0] is the sub-list for extension extendee0, // [0:0] is the sub-list for field type_name
}func init() { file_helloworld_proto_init() }
func file_helloworld_proto_init() {if File_helloworld_proto != nil {return}if !protoimpl.UnsafeEnabled {file_helloworld_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {switch v := v.(*HelloRequest); i {case 0:return &v.statecase 1:return &v.sizeCachecase 2:return &v.unknownFieldsdefault:return nil}}file_helloworld_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {switch v := v.(*HelloReply); i {case 0:return &v.statecase 1:return &v.sizeCachecase 2:return &v.unknownFieldsdefault:return nil}}}type x struct{}out := protoimpl.TypeBuilder{File: protoimpl.DescBuilder{GoPackagePath: reflect.TypeOf(x{}).PkgPath(),RawDescriptor: file_helloworld_proto_rawDesc,NumEnums:      0,NumMessages:   2,NumExtensions: 0,NumServices:   1,},GoTypes:           file_helloworld_proto_goTypes,DependencyIndexes: file_helloworld_proto_depIdxs,MessageInfos:      file_helloworld_proto_msgTypes,}.Build()File_helloworld_proto = out.Filefile_helloworld_proto_rawDesc = nilfile_helloworld_proto_goTypes = nilfile_helloworld_proto_depIdxs = nil
}

以及：

helloworld_grpc.pb.go

// Code generated by protoc-gen-go-grpc. DO NOT EDIT.
// versions:
// - protoc-gen-go-grpc v1.3.0
// - protoc             v4.25.3
// source: helloworld.protopackage protoimport (context "context"grpc "google.golang.org/grpc"codes "google.golang.org/grpc/codes"status "google.golang.org/grpc/status"
)// This is a compile-time assertion to ensure that this generated file
// is compatible with the grpc package it is being compiled against.
// Requires gRPC-Go v1.32.0 or later.
const _ = grpc.SupportPackageIsVersion7const (Greeter_SayHello_FullMethodName = "/Greeter/SayHello"
)// GreeterClient is the client API for Greeter service.
//
// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.
type GreeterClient interface {SayHello(ctx context.Context, in *HelloRequest, opts ...grpc.CallOption) (*HelloReply, error)
}type greeterClient struct {cc grpc.ClientConnInterface
}func NewGreeterClient(cc grpc.ClientConnInterface) GreeterClient {return &greeterClient{cc}
}func (c *greeterClient) SayHello(ctx context.Context, in *HelloRequest, opts ...grpc.CallOption) (*HelloReply, error) {out := new(HelloReply)err := c.cc.Invoke(ctx, Greeter_SayHello_FullMethodName, in, out, opts...)if err != nil {return nil, err}return out, nil
}// GreeterServer is the server API for Greeter service.
// All implementations should embed UnimplementedGreeterServer
// for forward compatibility
type GreeterServer interface {SayHello(context.Context, *HelloRequest) (*HelloReply, error)
}// UnimplementedGreeterServer should be embedded to have forward compatible implementations.
type UnimplementedGreeterServer struct {
}func (UnimplementedGreeterServer) SayHello(context.Context, *HelloRequest) (*HelloReply, error) {return nil, status.Errorf(codes.Unimplemented, "method SayHello not implemented")
}// UnsafeGreeterServer may be embedded to opt out of forward compatibility for this service.
// Use of this interface is not recommended, as added methods to GreeterServer will
// result in compilation errors.
type UnsafeGreeterServer interface {mustEmbedUnimplementedGreeterServer()
}func RegisterGreeterServer(s grpc.ServiceRegistrar, srv GreeterServer) {s.RegisterService(&Greeter_ServiceDesc, srv)
}func _Greeter_SayHello_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {in := new(HelloRequest)if err := dec(in); err != nil {return nil, err}if interceptor == nil {return srv.(GreeterServer).SayHello(ctx, in)}info := &grpc.UnaryServerInfo{Server:     srv,FullMethod: Greeter_SayHello_FullMethodName,}handler := func(ctx context.Context, req interface{}) (interface{}, error) {return srv.(GreeterServer).SayHello(ctx, req.(*HelloRequest))}return interceptor(ctx, in, info, handler)
}// Greeter_ServiceDesc is the grpc.ServiceDesc for Greeter service.
// It's only intended for direct use with grpc.RegisterService,
// and not to be introspected or modified (even as a copy)
var Greeter_ServiceDesc = grpc.ServiceDesc{ServiceName: "Greeter",HandlerType: (*GreeterServer)(nil),Methods: []grpc.MethodDesc{{MethodName: "SayHello",Handler:    _Greeter_SayHello_Handler,},},Streams:  []grpc.StreamDesc{},Metadata: "helloworld.proto",
}

尝试编写业务逻辑（这里应该写在handler中，由于业务过于简单，先写在server中）：

server.go：

注意这里 grpc 帮我们处理了服务器不能连续被访问的问题，不需要我们手动通过一个死循环进行处理

package mainimport ("context""net""google.golang.org/grpc""FirstGo/goon/grpc_test/proto"
)type Server struct{}// 业务逻辑
// 第一个参数必须是context，error必须加
func (s *Server) SayHello(ctx context.Context, request *proto.HelloRequest) (*proto.HelloReply, error) {return &proto.HelloReply{Message: "hello " + request.Name,}, nil
}func main() {g := grpc.NewServer()proto.RegisterGreeterServer(g, &Server{})lis, err := net.Listen("tcp", "0.0.0.0:8080")if err != nil {panic("failed to listen: " + err.Error())}err = g.Serve(lis)if err != nil {panic("failed to start grpc: " + err.Error())}
}

编写客户端

client.go：

package mainimport ("FirstGo/goon/grpc_test/proto""context""fmt""google.golang.org/grpc"
)func main() {// 尝试拨号conn, err := grpc.Dial("127.0.0.1:8080", grpc.WithInsecure())if err != nil {panic(err)}defer conn.Close()// 创建客户端c := proto.NewGreeterClient(conn)// 调用对应的方法r, err := c.SayHello(context.Background(), &proto.HelloRequest{Name: "Chen"})if err != nil {panic(err)}fmt.Println(r.Message)
}

GRPC的四种数据传输模式

RPC 还具有四种数据模式，其分别是：

1. 简单模式（上述模式）

   客户端发起一次请求，服务器返回一次响应

2. 服务端数据流模式

   客户发起一次请求，服务器返回一段连续的数据流，最典型的例子是：客户发送一段股票代码，服务端实时将股票的数据源源不断的返回给客户端

3. 客户端数据流模式

   与服务端数据流模式相反，这种是由客户端源源不断的向服务端发送数据流，在发送结束后，由服务端发送一个响应，这种的典型例子是：物联网终端向服务器报送数据

4. 双向数据流模式

   这种是客户端和服务端都可以向双方发送数据流，这个时候双方的数据都可以相互发送，也就是可以实时交互，这种最典型的例子就是聊天机器人

实际测试：
建立文件结构：

stream_grpc_test

server

server.go

client

client.go

proto

stream.pb.gp

stream.proto

stream_grpc.pb.go

stream.proto：

syntax = "proto3";option go_package = ".;proto";service Greeter {rpc GetStream(StreamReqData) returns (stream StreamResData);  //  服务端流模式，返回的响应数据是流rpc PutStream(stream StreamReqData) returns (StreamResData);  // 客户端流模式，传给服务器的数据是流rpc AllStream(stream StreamReqData) returns (stream StreamResData);   // 双向流模式
}message StreamReqData {string data = 1;
}message StreamResData {string data = 1;
}

服务端流模式简单使用

只写了 服务端流模式的 server,go：

const PORT = ":50052"type server struct {
}// 对于服务端数据传输模式，参考以下内容：
// 没有 context 参数，而是将返回作为入参传入
func (s *server) GetStream(req *proto.StreamReqData, res proto.Greeter_GetStreamServer) error {i := 0for {i++_ = res.Send(&proto.StreamResData{Data: fmt.Sprintf("%v", time.Now().Unix()),})time.Sleep(time.Second)if i > 10 {break}}return nil
}// 客户端数据流模式
// 只有一个用来不断接收的入参
func (s *server) PutStream(cliStr proto.Greeter_PutStreamServer) error {return nil
}// 双向数据流模式，和客户端模式相同，只有一个入参
func (s *server) AllStream(allStr proto.Greeter_AllStreamServer) error {return nil
}func main() {lis, err := net.Listen("tcp", PORT)if err != nil {panic(err)}s := grpc.NewServer()proto.RegisterGreeterServer(s, &server{})s.Serve(lis)}

只针对于服务端流的client.go

func main() {// 拨号conn, err := grpc.Dial("localhost:50052", grpc.WithInsecure())if err != nil {panic(err)}defer conn.Close()c := proto.NewGreeterClient(conn)// 这里要注意：我们在进行调用的时候使用的仍让是 simple 的模式，而不是服务端的函数模式res, _ := c.GetStream(context.Background(), &proto.StreamReqData{Data: "Chen"})// 使用一个死循环来接收客户端传进来的数据for {a, err := res.Recv()if err != nil {fmt.Println(err)break}fmt.Println(a)}
}

客户端流模式的简单使用

server.go：

package mainimport ("FirstGo/goon/stream_grpc_test/proto""fmt""google.golang.org/grpc""net""time"
)const PORT = ":50052"type server struct {
}// 对于服务端数据传输模式，参考以下内容：
// 没有 context 参数，而是将返回作为入参传入
func (s *server) GetStream(req *proto.StreamReqData, res proto.Greeter_GetStreamServer) error {i := 0for {i++_ = res.Send(&proto.StreamResData{Data: fmt.Sprintf("%v", time.Now().Unix()),})time.Sleep(time.Second)if i > 10 {break}}return nil
}// 客户端数据流模式
// 只有一个用来不断接收的入参
func (s *server) PutStream(cliStr proto.Greeter_PutStreamServer) error {// 客户端流模式，客户端不断的发送数据给服务器for {if a, err := cliStr.Recv(); err != nil {fmt.Println(err)} else {fmt.Println(a.Data)}}return nil
}// 双向数据流模式，和客户端模式相同，只有一个入参
func (s *server) AllStream(allStr proto.Greeter_AllStreamServer) error {return nil
}func main() {lis, err := net.Listen("tcp", PORT)if err != nil {panic(err)}s := grpc.NewServer()proto.RegisterGreeterServer(s, &server{})s.Serve(lis)}

client.go：

package mainimport ("FirstGo/goon/stream_grpc_test/proto""context""fmt""google.golang.org/grpc""time"
)func main() {// 拨号conn, err := grpc.Dial("localhost:50052", grpc.WithInsecure())if err != nil {panic(err)}defer conn.Close()c := proto.NewGreeterClient(conn)putS, _ := c.PutStream(context.Background())i := 0for {i++putS.Send(&proto.StreamReqData{Data: fmt.Sprintf("Chen %d", i)})time.Sleep(time.Second)if i > 10 {break}}
}

双向流模式的简单使用

server.go：

package mainimport ("FirstGo/goon/stream_grpc_test/proto""fmt""google.golang.org/grpc""net""sync""time"
)const PORT = ":50052"type server struct {
}// 对于服务端数据传输模式，参考以下内容：
// 没有 context 参数，而是将返回作为入参传入
func (s *server) GetStream(req *proto.StreamReqData, res proto.Greeter_GetStreamServer) error {i := 0for {i++_ = res.Send(&proto.StreamResData{Data: fmt.Sprintf("%v", time.Now().Unix()),})time.Sleep(time.Second)if i > 10 {break}}return nil
}// 客户端数据流模式
// 只有一个用来不断接收的入参
func (s *server) PutStream(cliStr proto.Greeter_PutStreamServer) error {// 客户端流模式，客户端不断的发送数据给服务器for {if a, err := cliStr.Recv(); err != nil {fmt.Println(err)} else {fmt.Println(a.Data)}}return nil
}// 双向数据流模式，和客户端模式相同，只有一个入参
func (s *server) AllStream(allStr proto.Greeter_AllStreamServer) error {// 不可以像下面这样简单使用，因为 Recv() 会阻塞主线程//allStr.Recv()//allStr.Send()wg := sync.WaitGroup{}wg.Add(2) // 添加两个等待线程go func() {defer wg.Done()for {data, _ := allStr.Recv()fmt.Println("收到客户端消息：" + data.Data)}}()go func() {defer wg.Done()for {_ = allStr.Send(&proto.StreamResData{Data: "我是服务器"})time.Sleep(time.Second)}}()wg.Wait()return nil
}func main() {lis, err := net.Listen("tcp", PORT)if err != nil {panic(err)}s := grpc.NewServer()proto.RegisterGreeterServer(s, &server{})s.Serve(lis)}

client.go:

package mainimport ("FirstGo/goon/stream_grpc_test/proto""context""fmt""google.golang.org/grpc""sync""time"
)func main() {// 拨号conn, err := grpc.Dial("localhost:50052", grpc.WithInsecure())if err != nil {panic(err)}defer conn.Close()c := proto.NewGreeterClient(conn)// // 这里要注意：我们在进行调用的时候使用的仍让是 simple 的模式，而不是服务端的函数模式// 服务端模式的客户端//res, _ := c.GetStream(context.Background(), &proto.StreamReqData{Data: "Chen"})// // 使用一个死循环来接收客户端传进来的数据//for {//	a, err := res.Recv()//	if err != nil {//		fmt.Println(err)//		break//	}//	fmt.Println(a)//}// 客户端数据流模式//putS, _ := c.PutStream(context.Background())//i := 0//for {//	i++//	putS.Send(&proto.StreamReqData{Data: fmt.Sprintf("Chen %d", i)})//	time.Sleep(time.Second)//	if i > 10 {//		break//	}//}// 双向流模式allStr, _ := c.AllStream(context.Background())wg := sync.WaitGroup{}wg.Add(2) // 添加两个等待线程go func() {defer wg.Done()for {data, _ := allStr.Recv()fmt.Println("收到客户端消息：" + data.Data)}}()go func() {defer wg.Done()for {_ = allStr.Send(&proto.StreamReqData{Data: "我是Chen (客户端)"})time.Sleep(time.Second)}}()wg.Wait()
}