Go微服务实战｜第3章：gRPC Server-side Streaming RPC

原创

Madman

8677 次阅读

Synopsis: 由于 gRPC 支持 HTTP/2，所以可以使用 HTTP/2 的 Stream 特性，Server-side Streaming RPC 是指客户端发送一个请求 message 后，将会从服务端接收多个响应 messages，服务端返回 Streaming 数据流。它非常适合要传输的结构化数据比较大的场景，或者服务端需要主动 Push 数据的场景，比如直播、聊天等，服务端可以持续返回响应包，而无需客户端再发送请求

代码已上传到 https://github.com/wangy8961/grpc-go-tutorial/tree/v0.3 ，欢迎 star

1. Protocol buffers 定义

要定义 gRPC Server-side Streaming RPC，只需要在 Protocol buffers 中为 RPC 方法的响应前增加 stream 关键字即可

gRPC Server-side Streaming RPC

我们将为 Math 服务添加一个新的 RPC 调用，计算一个合数的所有质因数，比如 15 = 3 * 5，所以 15 的质因数为 3 和 5

修改 grpc-go-tutorial/math/mathpb/math.proto 文件：

syntax = "proto3";

option go_package="mathpb";

package math;

// The math service definition.
service Math {
    ...

    // PrimeFactors is server-side streaming RPC
    rpc PrimeFactors(PrimeFactorsRequest) returns (stream PrimeFactorsResponse) {};
}

...

// The request message for PrimeFactors.
message PrimeFactorsRequest {
    int64 num = 1;
}

// The response message for PrimeFactors.
message PrimeFactorsResponse {
    int64 result = 1;
}

然后切换到 .proto 文件所在的目录：

[root@CentOS ~]# cd grpc-go-tutorial/math/mathpb/
[root@CentOS mathpb]# protoc --go_out=plugins=grpc:. *.proto

将会更新 math.pb.go 文件

2. 实现 gRPC 服务端

服务端需要实现 math.pb.go 中的 server API：

// MathServer is the server API for Math service.
type MathServer interface {
    ...

    // PrimeFactors is server-side streaming RPC
    PrimeFactors(*PrimeFactorsRequest, Math_PrimeFactorsServer) error
}

所以，修改 grpc-go-tutorial/math/math_server/main.go 文件：

// Package main implements a server for Math service.
package main

import (
    "context"
    "flag"
    "fmt"
    "log"
    "net"

    pb "github.com/wangy8961/grpc-go-tutorial/math/mathpb"
    "google.golang.org/grpc"
)

// server is used to implement mathpb.MathServer.
type server struct{}

...

// PrimeFactors implements mathpb.MathServer
func (s *server) PrimeFactors(in *pb.PrimeFactorsRequest, stream pb.Math_PrimeFactorsServer) error {
    fmt.Printf("--- gRPC Server-side Streaming RPC ---\n")
    fmt.Printf("request received: %v\n", in)

    num := in.Num
    factor := int64(2)

    for num > 1 {
        if num%factor == 0 {
            stream.Send(&pb.PrimeFactorsResponse{Result: factor})
            num = num / factor
            continue
        }
        factor++
    }

    return nil
}

func main() {
    port := flag.Int("port", 50051, "the port to serve on")
    flag.Parse()

    lis, err := net.Listen("tcp", fmt.Sprintf(":%d", *port)) // Specify the port we want to use to listen for client requests
    if err != nil {
        log.Fatalf