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

服务端流式RPC

2022-05-15 18:34:26 阅读：191 来源： 互联网

标签：err stream grpc 流式 pb RPC 服务端 log

服务端流式RPC

一、前言

当数据量大或者需要不断传输数据时候，我们应该使用流式RPC，它允许我们边处理边传输数据。本篇先介绍服务端流式RPC。

服务端流式RPC：客户端发送请求到服务器，拿到一个流去读取返回的消息序列。客户端读取返回的流，直到里面没有任何消息。

情景模拟：实时获取股票走势。

1.客户端要获取某原油股的实时走势，客户端发送一个请求

2.服务端实时返回该股票的走势

源码

二、新建proto文件

新建serverStream.proto文件

1.定义发送信息

// 定义发送请求信息
message SimpleRequest{
    // 定义发送的参数，采用驼峰命名方式，小写加下划线，如：student_name
    // 请求参数
    string data = 1;
}

2.定义接收信息

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

3.定义服务方法ListValue

服务端流式rpc，只要在响应数据前添加stream即可

// 定义我们的服务（可定义多个服务,每个服务可定义多个接口）
service StreamServer{
    // 服务端流式rpc，在响应数据前添加stream
    rpc ListValue(SimpleRequest)returns(stream StreamResponse){};
}

4.编译proto文件

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


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

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


// 定义我们的服务（可定义多个服务,每个服务可定义多个接口）
service StreamServer{
  // 服务端流式rpc，在响应数据前添加stream
  rpc ListValue(SimpleRequest) returns (stream StreamResponse){};
}

// 定义发送请求信息
message SimpleRequest{
  // 定义发送的参数，采用驼峰命名方式，小写加下划线，如：student_name
  // 请求参数
  string data = 1;
}

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

进入go-grpc-example项目所在目录，运行指令:

go-grpc-example> protoc -I ./ --go_out=plugins=grpc:.\03serverStream\proto\ .\03serverStream\proto\serverStream.proto

三、创建Server端

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

package main

import (
	pb "go-grpc-example/03serverStream/proto"
	"log"
	"net"
	"strconv"
	"time"

	"google.golang.org/grpc"
)

/*
@author RandySun
@create 2022-03-27-20:41
*/

// StreamService 定义流服务
type StreamService struct {
}

// ListValue 实现ListValue方法
func (s *StreamService) ListValue(req *pb.SimpleRequest, srv pb.StreamServer_ListValueServer) error {
	for n := 0; n < 5; n++ {
		// 向流中发送消息,默认每次send送消息最大的长度为 math.MaxInt32 bytes
		err := srv.Send(&pb.StreamResponse{
			StreamValue: req.Data + strconv.Itoa(n),
		})
		if err != nil {
			return err
		}

		log.Println("发送数据n:", strconv.Itoa(n))
		time.Sleep(1 * time.Second)
	}
	return nil
}

初学者可能觉得比较迷惑，ListValue的参数和返回值是怎样确定的。其实这些都是编译proto时生成的.pb.go文件中有定义，我们只需要实现就可以了。

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

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服务器实例
	// 默认单次接收最大消息长度为`1024*1024*4`bytes(4M)，单次发送消息最大长度为`math.MaxInt32`bytes
	// grpcServer := grpc.NewServer(grpc.MaxRecvMsgSize(1024*1024*4), grpc.MaxSendMsgSize(math.MaxInt32))
	grpcServer := grpc.NewServer()
	pb.RegisterStreamServerServer(grpcServer, &StreamService{})

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

}

运行服务端

go run server.go
2022/03/27 21:20:09 :8000 net.Listing...

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

四、创建Client端

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

package main

import (
	"context"
	pb "grpc/03serverStream/proto"
	"io"
	"log"

	"google.golang.org/grpc/credentials/insecure"

	"google.golang.org/grpc"
)

/*
@author RandySun
@create 2022-03-27-21:07
*/

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

var grpcClient pb.StreamServerClient

func listValue() {
	// 创建发送结构体
	req := pb.SimpleRequest{
		Data: "stream server grpc",
	}

	// 调用ListValue 方法 路由
	stream, err := grpcClient.ListValue(context.Background(), &req)
	if err != nil {
		log.Fatalf("Call ListStr err: %v", err)
	}
	for {
		// Recv() 方法接收服务端消息,默认每类Recv() 最大消息长度 1024*1024*4 bytes(4M)
		res, err := stream.Recv()
		if err == io.EOF {
			break
		}
		if err != nil {
			log.Fatalf("ListStr get stream err: %v", err)
		}
		// 打印返回值
		log.Println(res.StreamValue)
	}
}

