go实现grpc四种数据流模式

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 main
import (
	"file_test/grpc_go_stream/proto"
	"fmt"
	"net"
	"sync"
	"time"
	"google.golang.org/grpc"
)
const port = 8082
type 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的数据流", req.Data),
		})
		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("本次客户端流数据发送完了:",err)
			break
		}
		fmt.Println("客户端发来消息：",res.Data)
	}
	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("本次客户端流数据发送完了:",err)
				break
			}
			fmt.Println("收到客户端发来消息：",res.Data)
		}
	}()
	//发送消息给客户端的协程
	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", "127.0.0.1:8082")
	if err != nil {
		panic("监听错误:" + err.Error())
	}
	err = g.Serve(lis)
	if err != nil {
		panic("启动错误:" + err.Error())
	}
}
func main() {
	start()
}

3.4 编写client文件

package main
import (
	"context"
	"file_test/grpc_go_stream/proto"
	"fmt"
	"sync"
	"time"

	"google.golang.org/grpc"
)
var rpc proto.GreeterClient
func 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("客户端发送完了:",err)
			return
		}
		fmt.Println("客户端返回数据流值:",data.Data)
	}
}
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("本次服务端流数据发送完了:", err)
				break
			}
			fmt.Println("收到服务端发来消息：", res.Data)
		}
	}()
	//发送消息给服务端的协程
	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", grpc.WithInsecure())
	if err != nil {
		panic("rpc连接错误：" + err.Error())
	}
	defer conn.Close()
	rpc = proto.NewGreeterClient(conn) //初始化
	serverStreamDemo() //服务端流模式
	clientStreamDemo()  //客户端流模式
	clientAndServerStreamDemo() // 双向流模式
}
func main() {
	start()
}

频道导航