gRPC使用实例（二）

gRPC , 评论
首先使用gRPC先定义protobuf协议文件

hello.proto

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

package services;

message HelloRequest {
    string name = 1;
}

message HelloReply {
    string message = 2;
}

service HelloService {
    rpc SayHello(HelloRequest) returns (HelloReply);
    rpc Streaming(stream HelloRequest) returns (stream HelloReply);
}

user.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
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
syntax = "proto3";

package services;

// 测试官方类型
import "google/protobuf/timestamp.proto";
import "google/protobuf/empty.proto";
import "google/protobuf/any.proto";

//import "common.proto";

// 测试枚举类型
enum LoginStatus {
    success     = 0;    // 登录成功
    failed      = 1;    // 登录失败
    error       = 2;    // 错误
}

// 用户登录请求
message UserLoginRequest {
    int64  user_id        = 1;       // 用户编号
    string user_name      = 2;       // 用户名
    string user_nickname  = 3;       // 用户昵称
    string user_country   = 4;       // 用户国籍
    int32  user_gender    = 5;       // 用户性别
    double user_credits   = 6;       // 用户积分
    bool   is_admin       = 7;       // 是否是管理员
    bytes  auth_code      = 8;       // 授权码
    float  user_balance   = 9;       // 用户余额
    // 用户爱好
    enum Hobby {
        music      = 0;  // 音乐
        basketball = 1;  // 篮球
        tour       = 2;  // 旅游
    }
    Hobby user_hobby                  = 10;     // 用户爱好
    
    repeated int32  lucky_numbers     = 11;     // 幸运数字
    map<string, string> user_email    = 12;     // 用户email
    map<string, int32> user_weight    = 13;     // 用户体重
    map<string, PhoneNumber> user_num = 14;     // 测试map
    
    LoginStatus login_status          = 15;     // 登录状态
    
    // 手机类型枚举
    enum PhoneType {
        mobile = 0;   // 移动
        home   = 1;   // 家庭
        work   = 2;   // 办公室
    }
    
    // 手机号码类型
    message PhoneNumber {
        string    phone_number   = 1;    // 手机号码
        PhoneType phone_type     = 2;    // 手机号码类型
    
    }
    
    repeated PhoneNumber phones                = 16;    // 手机号码
    google.protobuf.Timestamp last_update_date = 17;    // 最后更新日期
}

// 用户登录响应
message UserLoginResponse {
    int64  user_id        =  1;         // 用户编号
    string user_name      =  2;         // 用户名
    string user_pwd       =  3;         // 用户密码
    int32  user_gender    =  5;         // 用户性别
    double user_credits   =  6;         // 用户积分
    bool   is_admin       =  7;         // 是否是管理员
    bytes  auth_code      =  8;         // 授权码
    float  user_balance   =  9;         // 用户余额
    // 用户爱好
    enum Hobby {
        music      = 0;  // 音乐
        basketball = 1;  // 篮球
        tour       = 2;  // 旅游
    }
    Hobby user_hobby       = 10;     // 用户爱好
    // 手机类型
    enum PhoneType {
        mobile = 0;   // 移动
        home   = 1;   // 家庭
        work   = 2;   // 办公室
    }
    // 手机号码类型
    message PhoneNumber {
        string    phone_number   = 1;    // 手机号码
        PhoneType phone_type     = 2;    // 手机号码类型
    
    }
    repeated PhoneNumber phones                = 11;    // 手机号码
    google.protobuf.Timestamp last_update_date = 12;    // 最后更新日期
    map<string, PhoneNumber> user_num          = 13;    // 测试map
    repeated int32  lucky_numbers              = 14;    // 幸运数字
    map<string, string> user_email             = 15;    // 用户email
    map<string, int32> user_weight             = 16;    // 用户体重
    LoginStatus login_status                   = 17;    // 登录状态
}



message TestRequest {
    string   name = 1;
    repeated Extension extensions = 2;
}

message TestResponse {
    string msg      = 1;
    string name     = 2;
    repeated Extension extensions = 3;
}

message Extension {
    uint64 id = 1;
    google.protobuf.Any data = 2;   //测试官方any类型
}


// user service
service UserService {
    rpc Login(UserLoginRequest)  returns  (UserLoginResponse);
    rpc LoginStreaming(stream UserLoginRequest)  returns  (stream UserLoginResponse);
}

