首页 > 其他分享> 文章详细

Go双向流式gRPC

2022-05-15 18:33:00 阅读：295 来源： 互联网

标签：11 stream err gRPC 28 流式 rpc client Go

Go gRPC教程-双向流式RPC

一、前言

上一篇介绍了客户端流式RPC，客户端不断的向服务端发送数据流，在发送结束或流关闭后，由服务端返回一个响应。本篇将介绍双向流式RPC。

双向流式RPC：客户端和服务端双方使用读写流去发送一个消息序列，两个流独立操作，双方可以同时发送和同时接收。

情景模拟：双方对话（可以一问一答、一问多答、多问一答，形式灵活）。

源码

二、新建proto文件

新建bothStream.proto文件

1.定义发送信息

// 定义流式请求信息
message StreamRequest{
    //流请求参数
    string question = 1;
}

2.定义接收信息

// 定义流式响应信息
message StreamResponse{
    //流响应数据
    string answer = 1;
}

3.定义服务方法Conversations

双向流式rpc，只要在请求的参数前和响应参数前都添加stream即可

service Stream{
    // 双向流式rpc，同时在请求参数前和响应参数前加上stream
    rpc Conversations(stream StreamRequest) returns(stream StreamResponse){};
}

4.编译proto文件

syntax = "proto3";// 协议为proto3

package proto;
option go_package = "./;proto";

//  生成pb.go命令:  protoc -I ./ --go_out=plugins=grpc:.\05bothStream\proto\  .\05bothStream\proto\bothStream.proto

// 定义我们的服务（可定义多个服务,每个服务可定义多个接口）
service Stream{
  // 双向流式rpc，同时在请求参数前和响应参数前加上stream
  rpc Conversations(stream StreamRequest) returns(stream StreamResponse){};
}

// 定义发送请求信息
message StreamRequest{
  //流请求参数
  string question = 1;
}

// 定义流式响应信息
message StreamResponse{
  //流响应数据
  string answer = 1;
}

进入bothStream.proto所在目录，运行指令:

go-grpc-example> protoc -I ./ --go_out=plugins=grpc:.\05bothStream\proto\  .\05bothStream\proto\bothStream.proto

$E:\Lenrn_notes\21go\2.0笔记整理\image\image-20220507205208757.png$

三、创建Server端

1.定义我们的服务，并实现RouteList方法

这里简单实现对话中一问一答的形式

// StreamService 定义我们的服务
type StreamService struct{}
// Conversations 实现Conversations方法
func (s *StreamService) Conversations(srv pb.Stream_ConversationsServer) error {
	n := 1
	for {
		req, err := srv.Recv()
		if err == io.EOF {
			return nil
		}
		if err != nil {
			return err
		}
		err = srv.Send(&pb.StreamResponse{
			Answer: "from stream server answer: the " + strconv.Itoa(n) + " question is " + req.Question,
		})
		if err != nil {
			return err
		}
		n++
		log.Printf("from stream client question: %s", req.Question)
	}
}

2.启动gRPC服务器

const (
	// Address 监听地址
	Address string = ":8000"
	// Network 网络通信协议
	Network string = "tcp"
)

func main() {
	// 监听本地端口
	listener, err := net.Listen(Network, Address)
	if err != nil {
		log.Fatalf("net.Listen err: %v", err)
	}
	log.Println(Address + " net.Listing...")
	// 新建gRPC服务器实例
	grpcServer := grpc.NewServer()
	// 在gRPC服务器注册我们的服务
	pb.RegisterStreamServer(grpcServer, &StreamService{})

	//用服务器 Serve() 方法以及我们的端口信息区实现阻塞等待，直到进程被杀死或者 Stop() 被调用
	err = grpcServer.Serve(listener)
	if err != nil {
		log.Fatalf("grpcServer.Serve err: %v", err)
	}
}

运行服务端

go run server.go
:8000 net.Listing...

四、创建Client端

1.创建调用服务端Conversations方法

// conversations 调用服务端的Conversations方法
func conversations() {
	//调用服务端的Conversations方法，获取流
	stream, err := streamClient.Conversations(context.Background())
	if err != nil {
		log.Fatalf("get conversations stream err: %v", err)
	}
	for n := 0; n < 5; n++ {
		err := stream.Send(&pb.StreamRequest{Question: "stream client rpc " + strconv.Itoa(n)})
		if err != nil {
			log.Fatalf("stream request err: %v", err)
		}
		res, err := stream.Recv()
		if err == io.EOF {
			break
		}
		if err != nil {
			log.Fatalf("Conversations get stream err: %v", err)
		}
		// 打印返回值
		log.Println(res.Answer)
	}
	//最后关闭流
	err = stream.CloseSend()
	if err != nil {
		log.Fatalf("Conversations close stream err: %v", err)
	}
}

2.启动gRPC客户端

// Address 连接地址
const Address string = ":8000"

var streamClient pb.StreamClient

func main() {
	// 连接服务器
	conn, err := grpc.Dial(Address, grpc.WithInsecure())
	if err != nil {
		log.Fatalf("net.Connect err: %v", err)
	}
	defer conn.Close()

	// 建立gRPC连接
	streamClient = pb.NewStreamClient(conn)
	conversations()
}

服务端获取到来自客户端的提问

2022/03/28 11:28:00 :8000 net.Listing...
2022/03/28 11:28:11 from stream client question: stream client rpc 0
2022/03/28 11:28:11 from stream client question: stream client rpc 1
2022/03/28 11:28:11 from stream client question: stream client rpc 2
2022/03/28 11:28:11 from stream client question: stream client rpc 3
2022/03/28 11:28:11 from stream client question: stream client rpc 4

$E:\Lenrn_notes\21go\2.0笔记整理\image\image-20220328112903297.png$

运行客户端，获取到服务端的应答

$E:\Lenrn_notes\21go\2.0笔记整理\image\image-20220328112910435.png$

2022/03/28 11:28:11 from stream server answer: the 1question is stream client rpc 0
2022/03/28 11:28:11 from stream server answer: the 2question is stream client rpc 1
2022/03/28 11:28:11 from stream server answer: the 3question is stream client rpc 2
2022/03/28 11:28:11 from stream server answer: the 4question is stream client rpc 3
2022/03/28 11:28:11 from stream server answer: the 5question is stream client rpc 4

五、总结

本篇介绍了双向流式RPC的简单使用。
参考：gRPC官方文档中文版

标签：11,stream,err,gRPC,28,流式,rpc,client,Go
来源： https://www.cnblogs.com/randysun/p/16273937.html

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

ICode9