gRPC

gRPC入门

gRPC简介

gRPC来自Google，它是一个开源的框架；它同时也是Cloud Native Computation基金会（CNCF）的一部分，就像Docker和Kubernetes一样。
gRPC允许你为RPC（Remote Procedure call）定义请求和响应，然后gRPC会帮你处理一切剩余问题。
它速度快，执行效率高，基于HTTP/2构建，低延迟，支持流，与开发语言无关，并且可以很简单的插入身份认证、负载均衡、日志和监控等功能。
gRPC它是对RPC一种非常简洁的实现并且解决了很多RPC的问题。

image-20240807163126198

gRPC结构

image-20240810160423389

设计步骤

image-20240810160752936

生命周期

image-20240810160820157

如何学习gRPC

首先，你得学习Protocol Buffers（https://developers.google.com/protocol-buffers/），简单的说，它可以用来定义消息和服务。
然后，你只需要实现服务即可，剩余的gRPC代码将会自动为你生成。
.proto这个文件可以适用于十几种开发语言（包括服务端和客户端），并且它允许你使用同一个框架来支持每秒百万级以上的RPC调用。

开发模式

gPRC使用的是合约优先的API开发模式，它默认使用Protocol buffers（protobuf）作为接口设计语言（IDL），这个.proto文件包括两部分：

  • gRPC服务的定义
  • 服务端和客户端之间传递的消息

为什么使用Protocol Buffers

  1. 它和开发语言无关

  2. 可以生成所有主流开发语言的代码

  3. 数据是二进制格式的，串行化的效率高，Payload比较小

  4. 也很适合传递大量的数据

  5. 通过设定某些规则，使得API的进化也很简单

Hello gRPC

新建一个以 .proto 结尾的文件first.proto：

1
2
3
4
5
6
7
8
syntax = "proto3"; //使用proto3语法 默认是proto2

// 定义消息
message FirstMessage{
    int32 id = 1;
    string name = 2;
    bool is_male = 3;
}

消息定义

标量类型

  1. 数值型：数值型有很多种形式：double，float，int32，int64，uint32，uint64，sint32，sint64，fixed32，fixed64，sfixed32，sfixed64。
    根据需要选择对应的数值类型。
  2. 布尔型：bool型可以有True和False两个值。
  3. 字符串：string表示任意长度的文本，但是它必须包含的是UTF-8编码或7位ASCII的文本，长度不可超过232。
  4. 字节型：bytes可表示任意的byte数组序列，但是长度也不可以超过232，最后是由你来决定如何解释这些bytes。例如你可以使用这个类型来表示一个图片。
1
2
3
4
5
6
7
8
9
10
11
syntax = "proto3";

message Person {
    int32 id = 1;
    string name = 2;
    float height = 3;
    float weight = 4;
    bytes avatar = 5;
    string email = 6;
    bool email_verified = 7;
}

字段的数值(Tag)

在Protocol Buffers里面，字段的名其实没那么重要，但是写C#/go等语言代码的时候，字段名还是很重要的。
对于protobuf来说，这个tag是更为重要的。
可以使用的最小的tag数值是1，最大值是2^29-1，或者536,870,911。但是你不可以使用19000到19999之间的数，这部分数是保留的。

从1到15的Tag数只占用1个字节的空间，所以它们应该被用在频繁使用的字段上。而从16到2047，则占用两个字节，它们可以用在不频繁使用的字段上。

字段规则