// test service
service TestService {
    rpc Test(TestRequest)  returns (TestResponse);
    rpc Ping (google.protobuf.Empty) returns (google.protobuf.Empty);
}

common.proto

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

package services;

enum LoginStatus {
    success = 0;    // 登录成功
    failed  = 1;    // 登录失败
    error   = 3;    // 错误
}
根据protobuf文件生成pb文件

generate.sh

1
2
#!/bin/bash
protoc -I. --go_out=plugins=grpc:../services/pb *.proto
实现接口协议

services/hello.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
package services

import (
    "context"
    pb "tests/grpc/services/pb"
    "io"
)

type HelloService struct{}

func (h *HelloService) SayHello(ctx context.Context, request *pb.HelloRequest) (*pb.HelloReply, error) {
    reply := &pb.HelloReply{Message: "hello " + request.GetName()}
    return reply, nil
}

func (h *HelloService) Streaming(stream pb.HelloService_StreamingServer) error {
    for {
        req, err := stream.Recv()
        if err != nil {
            if err == io.EOF {
                return nil
            }
            return err
        }
    
        reply := &pb.HelloReply{Message: "hello " + req.GetName()}
    
        err = stream.Send(reply)
        if err != nil {
            return err
        }
        return nil
    }
}

services/user.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
package services

import (
    "context"
    pb "tests/grpc/services/pb"
    "io"
)

type UserService struct{}

func (u *UserService) Login(ctx context.Context, request *pb.UserLoginRequest) (*pb.UserLoginResponse, error) {
    var phones = []*pb.UserLoginResponse_PhoneNumber{
        &pb.UserLoginResponse_PhoneNumber{
            PhoneNumber: "返回手机号码:" + request.GetPhones()[0].PhoneNumber,
            PhoneType:   pb.UserLoginResponse_PhoneType(request.GetPhones()[0].PhoneType),
        },
    }
    var userNum = map[string]*pb.UserLoginResponse_PhoneNumber{
        "phone_001": &pb.UserLoginResponse_PhoneNumber{
            PhoneNumber: "001",
            PhoneType:   pb.UserLoginResponse_mobile,
        },
        "phone_002": &pb.UserLoginResponse_PhoneNumber{
            PhoneNumber: "002",
            PhoneType:   pb.UserLoginResponse_home,
        },
        "phone_003": &pb.UserLoginResponse_PhoneNumber{
            PhoneNumber: "003",
            PhoneType:   pb.UserLoginResponse_work,
        },
    }
    reply := &pb.UserLoginResponse{
        UserId:         request.GetUserId(),
        UserName:       "获取用户名称：" + request.GetUserName(),
        UserPwd:        "用户密码12345678900",
        UserGender:     request.GetUserGender(),
        UserCredits:    request.GetUserCredits(),
        IsAdmin:        request.GetIsAdmin(),
        AuthCode:       request.GetAuthCode(),
        UserBalance:    request.GetUserBalance(),
        UserHobby:      pb.UserLoginResponse_Hobby(request.GetUserHobby()),
        Phones:         phones,
        LastUpdateDate: request.GetLastUpdateDate(),
        UserNum:        userNum,
        LuckyNumbers:   request.GetLuckyNumbers(),
        UserEmail:      request.GetUserEmail(),
        UserWeight:     request.GetUserWeight(),
        LoginStatus:    request.GetLoginStatus(),
    }
    return reply, nil
}

