Go gRPC服务客户端流式RPC教程

目录
  • 前言
  • 情景模拟:客户端大量数据上传到服务端
    • 新建proto文件
    • 创建Server端
    • 创建Client端
  • 思考
  • 总结

前言

上一篇介绍了服务端流式RPC,客户端发送请求到服务器,拿到一个流去读取返回的消息序列。 客户端读取返回的流的数据。本篇将介绍客户端流式RPC。

客户端流式RPC:与服务端流式RPC相反,客户端不断的向服务端发送数据流,而在发送结束后,由服务端返回一个响应。

情景模拟:客户端大量数据上传到服务端

新建proto文件

新建client_stream.proto文件

1.定义发送信息

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

2.定义接收信息

// 定义响应信息
message SimpleResponse{
    //响应码
    int32 code = 1;
    //响应值
    string value = 2;
}

3.定义服务方法RouteList

客户端流式rpc,只要在请求的参数前添加stream即可

service StreamClient{
    // 客户端流式rpc,在请求的参数前添加stream
    rpc RouteList (stream StreamRequest) returns (SimpleResponse){};
}

4.编译proto文件

进入client_stream.proto所在目录,运行指令:

protoc --go_out=plugins=grpc:./ ./client_stream.proto

创建Server端

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

// SimpleService 定义我们的服务
type SimpleService struct{}
// RouteList 实现RouteList方法
func (s *SimpleService) RouteList(srv pb.StreamClient_RouteListServer) error {
	for {
		//从流中获取消息
		res, err := srv.Recv()
		if err == io.EOF {
			//发送结果,并关闭
			return srv.SendAndClose(&pb.SimpleResponse{Value: "ok"})
		}
		if err != nil {
			return err
		}
		log.Println(res.StreamData)
	}
}

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.RegisterStreamClientServer(grpcServer, &SimpleService{})

	//用服务器 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.创建调用服务端RouteList方法

// routeList 调用服务端RouteList方法
func routeList() {
	//调用服务端RouteList方法,获流
	stream, err := streamClient.RouteList(context.Background())
	if err != nil {
		log.Fatalf("Upload list err: %v", err)
	}
	for n := 0; n < 5; n++ {
		//向流中发送消息
		err := stream.Send(&pb.StreamRequest{StreamData: "stream client rpc " + strconv.Itoa(n)})
		if err != nil {
			log.Fatalf("stream request err: %v", err)
		}
	}
	//关闭流并获取返回的消息
	res, err := stream.CloseAndRecv()
	if err != nil {
		log.Fatalf("RouteList get response err: %v", err)
	}
	log.Println(res)
}

2.启动gRPC客户端

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

var streamClient pb.StreamClientClient

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.NewStreamClientClient(conn)
	routeList()
}

运行客户端

go run client.go
code:200 value:"hello grpc"
value:"ok"

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

stream client rpc 0
stream client rpc 1
stream client rpc 2
stream client rpc 3
stream client rpc 4

思考

服务端在没有接受完消息时候能主动停止接收数据吗(很少有这种场景)?

答案:可以的,但是客户端代码需要注意EOF判断

1.我们把服务端的RouteList方法实现稍微修改,当接收到一条数据后马上调用SendAndClose()关闭stream.

// RouteList 实现RouteList方法
func (s *SimpleService) RouteList(srv pb.StreamClient_RouteListServer) error {
	for {
		//从流中获取消息
		res, err := srv.Recv()
		if err == io.EOF {
			//发送结果,并关闭
			return srv.SendAndClose(&pb.SimpleResponse{Value: "ok"})
		}
		if err != nil {
			return err
		}
		log.Println(res.StreamData)
		return srv.SendAndClose(&pb.SimpleResponse{Value: "ok"})
	}
}

2.再把客户端调用RouteList方法的实现稍作修改

// routeList 调用服务端RouteList方法
func routeList() {
	//调用服务端RouteList方法,获流
	stream, err := streamClient.RouteList(context.Background())
	if err != nil {
		log.Fatalf("Upload list err: %v", err)
	}
	for n := 0; n < 5; n++ {
		//向流中发送消息
		err := stream.Send(&pb.StreamRequest{StreamData: "stream client rpc " + strconv.Itoa(n)})
		//发送也要检测EOF,当服务端在消息没接收完前主动调用SendAndClose()关闭stream,此时客户端还执行Send(),则会返回EOF错误,所以这里需要加上io.EOF判断
		if err == io.EOF {
			break
		}
		if err != nil {
			log.Fatalf("stream request err: %v", err)
		}
	}
	//关闭流并获取返回的消息
	res, err := stream.CloseAndRecv()
	if err != nil {
		log.Fatalf("RouteList get response err: %v", err)
	}
	log.Println(res)
}

