其他分享
首页 > 其他分享> > gRPC学习之四:实战四类服务方法

gRPC学习之四:实战四类服务方法

作者:互联网

欢迎访问我的GitHub

https://github.com/zq2599/blog_demos

内容:所有原创文章分类汇总及配套源码,涉及Java、Docker、Kubernetes、DevOPS等;

gRPC学习系列文章链接

  1. 在CentOS7部署和设置GO
  2. GO的gRPC开发环境准备
  3. 初试GO版gRPC开发
  4. 实战四类服务方法
  5. gRPC-Gateway实战
  6. gRPC-Gateway集成swagger

本篇概览

  1. 单项 RPC,即客户端发送一个请求给服务端,从服务端获取一个应答,就像一次普通的函数调用(前一篇文章就是此类);
  2. 服务端流式 RPC,即客户端发送一个请求给服务端,可获取一个数据流用来读取一系列消息。客户端从返回的数据流里一直读取直到没有更多消息为止;
  3. 客户端流式 RPC,即客户端用提供的一个数据流写入并发送一系列消息给服务端。一旦客户端完成消息写入,就等待服务端读取这些消息并返回应答;
  4. 双向流式 RPC,即两边都可以分别通过一个读写数据流来发送一系列消息。这两个数据流操作是相互独立的,所以客户端和服务端能按其希望的任意顺序读写,例如:服务端可以在写应答前等待所有的客户端消息,或者它可以先读一个消息再写一个消息,或者是读写相结合的其他方式。每个数据流里消息的顺序会被保持。

在这里插入图片描述

源码下载

名称 链接 备注
项目主页 https://github.com/zq2599/blog_demos 该项目在GitHub上的主页
git仓库地址(https) https://github.com/zq2599/blog_demos.git 该项目源码的仓库地址,https协议
git仓库地址(ssh) git@github.com:zq2599/blog_demos.git 该项目源码的仓库地址,ssh协议

在这里插入图片描述

在这里插入图片描述

提前说明文件和目录

[golang@centos7 src]$ tree grpcstream/
grpcstream/
├── client
│   └── client.go
├── grpcstream.pb.go
├── grpcstream.proto
└── server
    └── server.go

编写proto文件

// 协议类型
syntax = "proto3";

// 包名
package grpcstream;

// 服务端请求的数据结构
message SingleRequest {
  int32 id = 1;
}

// 服务端响应的数据结构
message SingleResponse {
  int32 id = 1;
  string name = 2;
}

// 定义的服务名
service IGrpcStremService {
  // 单项RPC :单个请求,单个响应
  rpc SingleReqSingleResp (SingleRequest) returns (SingleResponse);

  // 服务端流式 :单个请求,集合响应
  rpc SingleReqMultiResp (SingleRequest) returns (stream SingleResponse);

  // 客户端流式 :集合请求,单个响应
  rpc MultiReqSingleResp (stream SingleRequest) returns (SingleResponse);

  // 双向流式 :集合请求,集合响应
  rpc MultiReqMultiResp (stream SingleRequest) returns (stream SingleResponse);
}
  1. 方法SingleReqSingleResp非常简单,和上一篇文章中的demo一样,入参是一个数据结构,服务端返回的也是一个数据结构;
  2. 方法SingleReqSingleResp是服务端流式类型,特征是返回值用stream修饰;
  3. 方法MultiReqSingleResp是客户端流式类型,特征是入参用stream修饰;
  4. 方法MultiReqMultiResp是双向类型,特征是入参和返回值都用stream修饰;

根据proto生成go源码

  1. grpcstream.proto所在的目录,执行以下命令:
protoc --go_out=plugins=grpc:. grpcstream.proto
  1. 如果grpcstream.proto没有语法错误,会在当前目录生成文件grpcstream.pb.go,这里面是工具protoc-gen-go自动生成的代码,里面生成的代码在开发服务端和客户端时都会用到;
  2. 对服务端来说,grpcstream.pb.go中最重要的是IGrpcStremServiceServer接口 ,服务端需要实现该接口所有的方法作为业务逻辑,接口定义如下:
type IGrpcStremServiceServer interface {
	// 单项流式 :单个请求,单个响应
	SingleReqSingleResp(context.Context, *SingleRequest) (*SingleResponse, error)
	// 服务端流式 :单个请求,集合响应
	SingleReqMultiResp(*SingleRequest, IGrpcStremService_SingleReqMultiRespServer) error
	// 客户端流式 :集合请求,单个响应
	MultiReqSingleResp(IGrpcStremService_MultiReqSingleRespServer) error
	// 双向流式 :集合请求,集合响应
	MultiReqMultiResp(IGrpcStremService_MultiReqMultiRespServer) error
}
  1. 对客户端来说,grpcstream.pb.go中最重要的是IGrpcStremServiceClient接口,如下所示,这意味这客户端可以发起哪些远程调用 :
type IGrpcStremServiceClient interface {
	// 单项流式 :单个请求,单个响应
	SingleReqSingleResp(ctx context.Context, in *SingleRequest, opts ...grpc.CallOption) (*SingleResponse, error)
	// 服务端流式 :单个请求,集合响应
	SingleReqMultiResp(ctx context.Context, in *SingleRequest, opts ...grpc.CallOption) (IGrpcStremService_SingleReqMultiRespClient, error)
	// 客户端流式 :集合请求,单个响应
	MultiReqSingleResp(ctx context.Context, opts ...grpc.CallOption) (IGrpcStremService_MultiReqSingleRespClient, error)
	// 双向流式 :集合请求,集合响应
	MultiReqMultiResp(ctx context.Context, opts ...grpc.CallOption) (IGrpcStremService_MultiReqMultiRespClient, error)
}

编写服务端代码server.go并启动

package main

import (
	"context"
	"google.golang.org/grpc"
	pb "grpcstream"
	"io"
	"log"
	"net"
	"strconv"
)

// 常量:监听端口
const (
	port = ":50051"
)

// 定义结构体,在调用注册api的时候作为入参,
// 该结构体会带上proto中定义的方法,里面是业务代码
// 这样远程调用时就执行了业务代码了
type server struct {
	// pb.go中自动生成的,是个空结构体
	pb.UnimplementedIGrpcStremServiceServer
}

// 单项流式 :单个请求,单个响应
func (s *server) SingleReqSingleResp(ctx context.Context, req *pb.SingleRequest) (*pb.SingleResponse, error) {
	id := req.GetId()

	// 打印请求参数
	log.Println("1. 收到请求:", id)
	// 实例化结构体SingleResponse,作为返回值
	return &pb.SingleResponse{Id: id, Name: "1. name-" + strconv.Itoa(int(id))}, nil
}

// 服务端流式 :单个请求,集合响应
func (s *server) SingleReqMultiResp(req *pb.SingleRequest, stream pb.IGrpcStremService_SingleReqMultiRespServer) error {
	// 取得请求参数
	id := req.GetId()

	// 打印请求参数
	log.Println("2. 收到请求:", id)

	// 返回多条记录
	for i := 0; i < 10; i++ {
		stream.Send(&pb.SingleResponse{Id: int32(i), Name: "2. name-" + strconv.Itoa(i)})
	}

	return nil
}

// 客户端流式 :集合请求,单个响应
func (s *server) MultiReqSingleResp(reqStream pb.IGrpcStremService_MultiReqSingleRespServer) error {
	var addVal int32 = 0

	// 在for循环中接收流式请求
	for {
		// 一次接受一条记录
		singleRequest, err := reqStream.Recv()

		// 不等于io.EOF表示这是条有效记录
		if err == io.EOF {
			log.Println("3. 客户端发送完毕")
			break
		} else if err != nil {
			log.Fatalln("3. 接收时发生异常", err)
			break
		} else {
			log.Println("3. 收到请求:", singleRequest.GetId())
			// 收完之后,执行SendAndClose返回数据并结束本次调用
			addVal += singleRequest.GetId()
		}
	}

	return reqStream.SendAndClose(&pb.SingleResponse{Id: addVal, Name: "3. name-" + strconv.Itoa(int(addVal))})
}