// 双向流
func (u *UserService) LoginStreaming(stream pb.UserService_LoginStreamingServer) error {
    for {
        req, err := stream.Recv()
        if err != nil {
            if err == io.EOF {
                return nil
            }
            return err
        }
    
        var phones = []*pb.UserLoginResponse_PhoneNumber{
            &pb.UserLoginResponse_PhoneNumber{
                PhoneNumber: "返回手机号码:" + req.GetPhones()[0].PhoneNumber,
                PhoneType:   pb.UserLoginResponse_PhoneType(req.GetPhones()[0].PhoneType),
            },
        }
    
        var userNum = map[string]*pb.UserLoginResponse_PhoneNumber{
            "phone001": &pb.UserLoginResponse_PhoneNumber{
                PhoneNumber: "1111111111",
                PhoneType:   pb.UserLoginResponse_mobile,
            },
            "phone002": &pb.UserLoginResponse_PhoneNumber{
                PhoneNumber: "2222222222",
                PhoneType:   pb.UserLoginResponse_home,
            },
            "phone003": &pb.UserLoginResponse_PhoneNumber{
                PhoneNumber: "3333333333",
                PhoneType:   pb.UserLoginResponse_work,
            },
        }
    
        reply := &pb.UserLoginResponse{
            UserId:         req.GetUserId(),
            UserName:       "返回用户名称：" + req.GetUserName(),
            UserPwd:        "用户密码111111111111",
            UserGender:     req.GetUserGender(),
            UserCredits:    req.GetUserCredits(),
            IsAdmin:        req.GetIsAdmin(),
            AuthCode:       req.GetAuthCode(),
            UserBalance:    req.GetUserBalance(),
            UserHobby:      pb.UserLoginResponse_Hobby(req.GetUserHobby()),
            Phones:         phones,
            LastUpdateDate: req.GetLastUpdateDate(),
            UserNum:        userNum,
            LuckyNumbers:   req.GetLuckyNumbers(),
            UserEmail:      req.GetUserEmail(),
            UserWeight:     req.GetUserWeight(),
            LoginStatus:    req.GetLoginStatus(),
        }
    
        err = stream.Send(reply)
        if err != nil {
            return err
        }
        return nil
    }
}

services/test.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
package services

import (
    "context"
    pb "tests/grpc/services/pb"
    "strconv"
    
    "github.com/golang/protobuf/ptypes/empty"
    anypb "google.golang.org/protobuf/types/known/anypb"
)

type TestService struct{}

func (t *TestService) Test(ctx context.Context, req *pb.TestRequest) (*pb.TestResponse, error) {
    var extension = []*pb.Extension{
        &pb.Extension{
            Id:   req.GetExtensions()[0].Id,
            Data: req.GetExtensions()[0].GetData(),
        },
        &pb.Extension{
            Id:   100,
            Data: &anypb.Any{TypeUrl: "test any type `type_url`"},
        },
    }
    reply := &pb.TestResponse{
        Msg:        "ok 200",
        Name:       req.GetName() + ":" + strconv.FormatInt(int64(req.Extensions[0].Id), 10),
        Extensions: extension,
    }
    return reply, nil
}

func (t *TestService) Ping(ctx context.Context, req *empty.Empty) (*empty.Empty, error) {
    return &empty.Empty{}, nil
}
实现gRPC客户端

client/client.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
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
package main

import (
    "context"
    "crypto/tls"
    "crypto/x509"
    "flag"
    pb "tests/grpc/services/pb"
    "io"
    "io/ioutil"
    "log"
    "time"
    
    "github.com/golang/protobuf/ptypes/empty"
    "github.com/golang/protobuf/ptypes/timestamp"
    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials"
    "google.golang.org/protobuf/types/known/anypb"
)

var (
    address = flag.String("addr", "localhost:8888", "addr")
    name    = flag.String("name", "world", "name")
)

func main() {
    flag.Parse()
    
    cert, err := tls.LoadX509KeyPair("cert/client.pem", "cert/client.key")
    if err != nil {
        log.Fatalf("load cert files failed,err:%v", err)
    }
    
    certPool := x509.NewCertPool()
    ca, err := ioutil.ReadFile("cert/ca.pem")
    if err != nil {
        log.Fatalf("read ca file failed,err:%v", err)
    }
    if ok := certPool.AppendCertsFromPEM(ca); !ok {
        log.Fatal("failed to append ca certs")
    }
    
    creds := credentials.NewTLS(&tls.Config{
        Certificates: []tls.Certificate{cert}, // 客户端证书
        ServerName:   "localhost",
        RootCAs:      certPool,
    })
    
    //conn, err := grpc.Dial(*address, grpc.WithInsecure())
    conn, err := grpc.Dial(*address, grpc.WithTransportCredentials(creds))
    if err != nil {
        log.Fatalf("failed to connect server,err:%v", err)
    }
    defer conn.Close()
    
    // hello client
    helloClient := pb.NewHelloServiceClient(conn)
    sayHello(helloClient)
    
    // user client
    userClient := pb.NewUserServiceClient(conn)
    login(userClient)
    loginStreaming(userClient)
    
    // test client
    testClient := pb.NewTestServiceClient(conn)
    test(testClient)
    ping(testClient)
}

