什么是数据流

grpc中的stream,srteam顾名思义就是一种流,可以源源不断的推送数据,很适合传输一些大数据,或者服务端和客户端长时间数据交互,比如客户端可以向服务端订阅一个数据,服务端就可以利用stream,源源不断地推送数据。

grpc的四种数据流

  • 1.简单模式
  • 2.服务端数据流模式(Server-side streaming RPC)
  • 3.客户端数据流模式(Client-side streaming RPC)
  • 4.双向数据流模式(Bidirectional streaming RPC)

简单模式

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

服务端数据流模式

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

客户端数据流模式

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

双向数据流

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

上代码

代码目录

编写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:.

编写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()
}

编写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()
}

转自:https://www.yixuebiancheng.com/article/94795.html