客户端Send()需要检测err是否为EOF,因为当服务端在消息没接收完前主动调用SendAndClose()关闭stream,若此时客户端继续执行Send(),则会返回EOF错误。

总结

本篇介绍了客户端流式RPC的简单使用,下篇将介绍双向流式RPC。

教程源码地址:https://github.com/Bingjian-Zhu/go-grpc-example

参考:gRPC官方文档中文版

以上就是Go gRPC服务客户端流式RPC教程的详细内容,更多关于Go gRPC客户端流式RPC的资料请关注我们其它相关文章!

(0)

相关推荐

  • 关于go-micro与其它gRPC框架之间的通信问题及解决方法

    目录 客户端改造 服务端改造 运行效果 在之前的文章中分别介绍了使用gRPC官方插件和go-micro插件开发gRPC应用程序的方式,都能正常走通.不过当两者混合使用的时候,互相访问就成了问题.比如使用go-micro插件生成的gRPC客户端访问基于gRPC官方插件创建的服务端时就会出现如下错误: {"id":"go.micro.client","code":501,"status":"Not Implemented

  • Go gRPC服务端流式RPC教程示例

    目录 前言 情景模拟:实时获取股票走势 新建proto文件 创建Server端 创建Client端 思考 总结 前言 上一篇介绍了简单模式RPC,当数据量大或者需要不断传输数据时候,我们应该使用流式RPC,它允许我们边处理边传输数据.本篇先介绍服务端流式RPC. 服务端流式RPC:客户端发送请求到服务器,拿到一个流去读取返回的消息序列. 客户端读取返回的流,直到里面没有任何消息. 情景模拟:实时获取股票走势 1.客户端要获取某原油股的实时走势,客户端发送一个请求 2.服务端实时返回该股票的走势

  • Go gRPC超时控制Deadlines用法详解

    目录 Deadlines 为什么要设置 gRPC Client Server 验证 总结 Deadlines 今天将介绍 gRPC Deadlines 的用法,这一个必备技巧.内容也比较简单. Deadlines 意指截止时间,在 gRPC 中强调 TL;DR(Too long, Don't read)并建议始终设定截止日期,为什么呢? 为什么要设置 当未设置 Deadlines 时,将采用默认的 DEADLINE_EXCEEDED(这个时间非常大) 如果产生了阻塞等待,就会造成大量正在进行的请

  • Go gRPC教程实现Simple RPC

    目录 前言 新建proto文件 创建Server端 创建Client端 总结 前言 gRPC主要有4种请求和响应模式,分别是简单模式(Simple RPC).服务端流式(Server-side streaming RPC).客户端流式(Client-side streaming RPC).和双向流式(Bidirectional streaming RPC). 简单模式(Simple RPC):客户端发起请求并等待服务端响应. 服务端流式(Server-side streaming RPC):客户

  • Go gRPC环境安装教程示例详解

    目录 前言 安装 protobuf 安装相关包 创建并编译proto文件 VSCode-proto3插件介绍 前言 gRPC 是一个高性能.开源和通用的 RPC 框架,面向移动和 HTTP/2 设计,带来诸如双向流.流控.头部压缩.单 TCP 连接上的多复用请求等特.这些特性使得其在移动设备上表现更好,更省电和节省空间占用. 在 gRPC 里客户端应用可以像调用本地对象一样直接调用另一台不同的机器上服务端应用的方法,使得您能够更容易地创建分布式应用和服务. gRPC 默认使用 protocol

  • Go语言程序开发gRPC服务

    目录 前言 介绍 入门 proto server client 流方式 proto server client 验证器 proto Token 认证 认证函数: 拦截器: 初始化: 实现接口: 连接: 单向证书认证 生成证书 gRPC 代码 双向证书认证 生成带 SAN 的证书 gRPC 代码 Python 客户端 总结 前言 gRPC 这项技术真是太棒了,接口约束严格,性能还高,在 k8s 和很多微服务框架中都有应用. 作为一名程序员,学就对了. 之前用 Python 写过一些 gRPC 服务

  • Go gRPC服务客户端流式RPC教程

    目录 前言 情景模拟:客户端大量数据上传到服务端 新建proto文件 创建Server端 创建Client端 思考 总结 前言 上一篇介绍了服务端流式RPC,客户端发送请求到服务器,拿到一个流去读取返回的消息序列. 客户端读取返回的流的数据.本篇将介绍客户端流式RPC. 客户端流式RPC:与服务端流式RPC相反,客户端不断的向服务端发送数据流,而在发送结束后,由服务端返回一个响应. 情景模拟:客户端大量数据上传到服务端 新建proto文件 新建client_stream.proto文件 1.定义

  • Go gRPC服务双向流式RPC教程

    目录 前言 情景模拟 新建proto文件 创建Server端 创建Client端 总结 前言 上一篇介绍了客户端流式RPC,客户端不断的向服务端发送数据流,在发送结束或流关闭后,由服务端返回一个响应.本篇将介绍双向流式RPC. 双向流式RPC:客户端和服务端双方使用读写流去发送一个消息序列,两个流独立操作,双方可以同时发送和同时接收. 情景模拟 双方对话(可以一问一答.一问多答.多问一答,形式灵活). 新建proto文件 新建both_stream.proto文件 1.定义发送信息 // 定义流

  • Go gRPC服务proto数据验证进阶教程

    前言 上篇介绍了go-grpc-middleware的grpc_zap.grpc_auth和grpc_recovery使用,本篇将介绍grpc_validator,它可以对gRPC数据的输入和输出进行验证. 创建proto文件,添加验证规则 这里使用第三方插件go-proto-validators自动生成验证规则. go get github.com/mwitkow/go-proto-validators 1.新建simple.proto文件 syntax = "proto3"; pa

  • Java8中的Stream流式操作教程之王者归来

    前言 相对于Java8之前的Java的相关操作简直是天差地别,Java8 的流式操作的出现,也很大程度上改变了开发者对于Java的繁琐的操作的印象,从此,Java也走向了函数式编程的道路! 1 流的创建 1.1 流的创建方法 既然需要聊聊流的操作,那么,首先还是先看看怎么创建流. 创建流的方法有三种,分别是:Stream.of().Stream.iterate().Stream.generate(),然后,分别看一下这三个方法的声明. static <T> Stream<T> of

  • Go gRPC服务进阶middleware使用教程

    目录 前言 go-grpc-middleware简介 grpc_zap日志记录 grpc_auth认证 grpc_recovery恢复 总结 前言 之前介绍了gRPC中TLS认证和自定义方法认证,最后还简单介绍了gRPC拦截器的使用.gRPC自身只能设置一个拦截器,所有逻辑都写一起会比较乱.本篇简单介绍go-grpc-middleware的使用,包括grpc_zap.grpc_auth和grpc_recovery. go-grpc-middleware简介 go-grpc-middleware封

  • 图析ASP.NET Core引入gRPC服务模板

    早就听说ASP.NET Core 3.0中引入了gRPC的服务模板,正好趁着家里电脑刚做了新系统,然后装了VS2019的功夫来体验一把.同时记录体验的过程.如果你也想按照本文的步骤体验的话,那你得先安装.NET Core3.0预览版的SDK.至于开发工具我用的时VS2019,当然你也可以使用VS Code进行. gRPC的简单介绍 gRPC 是一种与语言无关的高性能远程过程调用 (RPC) 框架. 有关 gRPC 基础知识的详细信息,请参阅 gRPC 文档页. gRPC 的主要优点是: 现代高性

  • WIN7下ORACLE10g服务端和客户端的安装图文教程

    win7下安装oracle10g服务端和客户端方法如下所示: 如何卸载Oracle 10g 软件环境: 1.Windows XP + Oracle 10g 2.Oracle安装路径为:d:\Oracle 实现方法: 1.开始->设置->控制面板->管理工具->服务停止所有Oracle服务; 2.开始->程序->Oracle – OraDb10g_home1>Oracle Installation Products-> Universal Installer

  • Win2003下cwRsyncServer服务端与cwRsync客户端数据同步实例教程

    说明: cwRsyncServer服务端   IP:192.168.21.128 cwRsync客户端   IP:192.168.21.129 实现目的: 把服务端D:\data目录中的数据通过任务计划定期同步到客户端D:\data目录中 附件: cwRsyncServer下载地址与cwRsync下载地址:http://s.jb51.net 具体操作: 一.在服务端安装cwRsyncServer 解压cwRsyncServer_4.0.3_Installer.zip,双击cwRsyncServe

随机推荐