func sayHello(client pb.HelloServiceClient) {
    reply, err := client.SayHello(context.Background(), &pb.HelloRequest{Name: *name})
    if err != nil {
        log.Fatal(err)
    }
    log.Println(reply.GetMessage())
    
    stream, err := client.Streaming(context.Background())
    if err != nil {
        log.Fatalf("failed to call:%v", err)
    }
    go func() {
        for {
            err := stream.Send(&pb.HelloRequest{Name: *name})
            if err != nil {
                log.Fatalf("failed to send,err:%v", err)
                break
            }
            time.Sleep(time.Second)
        }
    }()
    
    for {
        reply, err := stream.Recv()
        if err != nil {
            if err == io.EOF {
                break
            }
            log.Fatalf("failed to recv,err:%v", err)
            break
        }
        log.Println(reply.GetMessage())
    }
}

func login(client pb.UserServiceClient) {
    var userEmail = map[string]string{"email": "test@qq.com"}
    var userWeight = map[string]int32{"weight": 170}
    var userNum = map[string]*pb.UserLoginRequest_PhoneNumber{
        "001": {PhoneNumber: "001", PhoneType: pb.UserLoginRequest_mobile},
        "002": {PhoneNumber: "002", PhoneType: pb.UserLoginRequest_home},
        "003": {PhoneNumber: "003", PhoneType: pb.UserLoginRequest_work},
    }
    var phones = []*pb.UserLoginRequest_PhoneNumber{
        {PhoneNumber: "1111111", PhoneType: pb.UserLoginRequest_mobile},
        {PhoneNumber: "2222222", PhoneType: pb.UserLoginRequest_home},
        {PhoneNumber: "3333333", PhoneType: pb.UserLoginRequest_work},
    }
    var lastUpdateDate = &timestamp.Timestamp{
        Seconds: time.Now().Unix(),
        Nanos:   int32(999999999),
    }
    
    request := &pb.UserLoginRequest{
        UserId:         int64(100),
        UserName:       "aaaaa",
        UserNickname:   "用户昵称",
        UserCountry:    "china",
        UserGender:     1,
        UserCredits:    111.111,
        IsAdmin:        true,
        AuthCode:       []byte("abcdefg"),
        UserBalance:    10000000.0000,
        UserHobby:      pb.UserLoginRequest_music,
        LuckyNumbers:   []int32{2, 4, 6, 8, 10},
        UserEmail:      userEmail,
        UserWeight:     userWeight,
        UserNum:        userNum,
        LoginStatus:    pb.LoginStatus_success,
        Phones:         phones,
        LastUpdateDate: lastUpdateDate,
    }
    resp, err := client.Login(context.Background(), request)
    if err != nil {
        log.Fatal(err)
    }
    log.Println(resp)
}

func loginStreaming(client pb.UserServiceClient) {
    var userEmail = map[string]string{"email": "test123@qq.com"}
    var userWeight = map[string]int32{"weight": 180}
    var userNum = map[string]*pb.UserLoginRequest_PhoneNumber{
        "test001": {PhoneNumber: "001", PhoneType: pb.UserLoginRequest_mobile},
        "test002": {PhoneNumber: "002", PhoneType: pb.UserLoginRequest_home},
        "test003": {PhoneNumber: "003", PhoneType: pb.UserLoginRequest_work},
    }
    var phones = []*pb.UserLoginRequest_PhoneNumber{
        {PhoneNumber: "test-1111111", PhoneType: pb.UserLoginRequest_mobile},
        {PhoneNumber: "test-2222222", PhoneType: pb.UserLoginRequest_home},
        {PhoneNumber: "test-3333333", PhoneType: pb.UserLoginRequest_work},
    }
    var lastUpdateDate = &timestamp.Timestamp{
        Seconds: time.Now().Unix(),
        Nanos:   int32(999999999),
    }
    
    login := &pb.UserLoginRequest{
        UserId:         int64(101),
        UserName:       "bbbbbbb",
        UserNickname:   "用户昵称",
        UserCountry:    "china",
        UserGender:     2,
        UserCredits:    222.222,
        IsAdmin:        true,
        AuthCode:       []byte("xxxxx"),
        UserBalance:    20000000.0000,
        UserHobby:      pb.UserLoginRequest_tour,
        LuckyNumbers:   []int32{34, 55, 333, 33, 199},
        UserEmail:      userEmail,
        UserWeight:     userWeight,
        UserNum:        userNum,
        LoginStatus:    pb.LoginStatus_error,
        Phones:         phones,
        LastUpdateDate: lastUpdateDate,
    }
    stream, err := client.LoginStreaming(context.Background())
    if err != nil {
        log.Fatalf("failed to call:%v", err)
    }
    go func() {
        for {
            err := stream.Send(login)
            if err != nil {
                log.Fatalf("failed to send,err:%v", err)
                break
            }
            time.Sleep(time.Second)
        }
    }()
    
    for {
        reply, err := stream.Recv()
        if err != nil {
            if err == io.EOF {
                break
            }
            log.Printf("failed to recv,err:%v", err)
            break
        }
        log.Println(reply)
    }
}