2.启动gRPC客户端

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

	// 建立gRPC连接
	grpcClient = pb.NewStreamServerClient(conn)
	listValue()
}

运行客户端

go run client.go
stream server grpc 0
stream server grpc 1
stream server grpc 2
stream server grpc 3
stream server grpc 4

客户端不断从服务端获取数据

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

五、客户端暂停/继续接收数据

假如服务端不停发送数据，类似获取股票走势实时数据，客户端能自己停止获取数据吗？

答案：可以的

1.我们把服务端的ListValue方法稍微修改

package main

import (
	pb "grpc/03serverStream/proto"
	"log"
	"net"
	"strconv"
	"time"

	"google.golang.org/grpc"
)

/*
@author RandySun
@create 2022-03-27-20:41
*/

// StreamService 定义流服务
type StreamService struct {
}

// ListValue 实现ListValue方法
func (s *StreamService) ListValue(req *pb.SimpleRequest, srv pb.StreamServer_ListValueServer) error {
	for n := 0; n < 15; n++ {
		// 向流中发送消息,默认每次send送消息最大的长度为 math.MaxInt32 bytes
		err := srv.Send(&pb.StreamResponse{
			StreamValue: req.Data + strconv.Itoa(n),
		})
		if err != nil {
			return err
		}

		log.Println("发送数据n:", strconv.Itoa(n))
		time.Sleep(1 * time.Second)
	}
	return nil
}

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服务器实例
	// 默认单次接收最大消息长度为`1024*1024*4`bytes(4M)，单次发送消息最大长度为`math.MaxInt32`bytes
	// grpcServer := grpc.NewServer(grpc.MaxRecvMsgSize(1024*1024*4), grpc.MaxSendMsgSize(math.MaxInt32))
	grpcServer := grpc.NewServer()
	pb.RegisterStreamServerServer(grpcServer, &StreamService{})

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

}

2.再把客户端调用ListValue方法的实现稍作修改，就可以得到结果了

package main

import (
	"context"
	pb "grpc/03serverStream/proto"
	"io"
	"log"

	"google.golang.org/grpc/credentials/insecure"

	"google.golang.org/grpc"
)

/*
@author RandySun
@create 2022-03-27-21:07
*/

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

var grpcClient pb.StreamServerClient

func listValue() {
	// 创建发送结构体
	req := pb.SimpleRequest{
		Data: "stream server grpc",
	}

	// 调用ListValue 方法 路由
	// 同时传入了一个 context.Context ，在有需要时可以让我们改变RPC的行为，比如超时/取消一个正在运行的RPC
	stream, err := grpcClient.ListValue(context.Background(), &req)
	if err != nil {
		log.Fatalf("Call ListStr err: %v", err)
	}
	for {
		// Recv() 方法接收服务端消息,默认每类Recv() 最大消息长度 1024*1024*4 bytes(4M)
		res, err := stream.Recv()
		if err == io.EOF {
			break
		}
		if err != nil {
			log.Fatalf("ListStr get stream err: %v", err)
		}
		// 打印返回值
		log.Println(res.StreamValue)

		// 模拟终止服务端发送消息
		break // 退出接收数据
	}

	//可以使用CloseSend()关闭stream，这样服务端就不会继续产生流消息
	//调用CloseSend()后，若继续调用Recv()，会重新激活stream，接着之前结果获取消息
	stream.CloseSend()
	// 继续接收
	res, err := stream.Recv()

	if err == io.EOF {
		log.Fatalf("ListStr get stream err: %v", err)
	}
	if err != nil {
		log.Fatalf("ListStr get stream err: %v", err)
	}
	// 打印返回值
	log.Println(res.StreamValue)
}

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

	// 建立gRPC连接
	grpcClient = pb.NewStreamServerClient(conn)
	listValue()
}

只需要调用CloseSend()方法，就可以关闭服务端的stream，让它停止发送数据。值得注意的是，调用CloseSend()后，若继续调用Recv()，会重新激活stream，接着当前的结果继续获取消息。

这能完美解决客户端暂停->继续获取数据的操作。

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

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

六、总结

本篇介绍了服务端流式RPC的简单使用，客户端发起一个请求，服务端不停返回数据，直到服务端停止发送数据或客户端主动停止接收数据为止。参考：gRPC官方文档中文版

标签：err,stream,grpc,流式,pb,RPC,服务端,log
来源： https://www.cnblogs.com/randysun/p/16273932.html

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

ICode9

服务端流式RPC

服务端流式RPC

一、前言

情景模拟：实时获取股票走势。

二、新建proto文件

三、创建Server端

四、创建Client端

五、客户端暂停/继续接收数据

六、总结