golang工程——grpc一元请求与流请求

唉，难顶，像块砖一样到处搬。又跑去golang技术栈的项目俩月
本篇博客多有参考，记录一下近期学习

rpc、grpc原理

grpc 是一项进程间通信技术，可以用来连接、调用、操作和调试构建分布式程序，调用过程如同调用函数一样，整个过程操作起来很简单，就像调用本地方法一样。与许多rpc 系统一样，grpc 是定义服务的思想，服务器需要实现此接口并运行grpc 来处理客户端调用。

在这里插入图片描述

grpc 默认使用 Protocol Buffers 作为 IDL 定义服务接口。 Protocol Buffers 是中立的、与平台无关、实现结构化数据序列化的可扩展机制，虽然也可以使用json，但是 Protocol Buffers 是二进制编解码，所以编解码性能比Json好。服务接口定义在 proto 文件中指定，也就是在扩展名为 “. proto” 的普通文本文件中。我们要按照普通的 Protocol Buffers 范式来定义grpc服务，并将方法参数和返回类型指定为 Protocol buffers 的格式，也因为服务定义是Protocol Buffers 规范的扩展，所以可以借助特殊的 gRPC 插件来根据 proto 文件生成代码。

grpc常用通信模式

一元rpc模式

类似http 协议一问一答

服务端数据流模式

这种流模式可以理解为，服务器向客户端源源不断的发送数据流，应用场景很多，比如游戏中定时任务或者其他事件造成玩家数据变化需要将数据推送给客户端。

一元rpc模式下，grpc服务器端和 grpc 客户端在通信时始终只有一个请求和一个响应。在服务器端流rpc 模式下，服务端接收到一个请求后发送多个响应组成的序列，在服务器发送所有响应消息完毕后，发送trailer 元数据给客户端，标识流结束。

客户端接收流数据需要循环接收，直到出现io.EOF,代表服务器发送流数据已经完毕，后面会写grpc实现这个功能的原理

客户端数据流模式

客户端可以将数据源源不断发送给服务器，跟服务端流相反，客户端会发送多条响应，服务器发送一条响应，但是服务器不必等到发送完所有消息才响应。可以发送一条或几条消息就开始响应。

下面来看一个例子：物联网硬件将本地的缓存信息上传到服务器

服务器需要以流的方式去接收数据，当客户端关闭流的时候会返回io.EOF，这时候我们可以做响应。

双向数据流模式

双方都可以将数据源源不断发给对方。简单来说就是上面客户端流和服务器流的一个整合。

下面来看一个例子：玩家连续进行了多次战斗请求，服务器将操作结果响应给玩家

服务器读到客户端流关闭时返回nil，标记服务器流结束。
与之前客户端流模式不一样，客户端流模式是直接sendAndClose()。下面这样读到一半数据返回nil，也标识服务器流数据结束，只是可能会丢数据
启动一个协程异步接收数据，官方有说明，一个goroutine 读，一个goroutine 写是不会有并发问题的。
stream.CloseSend()代表关闭客户端流，标记客户端流已经结束
客户端需要通过定义rpc方法c.DataUpload(ctx)打开流，然后通过send 发送请求，发送完后调用CloseAndRecv关闭流等待消息响应，并处理错误，这里为了demo 演示 ,err就直接panic，实际情况可能更加复杂，对错误处理也很多种方式。

四种通信模式例子

例子

echo.proto

syntax = "proto3";
option go_package = "grpc/echo";
import  "google/protobuf/timestamp.proto";
package grpc.echo;

message EchoRequest {
  string message = 1;
  bytes bytes = 2;
  int32 length = 3;
  google.protobuf.Timestamp time = 4;

}
message EchoResponse {
  string message = 1;
  bytes bytes = 2;
  int32 length = 3;
  google.protobuf.Timestamp time = 4;
}

service Echo {
  //一元请求
  rpc UnaryEcho(EchoRequest) returns(EchoResponse) {}
  //服务端流
  rpc ServerStreamingEcho(EchoRequest) returns(stream EchoResponse){}
  //客户端流
  rpc ClientStreamingEcho(stream EchoRequest) returns( EchoResponse){}
  //双向流
  rpc BidirectionalStreamingEcho(stream EchoRequest) returns(stream EchoResponse){}
}

服务端
server.go

package server

import (
    "context"
    "fmt"
    "google.golang.org/protobuf/types/known/timestamppb"
    "grpc/echo"
    "io"
    "log"
    "os"
    "strconv"
    "sync"
    "time"
)