func test(client pb.TestServiceClient) {
    req := &pb.TestRequest{
        Name: "test",
        Extensions: []*pb.Extension{
            &pb.Extension{Id: 8888, Data: nil},
            &pb.Extension{Id: 9999, Data: &anypb.Any{TypeUrl: "type url"}},
        },
    }
    reply, err := client.Test(context.Background(), req)
    if err != nil {
        log.Fatal(err)
    }
    log.Println(reply)
}

func ping(client pb.TestServiceClient) {
    reply, err := client.Ping(context.Background(), &empty.Empty{})
    if err != nil {
        log.Fatal(err)
    }
    log.Printf("%v", reply)
}
gRPC服务端

server/server.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
package main

import (
    "crypto/tls"
    "crypto/x509"
    "flag"
    "tests/grpc/services"
    pb "tests/grpc/services/pb"
    "io/ioutil"
    "log"
    "net"
    
    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials"
)

var (
    serverPort = flag.String("port", ":8888", "port")
)

func main() {
    flag.Parse()
    
    cert, err := tls.LoadX509KeyPair("cert/server.pem", "cert/server.key")
    if err != nil {
        log.Fatalf("load cert files failed,err:%v", err)
    }
    
    certPool := x509.NewCertPool()
    ca, err := ioutil.ReadFile("cert/ca.pem")
    if err != nil {
        log.Fatalf("read ca file failed,err:%v", err)
    }
    
    if ok := certPool.AppendCertsFromPEM(ca); !ok {
        log.Fatal("failed to append certs")
    }
    
    creds := credentials.NewTLS(&tls.Config{
        Certificates: []tls.Certificate{cert}, // 服务端证书
        ClientAuth:   tls.RequireAndVerifyClientCert,
        ClientCAs:    certPool,
    })
    
    server := grpc.NewServer(grpc.Creds(creds))
    
    pb.RegisterHelloServiceServer(server, new(services.HelloService))
    pb.RegisterUserServiceServer(server, new(services.UserService))
    pb.RegisterTestServiceServer(server, new(services.TestService))
    
    l, err := net.Listen("tcp", *serverPort)
    if err != nil {
        log.Fatalf("failed to listen,err:%v", err)
    }
    
    log.Printf("server start at [%s]", *serverPort)
    
    err = server.Serve(l)
    if err != nil {
        log.Fatal(err)
    }
}
gRPC使用实例

查看目录结构：

1
cd tests

查看目录层级：

1
tree -L 1
1
2
3
4
5
6
7
└── grpc
    ├── client
    ├── proto
    ├── server
    └── services

5 directories, 0 files
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
└── grpc
    ├── client
    │   ├── cert
    │   │   ├── ca.pem
    │   │   ├── client.key
    │   │   └── client.pem
    │   └── client.go
    ├── proto
    │   ├── common.proto
    │   ├── generate.sh
    │   ├── hello.proto
    │   └── user.proto
    ├── server
    │   ├── auth_server.go
    │   ├── cert
    │   │   ├── ca.pem
    │   │   ├── server.key
    │   │   └── server.pem
    │   └── server.go
    └── services
        ├── hello.go
        ├── pb
        │   ├── common.pb.go
        │   ├── hello.pb.go
        │   └── user.pb.go
        ├── test.go
        └── user.go

8 directories, 19 files

🌴🌴🧩🧩🥥Gopher

心外无物，知行合一