grpc快速入门示例

为什么会有这个话题？

之前有大佬分享grpc的使用，当时很有兴趣听了。别人用很长使用得来的经验，把自己对于某个东西的使用做了总结。但是对于像我这样没有使用过的人来说，听了过后很多就忘了。所以我自己的理解是，如果要学一个东西并且使用，那么就得自己动手，自己撸代码，也就有了这个话题的由来。

与其说是学习分享，听后实践更恰当。
代码地址

https://github.com/Fish-pro/grpc-demo
1

1.定义.proto

service ToDoService{
  rpc Create(CreateRequest) returns (CreateResponse){
    option(google.api.http) = {
      post:"/v1/todo"
      body:"*"
    };
  }
  rpc Read(ReadRequest) returns (ReadResponse){
    option(google.api.http) = {
      get:"/v1/todo/{id}"
    };
  }
  rpc ReadAll(ReadAllRequest) returns (ReadAllResponse){
    option(google.api.http) = {
      get:"/v1/todo/all"
    };
  }
  rpc Update(UpdateRequest) returns (UpdateResponse){
    option(google.api.http) = {
      put:"/v1/todo/{toDo.id}"
      body:"*"
      additional_bindings:{
        patch:"/v1/todo/{toDo.id}"
      }
    };
  }
  rpc Delete(DeleteRequest) returns (DeleteResponse){
    option(google.api.http) = {
      delete:"/v1/todo/{id}"
    };
  }
}
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

2.实现接口

执行脚本，会生成.pb.go的文件，文件下可以看到如下的接口定义，那么我们只需要实现这些接口，就能实现我们所定义的服务。

protoc --go_out=plugins=grpc:. *.proto
1

// ToDoServiceServer is the server API for ToDoService service.
type ToDoServiceServer interface {
	Create(context.Context, *CreateRequest) (*CreateResponse, error)
	Read(context.Context, *ReadRequest) (*ReadResponse, error)
	ReadAll(context.Context, *ReadAllRequest) (*ReadAllResponse, error)
	Update(context.Context, *UpdateRequest) (*UpdateResponse, error)
	Delete(context.Context, *DeleteRequest) (*DeleteResponse, error)
}
1
2
3
4
5
6
7
8

在函数中实现接口，如下图所示

3.启动服务

启动grpc服务

	// New grpc server
	var server *grpc.Server
	// if need certificate
	opts := []grpc.ServerOption{}
	opts = middleware.AddLogging(opts)
	if cfg.OpenPem {
		opts = append(opts, grpc.Creds(helper.GetServerCred(cfg.Cert)))
	}
	server = grpc.NewServer(opts...)
	// registry todoServiceServer in version v1
	v1ToDoAPI := serviceV1.NewToDoServiceServer(db)
	v1.RegisterToDoServiceServer(server, v1ToDoAPI)

	go func() {
		log.Println("starting gRPC server...")
		err = server.Serve(listen)
		if err != nil {
			log.Fatalf("start gRPC server error:%v", err)
		}
	}()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

4.grpc-gateway

执行脚本，生成.pb.gw.go文件，注册服务后，即可开启http服务。

protoc --grpc-gateway_out=logtostderr=true:. *.proto
1

	// New Mux
	gwmux := runtime.NewServeMux()
	var opts []grpc.DialOption
	// if need certificate
	if cfg.OpenPem {
		opts = []grpc.DialOption{grpc.WithTransportCredentials(helper.GetClientCred())}
	} else {
		opts = []grpc.DialOption{grpc.WithInsecure()}
	}
	endpoint := strings.Join([]string{cfg.Host, cfg.GRPCPort}, ":")

	// registry handler endpoint
	err := v1.RegisterToDoServiceHandlerFromEndpoint(ctx, gwmux, endpoint, opts)
	if err != nil {
		log.Fatalf("registry http server error:%v", err)
	}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

5.swagger

执行脚本，生成swagger.json,

protoc --swagger_out=logtostderr=true:../swagger *.proto
1

	// add swagger api
	mux := http.NewServeMux()
	mux.Handle("/", gwmux)
	dir := filepath.Join(cfg.BaseDir, "api/proto/swagger")
	mux.Handle("/api/", http.StripPrefix("/api/", http.FileServer(http.Dir(dir))))
1
2
3
4
5

在html中指定swagger.json文件位置

6.双向认证

生成根证书

genrsa -out ca.key 2048
req -new -x509 -days 3650 -key ca.key -out ca.pem
1
2

生成服务端证书

genrsa -out server.key 2048
req -new -key server.key -out server.csr // 本地使用localhost作为域名
x509 -req -sha256 -CA ca.pem -CAkey ca.key -CAcreateserial -days 3650 -in server.csr -out server.pem
1
2
3

服务端添加证书认证

	cert, _ := tls.LoadX509KeyPair("cert/server.pem", "cert/server.key")
	certPool := x509.NewCertPool()
	ca, _ := ioutil.ReadFile("cert/ca.pem")
	certPool.AppendCertsFromPEM(ca)

	cred := credentials.NewTLS(&tls.Config{
		Certificates: []tls.Certificate{cert},        // 服务端证书
		ClientAuth:   tls.RequireAndVerifyClientCert, // 双向验证
		ClientCAs:    certPool,
	})

	rpcServer := grpc.NewServer(grpc.Creds(cred))
	services.RegisterProdServiceServer(rpcServer, new(services.ProdService))

	lis, err := net.Listen("tcp", ":8081")
	if err != nil {
		log.Println(err.Error())
		os.Exit(1)
	}
	rpcServer.Serve(lis)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

生成客户端证书