type EchoServer struct {
    echo.UnimplementedEchoServer
}
func (EchoServer) UnaryEcho(ctx context.Context, in *echo.EchoRequest) (*echo.EchoResponse, error) {
    fmt.Printf("server recv :%v\n", in.Message)

    return &echo.EchoResponse{
        Message: "server send message",
    }, nil
}
//服务端流，发送文件
func (EchoServer) ServerStreamingEcho(in *echo.EchoRequest, stream echo.Echo_ServerStreamingEchoServer) error {
    fmt.Printf("server recv :%v\n", in.Message)
    filePath := "echo-server-practice/files/server.jpg"

    file, err := os.Open(filePath)
    if err != nil {
        log.Fatal(err.Error())
    }
    defer file.Close()

    buf := make([]byte, 1024)

    for {
        n, err := file.Read(buf)
        if err == io.EOF {
            break
        }
        if err != nil {
            return err
        }

        stream.Send(&echo.EchoResponse{
            Message: "server sending files",
            Bytes: buf[:n],
            Time: timestamppb.New(time.Now()),
            Length: int32(n),
        })
    }
    //服务端流结束 return nil
    return nil
}
//客户端流，接收文件
func (EchoServer) ClientStreamingEcho(stream echo.Echo_ClientStreamingEchoServer) error {

    filePath := "echo-server-practice/files/" + strconv.FormatInt(time.Now().UnixMilli(), 10) + ".jpg"

    file, err := os.OpenFile(filePath, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0644)

    if err != nil {
        log.Fatal(err)
        return err
    }
    defer file.Close()

    for {
        res, err := stream.Recv()
        if err == io.EOF {
            break
        }

        if err != nil {
            log.Println(err)
            return err
        }
        file.Write(res.Bytes[:res.Length])
        fmt.Printf("server recv :%v\n", res.Message)

    }
    err = stream.SendAndClose(
        &echo.EchoResponse{
            Message: "server send complete",
        })
    return err
}
//双向流，互发文件
func (EchoServer) BidirectionalStreamingEcho(stream echo.Echo_BidirectionalStreamingEchoServer) error {

    wg := &sync.WaitGroup{}

    wg.Add(1)
    go func () {
        defer wg.Done()
        //接收客户端文件

        filePath := "echo-server-practice/files/" + strconv.FormatInt(time.Now().UnixMilli(), 10) + ".jpg"

        file, err := os.OpenFile(filePath, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0644)

        if err != nil {
            log.Fatal(err)
        }
        defer file.Close()

        for {
            res, err := stream.Recv()
            if err == io.EOF {
                break
            }

            if err != nil {
                log.Println(err)
                return
            }
            file.Write(res.Bytes[:res.Length])
            fmt.Printf("server recv :%v\n", res.Message)

        }
    }()

    wg.Add(1)

    go func() {
        defer wg.Done()
        // 发文件
        filePath := "echo-server-practice/files/server.jpg"

        file, err := os.Open(filePath)
        if err != nil {
            log.Fatal(err.Error())
        }
        defer file.Close()

        buf := make([]byte, 1024)

        for {
            n, err := file.Read(buf)
            if err == io.EOF {
                break
            }
            if err != nil {
                log.Fatal(err)
                return
            }

            stream.Send(&echo.EchoResponse{
                Message: "server sending files",
                Bytes: buf[:n],
                Time: timestamppb.New(time.Now()),
                Length: int32(n),
            })
        }
    } ()
    wg.Wait()
    // 服务端流关闭
    return nil
}

main.go

package main

import (
    "flag"
    "fmt"
    "google.golang.org/grpc"
    "grpc/echo"
    "grpc/echo-server-practice/server"
    "log"
    "net"
)

var (
    port = flag.Int("port", 50053, "port")
)

func main() {
    flag.Parse()

    lis, err := net.Listen("tcp", fmt.Sprintf(":%d", *port))

    if err != nil {
        log.Fatal(err)
    }

    // grpc server
    s := grpc.NewServer()
    echo.RegisterEchoServer(s, &server.EchoServer{})

    log.Printf("server listening at : %v\n", lis.Addr())

    if err := s.Serve(lis); err !=nil {
        log.Fatal(err)
    }

}

客户端
client.go

package client