protobuf的字段必须满足以下两个规则之一

  1. 单数字段(Singular）
    大概意思就是指这个字段只能出现0或1次（不能超过一次），这也是proto3的默认字段规则。
  2. 重复字段(Repeated）
    与singular相对的就是repeated。如果你想做一个list或数组的话，你可以使用重复字段这个概念。这个list可以有任何数量（包括0）的元素。它里面的值的顺序将会得到保留。
1
2
3
4
5
6
syntax = "proto3";

message Person {
    ...
    repeated string phone_numbers = 8; // 编码:packed
}

保留的字段

如果你对你定义的消息类型进行了更新，例如删除某个字段或者注释掉某个字段，那么其它开发者在以后更新这个消息类型的时候可能会重新使用被你删除/注释掉的字段的数值（tag）。如果以后还需要使用这个消息类型的老版本的proto文件，那么这将会引起严重的问题，例如数据损坏、隐私漏洞等等。
一种避免此类事情发生的解决办法就是将你删除/注释掉的这些字段的数值（或/并且包括字段名，因为字段名也可引起JSON序列化的问题）标记为reserved，如果其他人再使用这个数值作为字段标识符，那么编译器就会有错误提示。

1
2
3
4
5
6
7
syntax = "proto3";

message Person {
    ...
    reserved 9, 10, 20 to 100, 200 to max;
    reserved "foo", "bar";
}

字段的默认值

当消息被解析的时候，如果编码的消息里不含有特定的一个singular元素，那么在被解析对象里相应的字段就会被设为默认值。
常用类型的默认值如下：

  • string：空字符串
  • bytes：空的byte数组
  • bool：false
  • 数值型：0
  • 枚举enum：枚举里定义的第一个枚举值，值必须是0
  • repeated：通常是相应开发语言里的空list
  • 还有个消息类型的字段，它的默认值和开发语言有关，这个以后再说。

默认值在更新Protocol Buffer消息定义的时候有很重要的作用，它可以防止对现有代码/新代码造成破坏性影响。它们也可以保证字段永远不会有null值。
但是，默认值还是非常危险的：
你无法区分这个默认值到底是来白一个去失的字段还是字段的实际值正好等于默认值。

应该怎么办？
需要保证这个默认值对于业务来说是一个毫无意义的值。例如int32pop(入口)默认值就可以设置为-1.
再就是，可能需要在你的代码里来做一些对默认值的判断，从而进行处理。

枚举

枚举里面定义的第一个值就是这个枚举的默认值。
Enum的tag必须从0开始，所以0就是枚举的数值默认值。

1
2
3
4
5
6
7
8
9
10
11
12
syntax = "proto3";

message Person {
    ...

    Gender gender = 11;
    enum Gender {
        UNKNOWN = 0;
        MALE = 1;
        FEMALE = 2;
    }
}

为枚举值起别名

枚举值是可以起别名的，起别名的作用就是允许两个枚举值拥有同一个数值。
要想起别名，首先需要设置allow_alias这个option为true
然后我们为FEMALE这个枚举值起了一个别名叫做WOMAN，它们的数值是一样的。同样的MAN是MALE的数值也是一样的。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
syntax = "proto3";

message Person {
    ...

    Gender gender = 11;
    enum Gender {
        option allow_alias = true;
        UNKNOWN = 0;
        MALE = 1;
        FEMALE = 2;

        MAN = 1;
        WOMAN = 2;
    }
}

枚举里面的常量的值必须不能超过32位整型的数值，不建议使用负数。
枚举可以定义在message里面，也可以在外边单独定义以便复用。如果另一个消息想使用Person里面这个Gender枚举，那么可以使用Person.Gender这这种形式。

使用其它的信息类型

可以使用其它的信息类型作为字段的类型。

新建date.proto

1
2
3
4
5
6
7
syntax = "proto3";

message Date{
    int32 year = 1;
    int32 month = 2;
    int32 day = 3;
}

在person.proto中使用

1
2
3
4
5
6
7
8
syntax = "proto3";

import "date.proto"; // 导入date.proto

message Person {
    ...
    Date birthday = 12;
}

消息嵌套

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
syntax = "proto3";

import "date.proto"; // 导入date.proto

message Person {
	...
    repeated Address address = 13;

    // 嵌套定义
    message Address{
        string province = 1;
        string city = 2;
        string zip_code = 3;
        string street = 4;
        string number = 5;
    }
}

打包
你可以向proto文件添加可选的打包（package）说明符，以避免消息类型间的名称冲突，打包是很有必要的。

1
2
3
4
5
6
7
syntax = "proto3";

import "date.proto"; // 导入date.proto

package my.project; // 定义包名,即命名空间 C# namespace:My.Project
option csharp_namespace = "My.WebApis"; // 指定C#命名空间
...

设置ProtocolBuffers编译器

protoc编译器主要就是用来生成代码的，它的下载地址目前是：https://github.com/protocolbuffers/protobuf/releases/

image-20240807173101677

解压完后添加bin文件夹到环境变量中：

1
2
3
4
5
vim ~/.bashrc
#==================添加===================
export PATH=/home/liaojie1314/env/protoc/bin:$PATH # 根据你的bin文件夹修改/home/liaojie1314/env/protoc/bin
#=========================================
source ~/.bashrc

查看是否成功：

1
protoc

image-20240807173447158

出现内容说明安装成功。

image-20240807173658255

生成源文件

csharp：

1
protoc first.proto --csharp_out=csharp

需要提前创建csharp文件夹

发现在csharp文件下生成了一个First.cs文件。

这个文件我们不要修改，可以把它看成是一个库，我们只需要调用就行

一次性生成：

1
protoc *.proto --csharp_out=csharp

go语言例子

创建项目

image-20240807175537991

安装依赖

1
go get -u github.com/golang/protobuf/protoc-gen-go

编写proto文件

位置 src->proto->person.proto:

1
2
3
4
5
6
7
8
9
10
syntax = "proto3";

option go_package = "src/proto";

message PersonMessage{
  int32 id = 1;
  bool is_adult = 2;
  string name = 3;
  repeated int32 lucky_numbers = 4;
}

生成go代码

在项目根目录下运行:

1
protoc --proto_path src/ --go_out=./ src/proto/person.proto

发现生成了 person.pb.go 文件

image-20240809191945931

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
// Code generated by protoc-gen-go. DO NOT EDIT.
// versions:
//  protoc-gen-go v1.34.2
//  protoc        v3.20.3
// source: proto/person.proto

package proto

import (
    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 PersonMessage struct { // message 对应于 go 中 struct
    state         protoimpl.MessageState
    sizeCache     protoimpl.SizeCache
    unknownFields protoimpl.UnknownFields

    Id           int32   `protobuf:"varint,1,opt,name=id,proto3" json:"id,omitempty"` //反射
    IsAdult      bool    `protobuf:"varint,2,opt,name=is_adult,json=isAdult,proto3" json:"is_adult,omitempty"`
    Name         string  `protobuf:"bytes,3,opt,name=name,proto3" json:"name,omitempty"`
    LuckyNumbers []int32 `protobuf:"varint,4,rep,packed,name=lucky_numbers,json=luckyNumbers,proto3" json:"lucky_numbers,omitempty"`
}

func (x *PersonMessage) Reset() {
    *x = PersonMessage{}
    if protoimpl.UnsafeEnabled {
       mi := &file_proto_person_proto_msgTypes[0]
       ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
       ms.StoreMessageInfo(mi)
    }
}

func (x *PersonMessage) String() string {
    return protoimpl.X.MessageStringOf(x)
}

func (*PersonMessage) ProtoMessage() {}

func (x *PersonMessage) ProtoReflect() protoreflect.Message {
    mi := &file_proto_person_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 PersonMessage.ProtoReflect.Descriptor instead.
func (*PersonMessage) Descriptor() ([]byte, []int) {
    return file_proto_person_proto_rawDescGZIP(), []int{0}
}

func (x *PersonMessage) GetId() int32 {
    if x != nil {
       return x.Id
    }
    return 0
}

func (x *PersonMessage) GetIsAdult() bool {
    if x != nil {
       return x.IsAdult
    }
    return false
}

func (x *PersonMessage) GetName() string {
    if x != nil {
       return x.Name
    }
    return ""
}

func (x *PersonMessage) GetLuckyNumbers() []int32 {
    if x != nil {
       return x.LuckyNumbers
    }
    return nil
}

var File_proto_person_proto protoreflect.FileDescriptor

var file_proto_person_proto_rawDesc = []byte{
    0x0a, 0x12, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2f, 0x70, 0x65, 0x72, 0x73, 0x6f, 0x6e, 0x2e, 0x70,
    0x72, 0x6f, 0x74, 0x6f, 0x22, 0x73, 0x0a, 0x0d, 0x50, 0x65, 0x72, 0x73, 0x6f, 0x6e, 0x4d, 0x65,
    0x73, 0x73, 0x61, 0x67, 0x65, 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28,
    0x05, 0x52, 0x02, 0x69, 0x64, 0x12, 0x19, 0x0a, 0x08, 0x69, 0x73, 0x5f, 0x61, 0x64, 0x75, 0x6c,
    0x74, 0x18, 0x02, 0x20, 0x01, 0x28, 0x08, 0x52, 0x07, 0x69, 0x73, 0x41, 0x64, 0x75, 0x6c, 0x74,
    0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04,
    0x6e, 0x61, 0x6d, 0x65, 0x12, 0x23, 0x0a, 0x0d, 0x6c, 0x75, 0x63, 0x6b, 0x79, 0x5f, 0x6e, 0x75,
    0x6d, 0x62, 0x65, 0x72, 0x73, 0x18, 0x04, 0x20, 0x03, 0x28, 0x05, 0x52, 0x0c, 0x6c, 0x75, 0x63,
    0x6b, 0x79, 0x4e, 0x75, 0x6d, 0x62, 0x65, 0x72, 0x73, 0x42, 0x0b, 0x5a, 0x09, 0x73, 0x72, 0x63,
    0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
}

var (
    file_proto_person_proto_rawDescOnce sync.Once
    file_proto_person_proto_rawDescData = file_proto_person_proto_rawDesc
)

func file_proto_person_proto_rawDescGZIP() []byte {
    file_proto_person_proto_rawDescOnce.Do(func() {
       file_proto_person_proto_rawDescData = protoimpl.X.CompressGZIP(file_proto_person_proto_rawDescData)
    })
    return file_proto_person_proto_rawDescData
}

var file_proto_person_proto_msgTypes = make([]protoimpl.MessageInfo, 1)
var file_proto_person_proto_goTypes = []any{
    (*PersonMessage)(nil), // 0: PersonMessage
}
var file_proto_person_proto_depIdxs = []int32{
    0, // [0:0] is the sub-list for method output_type
    0, // [0:0] is the sub-list for method input_type
    0, // [0:0] is the sub-list for extension type_name
    0, // [0:0] is the sub-list for extension extendee
    0, // [0:0] is the sub-list for field type_name
}

func init() { file_proto_person_proto_init() }
func file_proto_person_proto_init() {
    if File_proto_person_proto != nil {
       return
    }
    if !protoimpl.UnsafeEnabled {
       file_proto_person_proto_msgTypes[0].Exporter = func(v any, i int) any {
          switch v := v.(*PersonMessage); i {
          case 0:
             return &v.state
          case 1:
             return &v.sizeCache
          case 2:
             return &v.unknownFields
          default:
             return nil
          }
       }
    }
    type x struct{}
    out := protoimpl.TypeBuilder{
       File: protoimpl.DescBuilder{
          GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
          RawDescriptor: file_proto_person_proto_rawDesc,
          NumEnums:      0,
          NumMessages:   1,
          NumExtensions: 0,
          NumServices:   0,
       },
       GoTypes:           file_proto_person_proto_goTypes,
       DependencyIndexes: file_proto_person_proto_depIdxs,
       MessageInfos:      file_proto_person_proto_msgTypes,
    }.Build()
    File_proto_person_proto = out.File
    file_proto_person_proto_rawDesc = nil
    file_proto_person_proto_goTypes = nil
    file_proto_person_proto_depIdxs = nil
}

该文件为生成文件，不要修改

编写go代码

main.go

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
package main

import (
	"github.com/golang/protobuf/jsonpb"
	"github.com/golang/protobuf/proto"
	personProto "github.com/liaojie1314/protobuf-go/src/proto"
	"io/ioutil"
	"log"
)

func main() {
	//写入文件
	pm := NewPersonMessage()
	//_ = writeToFile("person.bin", pm)
	//读取文件信息
	blank := &personProto.PersonMessage{}
	//_ = readFromFile("person.bin", blank)
	//fmt.Println(blank)
	//转换为JSON
	pmStr := toJson(pm)
	log.Println(pmStr)
	//JSON转换为PB
	_ = fromJson(pmStr, blank)
	log.Println(blank)
}

func toJson(pb proto.Message) string {
	marshaler := jsonpb.Marshaler{Indent: "    "}

	str, err := marshaler.MarshalToString(pb)
	if err != nil {
		log.Fatalln("转换为JSON失败", err.Error())
	}
	return str
}

func fromJson(in string, pb proto.Message) error {
	err := jsonpb.UnmarshalString(in, pb)
	if err != nil {
		log.Fatalln("JSON转换为PB失败", err.Error())
	}
	return nil
}

// 写入文件
func writeToFile(fileName string, pb proto.Message) error {
	// 序列化
	dataBytes, err := proto.Marshal(pb)
	if err != nil {
		log.Fatalln("序列化失败", err.Error())
	}
	// 写入文件
	if err := ioutil.WriteFile(fileName, dataBytes, 0644); err != nil {
		log.Fatalln("写入文件失败", err.Error())
	}
	log.Println("写入文件成功")
	return nil
}

// 读取文件
func readFromFile(fileName string, pb proto.Message) error {
	// 读取文件
	dataBytes, err := ioutil.ReadFile(fileName)
	if err != nil {
		log.Fatalln("读取文件失败", err.Error())
	}
	// 反序列化
	if err := proto.Unmarshal(dataBytes, pb); err != nil {
		log.Fatalln("反序列化失败", err.Error())
	}
	log.Println("读取文件成功")
	return nil
}

func NewPersonMessage() *personProto.PersonMessage {
	pm := personProto.PersonMessage{
		Id:           1234,
		IsAdult:      true,
		Name:         "YuanYuan",
		LuckyNumbers: []int32{6, 8},
	}

	return &pm
}

image-20240809202120835

枚举使用

enum.proto

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
syntax = "proto3";

option go_package = "src/proto";

message EnumMessage{
  int32 id = 1;
  Gender gender = 2;
}

enum Gender{
  option allow_alias = true;
  NOT_SPECIFIED = 0;
  WOMAN = 1;
  MAN = 2;
  FEMALE = 1;
  MALE = 2;
}

生成go代码

1
protoc --proto_path src/ --go_out=./ src/proto/enum.proto

使用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
package main

import (
    "fmt"
    myProto "github.com/liaojie1314/protobuf-go/src/proto"
)

func main() {
    em := NewEnumMessage()
    fmt.Println(myProto.Gender_name[int32(em.Gender)])
}

func NewEnumMessage() *myProto.EnumMessage {
    em := myProto.EnumMessage{
       Id:     888,
       Gender: myProto.Gender_MALE,
    }
    return &em
}

复杂类型使用

complex.proto

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
syntax = "proto3";

option go_package = "src/proto";

message DepartmentMessage {
  int32 id = 1;
  string name = 2;
  repeated EmployeeMessage employees = 3;
  DepartmentMessage parent_department = 4;
  repeated DepartmentMessage children_department = 5;
}

message EmployeeMessage {
  int32 id = 1;
  string name = 2;
}

生成go代码

1
protoc --proto_path src/ --go_out=./ src/proto/complex.proto

使用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
package main

import (
    "fmt"
    myProto "github.com/liaojie1314/protobuf-go/src/proto"
)

func main() {
    dm := NewDepartMessage()
    fmt.Println(dm)
}


func NewDepartMessage() *myProto.DepartmentMessage {
    dm := myProto.DepartmentMessage{
       Id:   111,
       Name: "开发部",
       Employees: []*myProto.EmployeeMessage{
          {
             Id:   22,
             Name: "YuanYuan",
          },
          {
             Id:   8888,
             Name: "liaojie1314",
          },
       },
       ParentDepartment: &myProto.DepartmentMessage{
          Id:   1122,
          Name: "总公司",
       },
    }

    return &dm
}

image-20240809204118945

更新消息类型

需求会发生变化

有一些字段可能会发生变化，可能会添加一些字段，也可能会删除一些字段。
但是可能有很多程序正在使用/读取你的Protocol Buffer的消息，但是它们没法都随着需求进行更新。
所以，在你对源数据进行演进的时候，一定不要引起破坏性变化，否则其它的程序可能就无法正常工作了。

两种变更情景

向前兼容变更：使用新的.proto文件来写数据----从旧的.proto文件读取数据
向后兼容变更：使用旧的.proto文件来写数据----从新的.proto文件读取数据

更新消息类型的规则

不要修改任何现有字段的数字（tag）
你可以添加新的字段，那些使用旧的消息格式的代码仍然可以将消息序列化，您应该注意这些元素的默认值，以便新代码可以与旧代码生成的消息正确交互。类似的，新代码所创建的消息也可以被旧代码解析：旧的二进制在解析的时候会忽略新的字段。
字段可以被删除，只要它们的数字（tag）在更新后的消息类型中不再使用即可。你也可以把字段名改为使用“OBSOLETE_"前而不是删除字段，或者把这些字段的数字（tag）进行保留（reserved），以免未来其它开发者不消息使用了删除字段的数字。
对于数据类型的变化，例如int32到int64，string到bytes等等，可以参考官方文档：
https://developers.google.com/protocol-buffers/docs/proto3#updating。但是建议还是尽量不要去修改字段的数据类型。

身份认证

这里指的不是用户的身份认证，而是指多个server和client之间，它们如何识别出来谁是谁，并且能安全的进行消息传输。
在身份认证这方面，gRPC一共有4种身份认证的机制：

  • 不采取任何措施的连接，也就是不安全的连接。
  • TLS/SSL连接
  • 基于GoogleToken 的身份认证。
  • 自定义的身份认证提供商。

消息传输类型

gRPC的消息传输类型有4种

  • 第一种是一元的消息，就是简单的请求–响应。

    image-20240810161726531

  • 第二种是server streaming（流），server会把数据streaming回给client。

    image-20240810162000966

  • 第三种是client streaming，也就是client会把数据streaming给server。

    image-20240810162310743

  • 最后是双向streaming。

    image-20240810162203623

go Server

创建proto文件

新建protos->messages.proto

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
syntax = "proto3";

option go_package = "./pb";

import "protos/enums.proto";
import "google/protobuf/timestamp.proto";

message Employee{
  int32 id = 1;
  int32 no = 2;
  string firstname = 3;
  string lastname = 4;

  MonthSalary monthSalary = 6;
  EmployeeStatus status = 7;
  google.protobuf.Timestamp lastModified = 8;

  reserved 5;
  reserved "salary";
}

message MonthSalary {
  float basic = 1;
  float bonus = 2;
}

message GetByNoRequest{
  int32 no = 1;
}

message GetAllRequest {}

message AddPhotoRequest{
  bytes data = 1;
}

message EmployeeRequest{
  Employee employee = 1;
}

message EmployeeResponse{
  Employee employee = 1;
}

message AddPhotoResponse{
  bool isOk = 1;
}

service EmployeeService{
  rpc GetByNo(GetByNoRequest) returns(EmployeeResponse);
  rpc GetAll(GetAllRequest) returns(stream EmployeeResponse);
  rpc AddPhoto(stream AddPhotoRequest) returns(AddPhotoResponse);
  rpc Save(EmployeeRequest) returns(EmployeeResponse);
  rpc SaveAll(stream EmployeeRequest) returns(stream EmployeeResponse);
}

新建protos->enums.proto

1
2
3
4
5
6
7
8
9
10
syntax = "proto3";

option go_package = "./pb";

enum EmployeeStatus {
  NORMAL = 0;
  ON_VACATION = 1;
  RESIGNED = 2;
  RETIRED = 3;
}

添加依赖

1
go get -u google.golang.org/grpc

生成pb代码

1
2
protoc ./protos/*.proto --go_out=./
protoc ./protos/*.proto --go-grpc_out=./

编写go代码

模拟数据库：data.go

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
package main

import (
	"google.golang.org/protobuf/types/known/timestamppb"
	"protobuf-server/pb"
	"time"
)

var employees = []pb.Employee{
	{
		Id:        1,
		No:        2001,
		Firstname: "Liao",
		Lastname:  "Jie",
		MonthSalary: &pb.MonthSalary{
			Basic: 1000,
			Bonus: 500,
		},
		Status: pb.EmployeeStatus_NORMAL,
		LastModified: &timestamppb.Timestamp{
			Seconds: time.Now().Unix(),
		},
	},
	{
		Id:        2,
		No:        2002,
		Firstname: "Yang",
		Lastname:  "JingNi",
		MonthSalary: &pb.MonthSalary{
			Basic: 1000,
			Bonus: 500,
		},
		Status: pb.EmployeeStatus_NORMAL,
		LastModified: &timestamppb.Timestamp{
			Seconds: time.Now().Unix(),
		},
	},
	{
		Id:        3,
		No:        2003,
		Firstname: "Yuan",
		Lastname:  "Yuan",
		MonthSalary: &pb.MonthSalary{
			Basic: 1000,
			Bonus: 500,
		},
		Status: pb.EmployeeStatus_NORMAL,
		LastModified: &timestamppb.Timestamp{
			Seconds: time.Now().Unix(),
		},
	},
}

生成证书：

1
openssl req -newkey rsa:4096 -nodes -sha256 -keyout server_private.key -x509 -days 36500 -out server.pem -addext "subjectAltName =DNS:www.yuanyuan.blog"

没安装openssl的可以使用 sudo apt-get install openssl 安装一下（ubuntu），其他平台自己搜素一下。

编写main.go：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
package main

import (
    "errors"
    "fmt"
    "golang.org/x/net/context"
    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials"
    "google.golang.org/grpc/metadata"
    "grpc-server/pb"
    "io"
    "log"
    "net"
    "time"
)

const port = ":5001"

func main() {
    listen, err := net.Listen("tcp", port)
    if err != nil {
       log.Fatalln(err.Error())
    }
    creds, err := credentials.NewServerTLSFromFile("server.pem", "server_private.key")
    if err != nil {
       log.Fatalln(err.Error())
    }
    options := []grpc.ServerOption{grpc.Creds(creds)}
    server := grpc.NewServer(options...)
    pb.RegisterEmployeeServiceServer(server, new(employeeService))
    log.Println("Starting server on port " + port)
    server.Serve(listen)
}

type employeeService struct {
    pb.UnimplementedEmployeeServiceServer
}

func (s *employeeService) GetAll(req *pb.GetAllRequest, stream pb.EmployeeService_GetAllServer) error {
    for _, e := range employees {
       stream.Send(&pb.EmployeeResponse{
          Employee: &e,
       })
       time.Sleep(2 * time.Second)
    }
    return nil
}

func (s *employeeService) AddPhoto(stream pb.EmployeeService_AddPhotoServer) error {
    md, ok := metadata.FromIncomingContext(stream.Context())
    if ok {
       fmt.Printf("Employee: %s\n", md["no"][0])
    }
    var img []byte
    for {
       data, err := stream.Recv()
       if err == io.EOF {
          fmt.Printf("File size: %d\n", len(img))
          return stream.SendAndClose(&pb.AddPhotoResponse{IsOk: true})
       }
       if err != nil {
          return err
       }
       fmt.Printf("File received: %d\n", len(data.Data))
       img = append(img, data.Data...)
    }
}

func (s *employeeService) Save(ctx context.Context, request *pb.EmployeeRequest) (*pb.EmployeeResponse, error) {
    //TODO implement me
    panic("implement me")
}

func (s *employeeService) SaveAll(stream pb.EmployeeService_SaveAllServer) error {
    for {
       emoReq, err := stream.Recv()
       if err == io.EOF {
          break
       }
       if err != nil {
          return err
       }
       employees = append(employees, *emoReq.Employee)
       stream.Send(&pb.EmployeeResponse{
          Employee: emoReq.Employee,
       })
    }
    for _, emp := range employees {
       fmt.Println(emp)
    }
    return nil
}

func (s *employeeService) GetByNo(ctx context.Context,
    req *pb.GetByNoRequest) (*pb.EmployeeResponse, error) {
    for _, e := range employees {
       if req.No == e.No {
          return &pb.EmployeeResponse{
             Employee: &e,
          }, nil
       }
    }
    return nil, errors.New("employee not found")
}

go client

创建项目，复制protos文件夹，证书文件，安装依赖，生成pb代码

编写main.go：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
package main

import (
    "fmt"
    "golang.org/x/net/context"
    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials"
    "google.golang.org/grpc/metadata"
    "google.golang.org/protobuf/types/known/timestamppb"
    "grpc-client/pb"
    "io"
    "log"
    "os"
    "time"
)

const port = ":5001"

func main() {
    creds, err := credentials.NewClientTLSFromFile("server.pem", "www.yuanyuan.blog")
    if err != nil {
       log.Fatalln(err.Error())
    }

    options := []grpc.DialOption{grpc.WithTransportCredentials(creds)}

    conn, err := grpc.Dial("localhost"+port, options...)
    if err != nil {
       log.Fatalln(err.Error())
    }
    defer conn.Close()
    client := pb.NewEmployeeServiceClient(conn)
    //getByNo(client)
    //getAll(client)
    //addPhoto(client)
    saveAll(client)
}

func saveAll(client pb.EmployeeServiceClient) {
    employees := []pb.Employee{
       pb.Employee{
          Id:        100,
          No:        1000,
          Firstname: "John",
          Lastname:  "Doe",
          MonthSalary: &pb.MonthSalary{
             Basic: 10000,
             Bonus: 2000,
          },
          Status: pb.EmployeeStatus_NORMAL,
          LastModified: &timestamppb.Timestamp{
             Seconds: time.Now().Unix(),
          },
       },
       pb.Employee{
          Id:        200,
          No:        2000,
          Firstname: "Jane",
          Lastname:  "Doe",
          MonthSalary: &pb.MonthSalary{
             Basic: 20000,
             Bonus: 4000,
          },
          Status: pb.EmployeeStatus_ON_VACATION,
          LastModified: &timestamppb.Timestamp{
             Seconds: time.Now().Unix(),
          },
       },
       pb.Employee{
          Id:        300,
          No:        3000,
          Firstname: "Jack",
          Lastname:  "Doe",
          MonthSalary: &pb.MonthSalary{
             Basic: 30000,
             Bonus: 6000,
          },
          Status: pb.EmployeeStatus_ON_VACATION,
          LastModified: &timestamppb.Timestamp{
             Seconds: time.Now().Unix(),
          },
       },
    }

    stream, err := client.SaveAll(context.Background())
    if err != nil {
       log.Fatalln(err.Error())
    }
    finishChannel := make(chan struct{})
    go func() {
       for {
          res, err := stream.Recv()
          if err == io.EOF {
             finishChannel <- struct{}{}
             break
          }
          if err != nil {
             log.Fatalln(err.Error())
          }
          fmt.Println(res.Employee)
       }
    }()
    for _, e := range employees {
       err := stream.Send(&pb.EmployeeRequest{Employee: &e})
       if err != nil {
          log.Fatalln(err.Error())
       }
    }
    stream.CloseSend()
    <-finishChannel
}

func getByNo(client pb.EmployeeServiceClient) {
    res, err := client.GetByNo(context.Background(), &pb.GetByNoRequest{No: 2001})
    if err != nil {
       log.Fatalln(err.Error())
    }
    fmt.Println(res.Employee)
}

func getAll(client pb.EmployeeServiceClient) {
    stream, err := client.GetAll(context.Background(), &pb.GetAllRequest{})
    if err != nil {
       log.Fatalln(err.Error())
    }
    for {
       res, err := stream.Recv()
       if err == io.EOF {
          break
       }
       if err != nil {
          log.Fatalln(err.Error())
       }
       fmt.Println(res.Employee)
    }
}

func addPhoto(client pb.EmployeeServiceClient) {
    imgFile, err := os.Open("avatar.png")
    if err != nil {
       log.Fatalln(err.Error())
    }

    defer imgFile.Close()
    md := metadata.New(map[string]string{"no": "2001"})
    ctx := context.Background()
    ctx = metadata.NewOutgoingContext(ctx, md)

    stream, err := client.AddPhoto(ctx)
    if err != nil {
       log.Fatalln(err.Error())
    }
    for {
       chunk := make([]byte, 128*1024)
       chunkSize, err := imgFile.Read(chunk)
       if err == io.EOF {
          break
       }
       if err != nil {
          log.Fatalln(err.Error())
       }
       if chunkSize < len(chunk) {
          chunk = chunk[:chunkSize]
       }
       stream.Send(&pb.AddPhotoRequest{Data: chunk})
       time.Sleep(time.Millisecond * 500)
    }
    res, err := stream.CloseAndRecv()
    if err != nil {
       log.Fatalln(err.Error())
    }
    fmt.Println(res.IsOk)
}

测试

server：

image-20240817191615333

client：

image-20240817191536345