ICode9

精准搜索请尝试: 精确搜索
首页 > 其他分享> 文章详细

Go-客户端流式gRPC

2022-05-15 18:34:14  阅读:169  来源: 互联网

标签:err stream gRPC 流式 log Go RouteList 服务端 客户端


Go gRPC教程-客户端流式gRPC

一、前言

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

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

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

源码

二、新建proto文件

新建clientStream.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文件

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


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

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

// 定义我们的服务(可定义多个服务,每个服务可定义多个接口)
service StreamClient{
  // 客户端流式rpc,在请求的参数前添加stream
  rpc RouteList (stream StreamRequest) returns (SimpleResponse){};
}

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

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

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

go-grpc-example> protoc -I ./ --go_out=plugins=grpc:.\04clientStream\proto\  .\04clientStream\proto\clientStream.proto

三、创建Server端

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

package main

import (
	pb "grpc/04clientStream/proto"
	"io"
	"log"
	"net"

	"google.golang.org/grpc"
)

/*
@author RandySun
@create 2022-03-27-22:22
*/

// 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{
				Code:  200,
				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)
	}
}

运行服务端

cd .\04clientStream\server\
go build .\service.go
.\service.exe
2022/03/28 09:39:19 :8000net.Listing...

image-20220328094100994

四、创建Client端

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

package main

import (
	"context"
	pb "grpc/04clientStream/proto"
	"log"
	"strconv"

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

/*
@author RandySun
@create 2022-03-27-22:22
*/

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

var streamClient pb.StreamClientClient

// 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客户端

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连接
	streamClient = pb.NewStreamClientClient(conn)
	routeList()
}

运行客户端

cd .\04clientStream\cleint\
go build .\client.go
.\client.exe
2022/03/28 09:39:28 code:200  value:"ok"

image-20220328100541651

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

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错误。

image-20220328101300106

image-20220328101311336

六、总结

本篇介绍了客户端流式RPC的简单使用,下篇将介绍双向流式RPC
参考:gRPC官方文档中文版

标签:err,stream,gRPC,流式,log,Go,RouteList,服务端,客户端
来源: https://www.cnblogs.com/randysun/p/16273935.html

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

专注分享技术,共同学习,共同进步。侵权联系[81616952@qq.com]

Copyright (C)ICode9.com, All Rights Reserved.

ICode9版权所有