ecparam -genkey -name secp384r1 -out client.key
req -new -key client.key -out client.csr
x509 -req -sha256 -CA ca.pem -CAkey ca.key -CAcreateserial -days 3650 -in client.csr -out client.pem
1
2
3

客户端添加证书

	cert, _ := tls.LoadX509KeyPair("cert/client.pem", "cert/client.key")
	certPool := x509.NewCertPool()
	ca, _ := ioutil.ReadFile("cert/ca.pem")
	certPool.AppendCertsFromPEM(ca)

	cred := credentials.NewTLS(&tls.Config{
		Certificates: []tls.Certificate{cert}, // 客户端证书
		ServerName:   "localhost",             // 域名
		RootCAs:      certPool,
	})
1
2
3
4
5
6
7
8
9
10

7.流模式

普通方法

服务端代码

// 普通方法
func (*UserService) GetUserScore(ctx context.Context, in *UserScoreRequest) (*UserScoreResponse, error) {
	var score int32 = 101
	users := make([]*UserInfo, 0)
	for _, user := range in.Users {
		user.UserScore = score
		score++
		users = append(users, user)
	}
	return &UserScoreResponse{Users: users}, nil
}
1
2
3
4
5
6
7
8
9
10
11

客户端代码

	// 普通方法获取
	var i int32
	req := services.UserScoreRequest{}
	req.Users = make([]*services.UserInfo, 0)
	
	for i = 1; i < 6; i++ {
		req.Users = append(req.Users, &services.UserInfo{UserId: i})
	}
	
	response, err := userClient.GetUserScore(context.Background(), &req)
	if err != nil {
		log.Println(err.Error())
		os.Exit(1)
	}
	fmt.Println(response.Users)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

服务端流式

服务端代码

//服务端流式
func (*UserService) GetUserScoreByServerStream(in *UserScoreRequest, stream UserService_GetUserScoreByServerStreamServer) error {
	var score int32 = 101
	users := make([]*UserInfo, 0)
	for index, user := range in.Users {
		user.UserScore = score
		score++
		users = append(users, user)
		if (index+1)%2 == 0 && index > 0 {
			err := stream.Send(&UserScoreResponse{Users: users})
			if err != nil {
				return err
			}
			users = users[0:0]
		}
		time.Sleep(time.Second * 1)
	}
	if len(users) > 0 {
		err := stream.Send(&UserScoreResponse{Users: users})
		if err != nil {
			return err
		}
	}
	return nil
}
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

客户端代码

	// 服务端流式
	var i int32
	req := services.UserScoreRequest{}
	req.Users = make([]*services.UserInfo, 0)
	
	for i = 1; i < 6; i++ {
		req.Users = append(req.Users, &services.UserInfo{UserId: i})
	}
	
	stream, err := userClient.GetUserScoreByServerStream(context.Background(), &req)
	if err != nil {
		log.Println(err.Error())
		os.Exit(1)
	}
	
	for {
		res, err := stream.Recv()
		if err == io.EOF {
			break
		}
		if err != nil {
			log.Println(err.Error())
			os.Exit(1)
		}
		fmt.Println(res.Users)
	}

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

客户端流式

服务端代码

// 客户端流式
func (*UserService) GetUserScoreByClientStream(stream UserService_GetUserScoreByClientStreamServer) error {
	var score int32 = 101
	users := make([]*UserInfo, 0)
	for {
		req, err := stream.Recv()
		if err == io.EOF {
			return stream.SendAndClose(&UserScoreResponse{Users: users})
		}
		if err != nil {
			return err
		}
		for _, user := range req.Users {
			user.UserScore = score
			score++
			users = append(users, user)
		}
	}
	return nil
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

客户端代码

	// 客户端流式
	var i int32
	
	stream, err := userClient.GetUserScoreByClientStream(context.Background())
	if err != nil {
		log.Println(err.Error())
		os.Exit(1)
	}
	
	for j := 1; j <= 3; j++ {
		req := services.UserScoreRequest{}
		req.Users = make([]*services.UserInfo, 0)
		for i = 1; i < 6; i++ {
			req.Users = append(req.Users, &services.UserInfo{UserId: i})
		}
		err = stream.Send(&req)
		if err != nil {
			log.Println(err.Error())
			os.Exit(1)
		}
	}
	response, err := stream.CloseAndRecv()
	if err != nil {
		log.Println(err.Error())
		os.Exit(1)
	}
	fmt.Println(response.Users)
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

双向流式

服务端代码

// 双向流式
func (*UserService) GetUserScoreByTWS(stream UserService_GetUserScoreByTWSServer) error {
	var score int32 = 101
	users := make([]*UserInfo, 0)
	for {
		req, err := stream.Recv()
		if err == io.EOF {
			return nil
		}
		if err != nil {
			return err
		}
		for _, user := range req.Users {
			user.UserScore = score
			score++
			users = append(users, user)
		}
		err = stream.Send(&UserScoreResponse{Users: users})
		if err != nil {
			fmt.Println(err.Error())
		}
		users = users[0:0]
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

客户端代码

	// 双向流式
	stream, err := userClient.GetUserScoreByTWS(context.Background())
	if err != nil {
		log.Println(err.Error())
		os.Exit(1)
	}

	var uid int32 = 1
	for j := 1; j <= 3; j++ {
		req := services.UserScoreRequest{}
		req.Users = make([]*services.UserInfo, 0)
		for i := 1; i < 6; i++ {
			req.Users = append(req.Users, &services.UserInfo{UserId: uid})
			uid++
		}
		err = stream.Send(&req)
		if err != nil {
			log.Println(err.Error())
			os.Exit(1)
		}

		res, err := stream.Recv()
		if err == io.EOF {
			break
		}
		if err != nil {
			log.Println(err.Error())
		}
		fmt.Println(res.Users)
	}

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