// 双向流式 :集合请求,集合响应
func (s *server) MultiReqMultiResp(reqStream pb.IGrpcStremService_MultiReqMultiRespServer) error {
	// 简单处理,对于收到的每一条记录都返回一个响应
	for {
		singleRequest, err := reqStream.Recv()

		// 不等于io.EOS表示这是条有效记录
		if err == io.EOF {
			log.Println("4. 接收完毕")
			return nil
		} else if err != nil {
			log.Fatalln("4. 接收时发生异常", err)
			return err
		} else {
			log.Println("4. 接收到数据", singleRequest.GetId())

			id := singleRequest.GetId()

			if sendErr := reqStream.Send(&pb.SingleResponse{Id: id, Name: "4. name-" + strconv.Itoa(int(id))}); sendErr != nil {
				log.Println("4. 返回数据异常数据", sendErr)
				return sendErr
			}
		}
	}
}

func main() {
	// 要监听的协议和端口
	lis, err := net.Listen("tcp", port)
	if err != nil {
		log.Fatalf("failed to listen: %v", err)
	}

	// 实例化gRPC server结构体
	s := grpc.NewServer()

	// 服务注册
	pb.RegisterIGrpcStremServiceServer(s, &server{})

	log.Println("开始监听,等待远程调用...")

	if err := s.Serve(lis); err != nil {
		log.Fatalf("failed to serve: %v", err)
	}
}
  1. SingleReqMultiResp方法的作用是收到客户端一个请求参数,然后向客户端发送多个响应,可见多次调用stream.Send方法即可多次发送数据到客户端;
  2. MultiReqSingleResp方法可以从客户端收到多条数据,可见是在for循环中重复调用reqStream.Recv()方法,直到收到客户端的io.EOF为止,这就要就客户端在发送完数据后再给一个io.EOF过来,稍后的客户端代码会展示如何做;
  3. MultiReqMultiResp方法持续接受客户端数据,并且持续发送数据给客户端,一定要把顺序问题考虑清楚,否则会陷入异常(例如一方死循环),我这里的逻辑是收到客户端的io.EOF为止,这就要就客户端在发送完数据后再给一个io.EOF过来,如果客户端也在用for循环一直等数据,那就是双方都在等数据了,无法终止程序,稍后的客户端代码会展示如何做;
[golang@centos7 server]$ go run server.go 
2020/12/13 21:29:19 开始监听,等待远程调用...

编写客户端代码client.go

package main

import (
	"context"
	"google.golang.org/grpc"
	"io"
	"log"
	"time"

	pb "grpcstream"
)

const (
	address     = "localhost:50051"
	defaultId = "666"
)

func main() {
	// 远程连接服务端
	conn, err := grpc.Dial(address, grpc.WithInsecure(), grpc.WithBlock())
	if err != nil {
		log.Fatalf("did not connect: %v", err)
	}

	// main方法执行完毕后关闭远程连接
	defer conn.Close()

	// 实例化数据结构
	client := pb.NewIGrpcStremServiceClient(conn)

	// 超时设置
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)

	defer cancel()

	log.Println("测试单一请求应答,一对一")
	singleReqSingleResp(ctx, client)

	log.Println("测试服务端流式应答,一对多")
	singleReqMultiResp(ctx, client)

	log.Println("测试客户端流式请求,多对一")
	multiReqSingleResp(ctx, client)

	log.Println("测试双向流式请求应答,多对多")
	multiReqMultiResp(ctx, client)

	log.Println("测试完成")
}


func singleReqSingleResp(ctx context.Context, client pb.IGrpcStremServiceClient) error {
	// 远程调用
	r, err := client.SingleReqSingleResp(ctx, &pb.SingleRequest{Id: 101})

	if err != nil {
		log.Fatalf("1. 远程调用异常 : %v", err)
		return err
	}

	// 将服务端的返回信息打印出来
	log.Printf("response, id : %d, name : %s", r.GetId(), r.GetName())

	return nil
}


