go实现grpc四种数据流模式

⽬录

1. 什么是数据流2. grpc的四种数据流2.1 简单模式2.2 服务端数据流模式2.3 客户端数据流模式2.4 双向数据流3. 上代码3.1 代码⽬录3.2 编写stream.proto⽂件3.3 编写server⽂件3.4 编写client⽂件1. 什么是数据流

grpc中的stream，srteam顾名思义就是⼀种流，可以源源不断的推送数据，很适合传输⼀些⼤数据，或者服务端和客户端长时间数据交互，⽐如客户端可以向服务端订阅⼀个数据，服务端就可以利⽤stream，源源不断地推送数据。底层还原成socket编程

2. grpc的四种数据流

1.简单模式

2.服务端数据流模式（Server-side streaming RPC）3.客户端数据流模式（Client-side streaming RPC）4.双向数据流模式（Bidirectional streaming RPC）

2.1 简单模式

这种模式最为传统，即客户端发起⼀次请求，服务端响应⼀个数据，这和⼤家平时熟悉的RPC没有什么⼤的区别，上两篇中介绍此模式。

2.2 服务端数据流模式

这种模式是客户端发起⼀次请求，服务端返回⼀段连续的数据流。典型的例⼦是客户端向服务端发送⼀个股票代码，服务端就把该股票的实时数据源源不断的返回给客户端

2.3 客户端数据流模式

与服务端数据流模式相反，这次是客户端源源不断的向服务端发送数据流，⽽在发送结束后，由服务端返回⼀个响应。典型的例⼦是物联⽹终端向服务器报送数据。

2.4 双向数据流

顾名思义，这是客户端和服务端都可以向对⽅发送数据流，这个时候双⽅的数据可以同时互相发送，也就是可以实现实时交互。典型的例⼦是聊天机器⼈。

3. 上代码

3.1 代码⽬录

3.2 编写stream.proto⽂件

stream是常量，写在哪⼀边，哪⼀边就是数据流

syntax = \"proto3\";

option go_package = \"./;proto\";service Greeter {

// 定义⽅法，stream是常量，流模式

rpc ServerStream (StreamRequestData) returns (stream StreamResponseData); //服务端流模式，拉消息 rpc ClientStream (stream StreamRequestData) returns (StreamResponseData); //客户端流模式，推消息 rpc AllStream (stream StreamRequestData) returns (stream StreamResponseData); //双向流模式，能推能拉}

message StreamRequestData { string data = 1; //编号}

message StreamResponseData { string data = 1; //编号}

⽣成go的protobuf⽂件命令：cd到proto⽬录下

命令：protoc -I . hello.proto --go_out=plugins=grpc:.

3.3 编写server⽂件

package mainimport (

\"file_test/grpc_go_stream/proto\" \"fmt\" \"net\" \"sync\" \"time\"

\"google.golang.org/grpc\")

const port = 8082type server struct{}

func (s *server) ServerStream(req *proto.StreamRequestData, res proto.Greeter_ServerStreamServer) error { i := 0 for { i++

//业务代码

_ = res.Send(&proto.StreamResponseData{

Data: fmt.Sprintf(\"这是发给%s的数据流\ })

time.Sleep(time.Second * 1) if i > 10 { break } }

return nil}

func (s *server) ClientStream(cliStr proto.Greeter_ClientStreamServer) error { for {

//业务代码

res, err := cliStr.Recv() if err != nil {

fmt.Println(\"本次客户端流数据发送完了:\ break }

fmt.Println(\"客户端发来消息：\ }

return nil

}

func (s *server) AllStream(allStr proto.Greeter_AllStreamServer) error { wg:=sync.WaitGroup{} wg.Add(2)

//接受客户端消息的协程 go func() {

defer wg.Done() for {

//业务代码

res, err := allStr.Recv() if err != nil {

fmt.Println(\"本次客户端流数据发送完了:\ break }

fmt.Println(\"收到客户端发来消息：\ } }()

//发送消息给客户端的协程 go func() {

defer wg.Done() i := 0 for { i++

//业务代码

_ = allStr.Send(&proto.StreamResponseData{ Data: fmt.Sprintf(\"这是发给客户端的数据流\"), })

time.Sleep(time.Second * 1) if i > 10 { break } } }()

wg.Wait() return nil}

// 启动

func start() {

// 1.实例化server

g := grpc.NewServer() // 2.注册逻辑到server中

proto.RegisterGreeterServer(g, &server{}) // 3.启动server

lis, err := net.Listen(\"tcp\ if err != nil {

panic(\"监听错误:\" + err.Error()) }

err = g.Serve(lis) if err != nil {

panic(\"启动错误:\" + err.Error()) }}

func main() { start()}

3.4 编写client⽂件

package mainimport ( \"context\"

\"file_test/grpc_go_stream/proto\" \"fmt\" \"sync\" \"time\"

\"google.golang.org/grpc\")

var rpc proto.GreeterClientfunc serverStreamDemo() { //服务端流模式

res,err:=rpc.ServerStream(context.Background(),&proto.StreamRequestData{Data: \"jeff\ if err != nil {

panic(\"rpc请求错误：\"+err.Error()) } for {

data,err:=res.Recv() // if err != nil {

fmt.Println(\"客户端发送完了:\ return }

fmt.Println(\"客户端返回数据流值:\ }}

func clientStreamDemo() { //客户端流模式

cliStr, err := rpc.ClientStream(context.Background()) if err != nil {

panic(\"rpc请求错误：\" + err.Error()) } i := 0 for { i++

_ = cliStr.Send(&proto.StreamRequestData{ Data: \"jeff\ })

time.Sleep(time.Second * 1) if i > 10 { break } }}

func clientAndServerStreamDemo() { //双向流模式

allStr, _ := rpc.AllStream(context.Background()) wg := sync.WaitGroup{} wg.Add(1)

//接受服务端消息的协程 go func() {

defer wg.Done() for {

//业务代码

res, err := allStr.Recv() if err != nil {

fmt.Println(\"本次服务端流数据发送完了:\ break }

fmt.Println(\"收到服务端发来消息：\ } }()

//发送消息给服务端的协程 go func() {

defer wg.Done() i := 0 for { i++

//业务代码

_ = allStr.Send(&proto.StreamRequestData{ Data: fmt.Sprintf(\"这是发给服务端的数据流\"), })

time.Sleep(time.Second * 1) if i > 10 { break } } }()

wg.Wait()}

// 启动

func start() {

conn, err := grpc.Dial(\"127.0.0.1:8082\ if err != nil {

panic(\"rpc连接错误：\" + err.Error()) }

defer conn.Close()

rpc = proto.NewGreeterClient(conn) //初始化 serverStreamDemo() //服务端流模式 clientStreamDemo() //客户端流模式

clientAndServerStreamDemo() // 双向流模式}

func main() { start()

}

以上就是go实现grpc四种数据流模式的详细内容，更多关于go实现grpc流模式的资料请关注其它相关⽂章！