import (
    "fmt"
    "golang.org/x/net/context"
    "google.golang.org/protobuf/types/known/timestamppb"
    "grpc/echo"
    "io"
    "log"
    "os"
    "strconv"
    "sync"
    "time"
)
//一元请求
func CallUnary(client echo.EchoClient) {
    ctx, cancel := context.WithTimeout(context.Background(), time.Second)

    defer cancel()
    in := &echo.EchoRequest{
        Message: "client send message",
        Time: timestamppb.New(time.Now()),
    }

    res, err := client.UnaryEcho(ctx, in)
    if err != nil {
        log.Fatal(err)
    }

    fmt.Printf("client recv: %v", res.Message)

}
//服务端流
func CallServerStream(client echo.EchoClient) {
    ctx, cancel := context.WithTimeout(context.Background(), time.Second)

    defer cancel()
    in := &echo.EchoRequest{
        Message: "client send message",
        Time: timestamppb.New(time.Now()),
    }

    stream, err := client.ServerStreamingEcho(ctx, in)

    if err != nil {
        log.Fatal(err)
    }

    filename := "echo-client-practice/files/" + strconv.FormatInt(time.Now().UnixMilli(), 10) + ".jpg"

    file, err := os.OpenFile(filename, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0644)

    if err != nil {
        log.Fatal(err)
        return
    }
    defer file.Close()

    for {
        res, err := stream.Recv()
        if err == io.EOF {
            break
        }

        if err != nil {
            log.Println(err)
            break
        }

        file.Write(res.Bytes[:res.Length])
        fmt.Printf("client recv %v\n", res.Message)
    }
    stream.CloseSend()
}
//客户端流
func CallClientSteam(client echo.EchoClient) {
    // 客户端流
    filePath := "echo-client-practice/files/client.jpg"

    file, err := os.Open(filePath)
    if err != nil {
        log.Fatal(err)
    }

    defer file.Close()

    ctx, cancel := context.WithTimeout(context.Background(), time.Second)

    defer cancel()

    stream, err := client.ClientStreamingEcho(ctx)

    if err != nil {
        log.Fatal(err)
    }

    buf := make([]byte, 1024)

    for {
        n, err := file.Read(buf)

        if err == io.EOF {break}
        if err != nil {
            log.Fatal(err)
        }

        stream.Send(&echo.EchoRequest{
            Message: "client sending file",
            Bytes: buf,
            Length: int32(n),
            Time: timestamppb.New(time.Now()),
        })

    }

    res, err := stream.CloseAndRecv()
    if err != nil {
        log.Fatal(err)
    }

    fmt.Printf("client recv : %v\n", res.Message)
}
// 双向流发文件
func CallBidirectional(client echo.EchoClient) {

    // 发送文件
    ctx, cancel := context.WithTimeout(context.Background(), 10 * time.Second)

    defer cancel()

    stream, err := client.BidirectionalStreamingEcho(ctx)
    if err != nil {
        log.Fatal(err)
    }

    wg := &sync.WaitGroup{}
    wg.Add(1)

    go func() {
        defer wg.Done()
        filePath := "echo-client-practice/files/client.jpg"

        file, err := os.Open(filePath)
        if err != nil {
            log.Fatal(err)
        }

        defer file.Close()

        buf := make([]byte, 1024)

        for {
            n, err := file.Read(buf)

            if err == io.EOF {break}
            if err != nil {
                log.Fatal(err)
            }

            stream.Send(&echo.EchoRequest{
                Message: "client sending file",
                Bytes: buf,
                Length: int32(n),
                Time: timestamppb.New(time.Now()),
            })
        }
        stream.CloseSend()
    }()

    wg.Add(1)

    go func() {
        //接收文件
        defer wg.Done()
        filename := "echo-client-practice/files/" + strconv.FormatInt(time.Now().UnixMilli(), 10) + ".jpg"

        file, err := os.OpenFile(filename, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0644)

        if err != nil {
            log.Fatal(err)
            return
        }
        defer file.Close()

        for {
            res, err := stream.Recv()
            if err == io.EOF {
                break
            }

            if err != nil {
                log.Println(err)
                break
            }

            file.Write(res.Bytes[:res.Length])
            fmt.Printf("client recv %v\n", res.Message)
        }
    }()
    wg.Wait()
}

main.go

package main

import (
    "flag"
    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials/insecure"
    "grpc/echo"
    "grpc/echo-client-practice/client"
    "log"
)

var (
    addr = flag.String("host", "localhost:50053", "")
)




func main() {
    flag.Parse()

    conn, err := grpc.Dial(*addr, grpc.WithTransportCredentials(insecure.NewCredentials()))

    if err != nil {
        log.Fatal(err)
    }

    defer conn.Close()
    c := echo.NewEchoClient(conn)

    //client.CallUnary(c)

   // client.CallServerStream(c)
  //client.CallClientSteam(c)
  client.CallBidirectional(c)
}