func singleReqMultiResp(ctx context.Context, client pb.IGrpcStremServiceClient) error {
	// 远程调用
	recvStream, err := client.SingleReqMultiResp(ctx, &pb.SingleRequest{Id: 201})

	if err != nil {
		log.Fatalf("2. 远程调用异常 : %v", err)
		return err
	}

	for {
		singleResponse, err := recvStream.Recv()
		if err == io.EOF {
			log.Printf("2. 获取数据完毕")
			break
		}

		log.Printf("2. 收到服务端响应, id : %d, name : %s", singleResponse.GetId(), singleResponse.GetName())
	}

	return nil
}

func multiReqSingleResp(ctx context.Context, client pb.IGrpcStremServiceClient) error {
	// 远程调用
	sendStream, err := client.MultiReqSingleResp(ctx)

	if err != nil {
		log.Fatalf("3. 远程调用异常 : %v", err)
		return err
	}

	// 发送多条记录到服务端
	for i:=0; i<10; i++ {
		if err = sendStream.Send(&pb.SingleRequest{Id: int32(300+i)}); err!=nil {
			log.Fatalf("3. 通过流发送数据异常 : %v", err)
			return err
		}
	}

	singleResponse, err := sendStream.CloseAndRecv()

	if err != nil {
		log.Fatalf("3. 服务端响应异常 : %v", err)
		return err
	}

	// 将服务端的返回信息打印出来
	log.Printf("response, id : %d, name : %s", singleResponse.GetId(), singleResponse.GetName())

	return nil
}

func multiReqMultiResp(ctx context.Context, client pb.IGrpcStremServiceClient) error {
	// 远程调用
	intOutStream, err := client.MultiReqMultiResp(ctx)

	if err != nil {
		log.Fatalf("4. 远程调用异常 : %v", err)
		return err
	}

	// 发送多条记录到服务端
	for i:=0; i<10; i++ {
		if err = intOutStream.Send(&pb.SingleRequest{Id: int32(400+i)}); err!=nil {
			log.Fatalf("4. 通过流发送数据异常 : %v", err)
			return err
		}
	}

	// 服务端一直在接收,直到收到io.EOF为止
	// 因此,这里必须发送io.EOF到服务端,让服务端知道发送已经结束(很重要)
	intOutStream.CloseSend()

	// 接收服务端发来的数据
	for {
		singleResponse, err := intOutStream.Recv()
		if err == io.EOF {
			log.Printf("4. 获取数据完毕")
			break
		} else if err != nil {
			log.Fatalf("4. 接收服务端数据异常 : %v", err)
			break
		}

		log.Printf("4. 收到服务端响应, id : %d, name : %s", singleResponse.GetId(), singleResponse.GetName())
	}

	return nil
}
  1. singleReqMultiResp方法会接收服务端的多条记录,在for循环中调用recvStream.Recv即可收到所有数据;
  2. multiReqSingleResp方法会向服务端发送多条数据,由于服务端在等待io.EOF作为结束标志,因此调用sendStream.CloseAndRecv即可发送io.EOF,并得到服务端的返回值;
  3. multiReqMultiResp方法在持续向服务端发送数据,并且也在持续获取服务端发来的数据,在发送数据完成后,必须调用intOutStream.CloseSend方法,即可发送io.EOF,让服务端不再接收数据,避免前面提到的死循环;
  4. 在main方法中,依次发起四类服务方法的调用;

执行客户端

[golang@centos7 client]$ go run client.go 
2020/12/13 21:39:35 测试单一请求应答,一对一
2020/12/13 21:39:35 response, id : 101, name : 1. name-101
2020/12/13 21:39:35 测试服务端流式应答,一对多
2020/12/13 21:39:35 2. 收到服务端响应, id : 0, name : 2. name-0
2020/12/13 21:39:35 2. 收到服务端响应, id : 1, name : 2. name-1
2020/12/13 21:39:35 2. 收到服务端响应, id : 2, name : 2. name-2
2020/12/13 21:39:35 2. 收到服务端响应, id : 3, name : 2. name-3
2020/12/13 21:39:35 2. 收到服务端响应, id : 4, name : 2. name-4
2020/12/13 21:39:35 2. 收到服务端响应, id : 5, name : 2. name-5
2020/12/13 21:39:35 2. 收到服务端响应, id : 6, name : 2. name-6
2020/12/13 21:39:35 2. 收到服务端响应, id : 7, name : 2. name-7
2020/12/13 21:39:35 2. 收到服务端响应, id : 8, name : 2. name-8
2020/12/13 21:39:35 2. 收到服务端响应, id : 9, name : 2. name-9
2020/12/13 21:39:35 2. 获取数据完毕
2020/12/13 21:39:35 测试客户端流式请求,多对一
2020/12/13 21:39:35 response, id : 3045, name : 3. name-3045
2020/12/13 21:39:35 测试双向流式请求应答,多对多
2020/12/13 21:39:35 4. 收到服务端响应, id : 400, name : 4. name-400
2020/12/13 21:39:35 4. 收到服务端响应, id : 401, name : 4. name-401
2020/12/13 21:39:35 4. 收到服务端响应, id : 402, name : 4. name-402
2020/12/13 21:39:35 4. 收到服务端响应, id : 403, name : 4. name-403
2020/12/13 21:39:35 4. 收到服务端响应, id : 404, name : 4. name-404
2020/12/13 21:39:35 4. 收到服务端响应, id : 405, name : 4. name-405
2020/12/13 21:39:35 4. 收到服务端响应, id : 406, name : 4. name-406
2020/12/13 21:39:35 4. 收到服务端响应, id : 407, name : 4. name-407
2020/12/13 21:39:35 4. 收到服务端响应, id : 408, name : 4. name-408
2020/12/13 21:39:35 4. 收到服务端响应, id : 409, name : 4. name-409
2020/12/13 21:39:35 4. 获取数据完毕
2020/12/13 21:39:35 测试完成
[golang@centos7 server]$ go run server.go 
2020/12/13 21:29:19 开始监听,等待远程调用...
2020/12/13 21:39:35 1. 收到请求: 101
2020/12/13 21:39:35 2. 收到请求: 201
2020/12/13 21:39:35 3. 收到请求: 300
2020/12/13 21:39:35 3. 收到请求: 301
2020/12/13 21:39:35 3. 收到请求: 302
2020/12/13 21:39:35 3. 收到请求: 303
2020/12/13 21:39:35 3. 收到请求: 304
2020/12/13 21:39:35 3. 收到请求: 305
2020/12/13 21:39:35 3. 收到请求: 306
2020/12/13 21:39:35 3. 收到请求: 307
2020/12/13 21:39:35 3. 收到请求: 308
2020/12/13 21:39:35 3. 收到请求: 309
2020/12/13 21:39:35 3. 客户端发送完毕
2020/12/13 21:39:35 4. 接收到数据 400
2020/12/13 21:39:35 4. 接收到数据 401
2020/12/13 21:39:35 4. 接收到数据 402
2020/12/13 21:39:35 4. 接收到数据 403
2020/12/13 21:39:35 4. 接收到数据 404
2020/12/13 21:39:35 4. 接收到数据 405
2020/12/13 21:39:35 4. 接收到数据 406
2020/12/13 21:39:35 4. 接收到数据 407
2020/12/13 21:39:35 4. 接收到数据 408
2020/12/13 21:39:35 4. 接收到数据 409
2020/12/13 21:39:35 4. 接收完毕

你不孤单,欣宸原创一路相伴

  1. Java系列
  2. Spring系列
  3. Docker系列
  4. kubernetes系列
  5. 数据库+中间件系列
  6. DevOps系列

欢迎关注公众号:程序员欣宸

微信搜索「程序员欣宸」,我是欣宸,期待与您一同畅游Java世界...
https://github.com/zq2599/blog_demos

标签:13,12,21,gRPC,39,四类,2020,之四,服务端
来源: https://www.cnblogs.com/bolingcavalry/p/15150683.html