云原生时代崛起的编程语言Go远程调用gRPC实战

@

目录

• 概述
  • 定义
  • 背景
  • 特点
  • 四种服务方法
• 实战
  • 环境配置
  • proto文件
  • 简单RPC
  • Token认证
  • 服务器流式RPC
  • 客户端流式RPC
  • 双向流式RPC

概述

定义

gRPC 官网地址 https://grpc.io/ 源码release最新版本v1.55.1

gRPC 官网文档地址 https://grpc.io/docs/

gRPC 源码地址 https://github.com/grpc/grpc

gRPC是一个现代的开源高性能远程过程调用(RPC)框架，可以在任何环境中运行。它可以高效地连接数据中心内和跨数据中心的服务，支持负载平衡、跟踪、运行状况检查和身份验证；同时也是一个CNCF孵化项目。

简单说gRPC是基于tcp协议使用http2.0，采用Protocol Buffers定义接口，因而相对于传统的Restful API来说，速度更快，数据更小，接口要求更严谨。

背景

在当前分布式和微服务主宰时代，服务拆分后服务与服务之间的通信就是进程与进程或服务器与服务器之间的调用，或许你马上就说可以采用http，http虽然便捷方便，但性能较低，这时间我们可以采用RPC（Remote Procredure Call）来实现，通过自定义协议发起TCP调用来提高传输效率；

RPC是一款语言中立、平台中立、开源的远程过程调用技术，客户端和服务端可以在多种环境中运行和交互，客户端和服务端可以采用不同语言开发。数据在进行网络传输的时候需要先进行序列化，而序列化协议有很多种，比如XML、Json、Thrift、Avro、Hessian、Kryo、Protocol Buffers、ProtoStuff。

• 序列化（serialization）就是将对象序列化为二进制形式（字节数组），一般也将序列化称为编码（Encode），主要用于网络传输、数据持久化等；
• 反序列化（deserialization）则是将从网络、磁盘等读取的字节数组还原成原始对象，以便后续业务的进行，一般也将反序列化称为解码（Decode），主要用于网络传输对象的解码，以便完成远程调用。

原生rpc(在go标准库net/rpc包下)编写相对复杂，需要自己去关注实现过程，没有代码提示。因此更多会使用gRPC。在gRPC中，客户机应用程序可以直接调用不同机器上的服务器应用程序上的方法，就像它是本地对象一样，使得更容易创建分布式应用程序和服务。gRPC基于定义服务的思想，指定可以远程调用的方法及其参数和返回类型。在服务器端，服务器实现这个接口，并运行gRPC服务器来处理客户端调用。在客户端，客户端有一个提供相同方法的存根(在某些语言中仅称为客户端)。gRPC客户端和服务器可以在各种环境中运行并相互通信——从Google内部的服务器到您自己的桌面——并且可以用任何gRPC支持的语言编写；如可以轻松地用Java创建gRPC服务器，用Go、Python或Ruby创建客户端。

特点

• 简单服务定义：使用Protocol Buffers(一种强大的二进制序列化工具集和语言)定义服务，设计一个新的协议，需要准确，高效和语言独立。
• 低延迟、高可扩展、分布式系统：快速开始并扩大规模，用一行代码安装运行时和开发环境，还可以使用框架扩展到每秒数百万个rpc。序列化后体积小，序列化和反序列化速度快。
• 跨语言和平台工作：以各种语言和平台为您的服务自动生成惯用的客户机和服务器存根.开发与云服务器通信的移动客户端。
• 双向流和集成验证：双向流和完全集成的基于HTTP/2传输的可插拔身份验证。
• 分层设计：支持扩展。身份验证、负载平衡、日志记录和监控等

四种服务方法

proto中rpc业务实际上是一个函数，由服务端重写（overwrite）的函数，根据rpc函数的入参和出参简单分为普通RPC、服务端流RPC、客户端流RPC、双端流RPC。

• 一元rpc：其中客户端向服务器发送单个请求并获得单个响应，就像普通的函数调用一样。
  • 一旦客户机调用了存根方法，服务器就会收到RPC调用的通知，其中包含该调用的客户机元数据、方法名称和指定的截止日期(如果适用)。
  • 然后，服务器可以直接发送回自己的初始元数据(必须在任何响应之前发送)，或者等待客户端的请求消息。首先发生的是特定于应用程序的。
  • 一旦服务器获得了客户机的请求消息，它就会执行创建和填充响应所需的任何工作。然后将响应(如果成功)连同状态详细信息(状态代码和可选状态消息)以及可选的尾随元数据一起返回给客户机。
  • 如果响应状态为OK，则客户机将获得响应，从而完成客户机端的调用。

rpc SayHello(HelloRequest) returns (HelloResponse);

• 服务器流式rpc：其中客户端向服务器发送请求并获得读取消息序列的流。客户端从返回的流中读取，直到没有更多的消息。gRPC保证了单个RPC调用中的消息排序。服务器流RPC类似于一元RPC，不同之处在于服务器在响应客户端的请求时返回消息流。发送完所有消息后，服务器的状态详细信息(状态码和可选状态消息)和可选的尾随元数据被发送到客户端。这就完成了服务器端的处理。客户机在拥有所有服务器消息后完成。

rpc LotsOfReplies(HelloRequest) returns (stream HelloResponse);

• 客户端流式rpc：其中客户端写入消息序列并将它们发送到服务器，同样使用提供的流。一旦客户机完成了消息的写入，它将等待服务器读取消息并返回其响应。gRPC再次保证了单个RPC调用中的消息排序。客户端流式RPC类似于一元RPC，不同之处是客户端向服务器发送消息流而不是单个消息。服务器用一条消息(连同它的状态详细信息和可选的尾随元数据)进行响应，通常是在它接收到所有客户机的消息之后，但不一定是这样。

rpc LotsOfGreetings(stream HelloRequest) returns (HelloResponse);

• 双向流式rpc：其中双方使用读写流发送消息序列。这两个流独立运行，因此客户端和服务器可以按照自己喜欢的顺序进行读写:例如，服务器可以等待接收到所有客户端消息后再写入响应，或者它可以交替地读取消息然后写入消息，或者其他一些读写组合，保留每个流中的消息顺序。在双向流RPC中，调用由调用方法的客户机和接收客户机元数据、方法名称和截止日期的服务器发起。服务器可以选择发回其初始元数据，或者等待客户端开始流式传输消息。客户端和服务器端流处理是特定于应用程序的。由于这两个流是独立的，客户端和服务器可以以任何顺序读写消息。例如，服务器可以等到接收到客户端的所有消息后再写入消息，或者服务器收到请求，然后发回响应，然后客户端根据响应发送另一个请求，以此类推。

rpc BidiHello(stream HelloRequest) returns (stream HelloResponse);

实战

环境配置

• 安装Protobuf

# 下载最新版本23.2的protoc，这个是protobuf代码生成工具，通过proto文件生成对应的代码，根据自己操作系统下载相应文件，这里以windows 64位系统为例
wget https://github.com/protocolbuffers/protobuf/releases/download/v23.2/protoc-23.2-win64.zip
# 解压并放在windows本地目录，并配置在Path路径下如D:\Program Files\protoc-23.2-win64\bin，在windows下命令行执行protoc --version检查是否安装配置正确

• 使用go get获取grpc的官方软件包

# 创建go项目grpc-demo，在GoLand IDE编写，并通过下面命令安装grpc核心库protoc,可以GoLand IDE安装protoc插件，实现语法高亮
go get google.golang.org/grpc

• 安装go的protoc代码生成工具。上面安装的是protocol编译器，它可以生成多种开发语言代码，这里使用go语言的工具是protoc-gen-go

# 在实际开发中最好指定具体的版本，这里是演示使用就直接用latest，在命令行中执行下面两条命令
go install google.golang.org/protobuf/cmd/protoc-gen-go@latest
go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@latest

在GOPATH目录下的bin目录就已经有刚刚安装的两个文件

proto文件

默认情况下gRPC使用Protocol Buffers（尽管它可以与JSON等其他数据格式一起使用)，Protocol Buffers是Google公司开发的一种跨语言和平台的序列化数据结构的方式,是一个灵活的、高效的用于序列化数据的协议。使用协议缓冲区时的第一步是定义要在proto文件中序列化的数据的结构:这是一个扩展名为.proto的普通文本文件。协议缓冲区数据被结构化为消息，其中每个消息都是包含一系列称为字段的名称-值对的信息的小逻辑记录。在普通的原型文件中定义gRPC服务，使用RPC方法参数和返回类型指定为协议缓冲区消息。一般来说，虽然可以使用proto2(当前默认协议缓冲区版本)，但建议将proto3与gRPC一起使用，因为它允许您使用所有gRPC支持的语言，并且避免了proto2客户端与proto3服务器通信的兼容性问题。

• message：用于在protobuf中定义消息类型，而消息就是需要进行传输的数据格式，类似于go中的struct，在消息中的数据字段由字段类型、字段名称、消息号，一个proto文件中可以定义多个消息类型，也即是多服务。
• 消息号：在消息体的定义中，每个字段都必须有一个唯一的标识号，范围是[1，2^29-1]
• 字段规则
  • required：默认规则，消息体中必填字段，不设置会导致编码解码的异常
  • optional：消息体中可选字段
  • repeated：消息体中可重复字段，重复的值的顺序会被保留，在go中重复的字段会定义为切片类型。
• 字段映射

• 默认值

• 嵌套消息：可以在其它消息类型中定义，使用消息类型，也可以使用外部定义的消息。
• import使用：import用于导入其它的proto文件。也即是类似有common.proto、user.proto、order.proto，在user.proto、order.proto可以通过import common.proto实现多proto文件之类引用管理。
• any任意类型：需要导入any.proto，属性使用google.protobuf.Any定义

简单RPC

简单RPC也叫一元RPC，其中客户端向服务器发送单个请求并获得单个响应，就像普通的函数调用一样。在go项目的src目录下创建simple目录，在simple创建proto目录，在创建user.proto

// 语法版本,指定使用proto3
syntax = "proto3";

// 指定生成的go_package,生成的go代码使用什么包package proto
option go_package = ".;proto";

// 服务定义,此处rpc服务的定义，一定要从服务端的角度考虑，即接受请求，处理请求并返回响应的一端
service UserService {
  // 远程调用方法定义
  rpc GetUser(UserRequest) returns (UserResponse) {}
}

// 包含用户编号的请求消息
message UserRequest {
  // 每个字典=最后序号1为唯一的标识号，必填
  int32 id = 1;
}

// 包含消息内容的响应消息
message UserResponse {
  int32 id = 1;
  string name = 2;
}

在命令行中执行如下操作用于生成go的代码文件

# 进入simple目录
cd src/simple
# 执行protoc命令
protoc --go_out=. --go-grpc_out=. proto/user.proto

如果client和server不在同一工程项目，proto/user.pb.go和proto/user_grpc.pb.go需要都复制到对应client和server相应项目下，这里先以此例说明。执行后生成proto/user.pb.go和proto/user_grpc.pb.go两个文件，其中包含用于填充、序列化和检索UserRequest和UserResponse消息类型的代码，生成客户机和服务器代码。

• 服务端流程及代码

  • 创建gRPC Server对象，可以简单理解为Server端抽象对象。
  • 将server（其包含被调用的服务端接口）注册到gRPC Server的内部注册中心，这样可以在接收到请求时，通过内部的服务发现，发现该服务端接口并转接进行逻辑处理。
  • 创建Listen，监听TCP端口。
  • gRPC Server开始Listen、Accept，直到Stop。

  将proto文件下的user.pb.go和user_grpc.pb.go复制一份放在src/simple/server目录，在src/simple/server创建main.go文件，内容如下

package main

import (
	"context"
	"fmt"
	"google.golang.org/grpc"
	"grpc-demo/src/simple/proto"
	"net"
)

type server struct {
	proto.UnimplementedUserServiceServer
}

func (s *server) GetUser(ctx context.Context, req *proto.UserRequest) (*proto.UserResponse, error) {
	// 服务端接口实现的业务逻辑
	fmt.Println("client端远程调用成功......, 当前请求传入id参数为", req.GetId())
	return &proto.UserResponse{
		Id:   req.GetId(),
		Name: "itxiaoshen",
	}, nil
}

func main() {
	//1. 开启端口
	listen, _ := net.Listen("tcp", ":7070")

	//2. 创建grpc服务
	grpcServer := grpc.NewServer()

	//3. 将编写好的服务注册到grpc
	proto.RegisterUserServiceServer(grpcServer, &server{})

	//4. 启动服务
	err := grpcServer.Serve(listen)
	if err != nil {
		fmt.Printf("failed to server: %v", err)
		return
	}
}

• 客户端流程及代码
  • 创建与给定目标（服务端）的连接交互。
  • 创建对应Client对象。
  • 发送RPC请求，等待同步响应，得到回调后返回响应结果。
  • 输出响应结果。

将proto文件下的user.pb.go和user_grpc.pb.go复制一份放在src/simple/client目录，在src/simple/client创建main.go文件，内容如下

package main

import (
	"context"
	"fmt"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
	pb "grpc-demo/src/simple/server/proto"
	"log"
)

func main() {
	//1. 与Server建立连接,此处禁用安全传输，这里没有使用加密验证
	conn, err := grpc.Dial("127.0.0.1:7070", grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		log.Fatalf("did not connect: %v", err)
	}
	defer conn.Close() //延时关闭连接

	//2. 与对应服务建立连接
	client := pb.NewUserServiceClient(conn)

	//3. 执行grpc调用
	resp, _ := client.GetUser(context.Background(), &pb.UserRequest{Id: 1})
	fmt.Println("client get user,id=", resp.GetId(), ",name=", resp.GetName())
}

启动server和client，客户端正确返回结果，服务端也打印请求日志。

Token认证

gRPC提供了一个接口PerRPCCredentials，接口位于credentials包下，接口中有两个方法，方法需要由客户端来实现

client的main.go

package main

import (
	"context"
	"fmt"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
	pb "grpc-demo/src/simple/server/proto"
	"log"
)

type ClientTokenAuth struct {
}

func (c ClientTokenAuth) GetRequestMetadata(ctx context.Context, uri ...string) (map[string]string, error) {
	return map[string]string{
		"appId":  "itxs",
		"appKey": "11223344",
	}, nil
}

func (c ClientTokenAuth) RequireTransportSecurity() bool {
	return false
}

func main() {
	var opts []grpc.DialOption
	opts = append(opts, grpc.WithTransportCredentials(insecure.NewCredentials())) //这里我们不使用TLS，因此这里传入空
	opts = append(opts, grpc.WithPerRPCCredentials(new(ClientTokenAuth)))         //传入我们自定义的验证方式【Token】
	conn, err := grpc.Dial("127.0.0.1:7070", opts...)

	//1. 与Server建立连接,此处禁用安全传输，这里没有使用加密验证
	//conn, err := grpc.Dial("127.0.0.1:7070", grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		log.Fatalf("did not connect: %v", err)
	}
	defer conn.Close() //延时关闭连接

	//2. 与对应服务建立连接
	client := pb.NewUserServiceClient(conn)

	//3. 执行grpc调用
	resp, err := client.GetUser(context.Background(), &pb.UserRequest{Id: 1})
	if err != nil {
		fmt.Println("client get user error=", err.Error())
	} else {
		fmt.Println("client get user,id=", resp.GetId(), ",name=", resp.GetName())
	}

}

server的main.go

package main

import (
	"context"
	"errors"
	"fmt"
	"google.golang.org/grpc"
	"google.golang.org/grpc/metadata"
	"grpc-demo/src/simple/proto"
	"net"
)

type server struct {
	proto.UnimplementedUserServiceServer
}

func (s *server) GetUser(ctx context.Context, req *proto.UserRequest) (*proto.UserResponse, error) {
	fmt.Println("接收client端远程调用请求")
	//获取客户端传入的元数据信息
	md, ok := metadata.FromIncomingContext(ctx)
	if !ok {
		return nil, errors.New("未传输token")
	}
	var appId string
	var appKey string
	if v, ok := md["appid"]; ok {
		appId = v[0]
	}
	if v, ok := md["appkey"]; ok {
		appKey = v[0]
	}
	if appId != "itxs" || appKey != "11223344" {
		fmt.Println("token 不正确")
		return nil, errors.New("token 不正确")
	}
	fmt.Println("token 验证正确")
	// 服务端接口实现的业务逻辑
	fmt.Println("client端远程调用成功......, 当前请求传入id参数为", req.GetId())
	return &proto.UserResponse{
		Id:   req.GetId(),
		Name: "itxiaoshen",
	}, nil
}

func main() {
	//1. 开启端口
	listen, _ := net.Listen("tcp", ":7070")

	//2. 创建grpc服务
	grpcServer := grpc.NewServer()

	//3. 将编写好的服务注册到grpc
	proto.RegisterUserServiceServer(grpcServer, &server{})

	//4. 启动服务
	err := grpcServer.Serve(listen)
	if err != nil {
		fmt.Printf("failed to server: %v", err)
		return
	}
}

启动server和client，返回正确的结果，反之如果客户端输入不正确appId或appKey则会返回token 不正确。

gRPC将各种认证方式浓缩到一个凭证（credentials）上，可以单独使用一种拼争，比如只使用TLS或者只使用自定义凭证，也可以多种凭证组合，gRPC提供统一的gRPC验证机制，使得研发人员使用方便，这也是gRPC设计的巧妙之处。

服务器流式RPC

服务器流式RPC这里使用文件下载的案例来演示，创建file.proto文件

// 语法版本,指定使用proto3
syntax = "proto3";

// 指定生成的go_package,生成的go代码使用什么包package proto
option go_package = "./proto;proto";

// 服务定义,此处rpc服务的定义，一定要从服务端的角度考虑，即接受请求，处理请求并返回响应的一端
service FileService {
  // 文件下载远程调用方法定义
  rpc DownLoad(FileRequest) returns (stream FileResponse) {}
}

// 包含文件名的文件请求消息
message FileRequest {
  // 每个字典=最后序号1为唯一的标识号，必填
  string name = 1;
}

// 包含消息内容的响应消息
message FileResponse {
  string name = 1;
  bytes content = 2;
}

在命令行中执行如下操作用于生成go的代码文件

# 进入simple目录
cd src/stream
# 执行protoc命令
protoc --go_out=. --go-grpc_out=. proto/file.proto

创建server_stream_server.go实现服务端文件下载

package main

import (
	"fmt"
	"google.golang.org/grpc"
	"grpc-demo/src/stream/proto"
	"io"
	"log"
	"net"
	"os"
)

type FileService struct {
	proto.UnimplementedFileServiceServer
}

func (FileService) DownLoad(req *proto.FileRequest, stream proto.FileService_DownLoadServer) error {
	fmt.Println(req)
	file, err := os.Open("src\\stream\\static\\winutils-master.zip")
	if err != nil {
		panic(err)
	}
	defer file.Close()

	for {
		buf := make([]byte, 1024)
		_, err = file.Read(buf)
		if err == io.EOF {
			break
		}

		if err != nil {
			panic(err)
		}
		stream.Send(&proto.FileResponse{
			Content: buf,
		})
	}
	return nil
}

func main() {
	listen, _ := net.Listen("tcp", ":7070")
	// 创建grpc服务
	grpcServer := grpc.NewServer()
	// 注册服务
	proto.RegisterFileServiceServer(grpcServer, &FileService{})
	// 启动服务
	err := grpcServer.Serve(listen)
	if err != nil {
		log.Fatal("服务启动失败：", err)
		return
	}
}

创建server_stream_client.go实现客户端文件下载

package main

import (
	"bufio"
	"context"
	"fmt"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
	"grpc-demo/src/stream/proto"
	"io"
	"log"
	"os"
)

func main() {
	conn, err := grpc.Dial("127.0.0.1:7070", grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		log.Fatal("连接失败:", err)
		return
	}
	defer conn.Close()

	// 建立连接
	client := proto.NewFileServiceClient(conn)
	// 执行rpc调用
	serverStream, err := client.DownLoad(context.Background(), &proto.FileRequest{Name: "hello.zip"})
	if err != nil {
		log.Fatalln("获取流出错", err)
	}

	file, err := os.OpenFile("winutils-master-new.zip", os.O_CREATE|os.O_WRONLY, 0600)
	if err != nil {
		panic(err)
	}
	defer file.Close()

	writer := bufio.NewWriter(file)

	for {
		resp, err := serverStream.Recv()
		if err != nil {
			if err == io.EOF {
				fmt.Println("客户端数据接收完成")
				err := serverStream.CloseSend()
				if err != nil {
					log.Fatal(err)
				}
				break
			}
		}
		writer.Write(resp.Content)
	}
	writer.Flush()
}

运行服务端和客户端，最终按照预期下载文件

客户端流式RPC

客户端流式RPC这里使用文件上传的案例来演示，修改file.proto文件，增加UpFileService相关内容

// 语法版本,指定使用proto3
syntax = "proto3";

// 指定生成的go_package,生成的go代码使用什么包package proto
option go_package = "./proto;proto";

// 服务定义,此处rpc服务的定义，一定要从服务端的角度考虑，即接受请求，处理请求并返回响应的一端
service FileService {
  // 文件下载远程调用方法定义
  rpc DownLoad(FileRequest) returns (stream FileResponse) {}
}

// 服务定义,此处rpc服务的定义，一定要从服务端的角度考虑，即接受请求，处理请求并返回响应的一端
service UpFileService {
  // 文件下载远程调用方法定义
  rpc UpLoad(stream UpFileRequest) returns (UpFileResponse) {}
}

// 包含文件名的文件请求消息
message FileRequest {
  // 每个字典=最后序号1为唯一的标识号，必填
  string name = 1;
}

// 包含消息内容的响应消息
message FileResponse {
  string name = 1;
  bytes content = 2;
}

// 包含文件名的文件请求消息
message UpFileRequest {
  string name = 1;
  bytes content = 2;
}

// 包含文件名的文件请求消息
message UpFileResponse {
  string state = 1;
}

在命令行中执行如下操作用于生成go的代码文件

# 进入simple目录
cd src/stream
# 执行protoc命令
protoc --go_out=. --go-grpc_out=. proto/file.proto

创建client_stream_server.go实现服务端文件下载

package main

import (
	"bufio"
	"google.golang.org/grpc"
	"grpc-demo/src/stream/proto"
	"io"
	"log"
	"net"
	"os"
)

type UpFileService struct {
	proto.UnimplementedUpFileServiceServer
}

func (UpFileService) UpLoad(stream proto.UpFileService_UpLoadServer) error {
	file, err := os.OpenFile("src/stream/static/apache-maven-3.8.6-bin-new.zip", os.O_CREATE|os.O_WRONLY, 0600)
	if err != nil {
		panic(err)
	}
	defer file.Close()

	writer := bufio.NewWriter(file)

	for {
		resp, err := stream.Recv()
		if err != nil {
			if err == io.EOF {
				break
			}
		}
		writer.Write(resp.Content)
	}

	writer.Flush()

	stream.SendAndClose(&proto.UpFileResponse{
		State: "success",
	})

	return nil
}

func main() {
	listen, _ := net.Listen("tcp", ":7070")
	// 创建grpc服务
	grpcServer := grpc.NewServer()
	// 注册服务
	proto.RegisterUpFileServiceServer(grpcServer, &UpFileService{})
	// 启动服务
	err := grpcServer.Serve(listen)
	if err != nil {
		log.Fatal("服务启动失败：", err)
		return
	}
}

创建client_stream_client.go实现客户端文件下载

package main

import (
	"context"
	"fmt"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
	"grpc-demo/src/stream/proto"
	"io"
	"log"
	"os"
)

func main() {
	conn, err := grpc.Dial("127.0.0.1:7070", grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		log.Fatal("连接失败:", err)
		return
	}
	defer conn.Close()

	// 建立连接
	client := proto.NewUpFileServiceClient(conn)
	// 执行rpc调用
	clientStream, err := client.UpLoad(context.Background())
	if err != nil {
		log.Fatalln("获取流出错", err)
	}

	file, err := os.Open("g:\\apache-maven-3.8.6-bin.zip")
	if err != nil {
		log.Fatalln(err)
	}
	defer file.Close()

	for {
		buf := make([]byte, 1024)
		_, err = file.Read(buf)
		if err == io.EOF {
			break
		}

		if err != nil {
			panic(err)
		}
		clientStream.Send(&proto.UpFileRequest{
			Name:    "apache-maven-3.8.6-bin.zip",
			Content: buf,
		})
	}

	resp, err := clientStream.CloseAndRecv()
	fmt.Println(resp, err)
}

运行服务端和客户端，最终按照预期上传文件

双向流式RPC

双向流式RPC这里类似聊天的场景，双方可以随时收发，修改file.proto文件，增加ChatService相关内容

// 语法版本,指定使用proto3
syntax = "proto3";

// 指定生成的go_package,生成的go代码使用什么包package proto
option go_package = "./proto;proto";

// 服务定义,此处rpc服务的定义，一定要从服务端的角度考虑，即接受请求，处理请求并返回响应的一端
service FileService {
  // 文件下载远程调用方法定义
  rpc DownLoad(FileRequest) returns (stream FileResponse) {}
}

// 服务定义,此处rpc服务的定义，一定要从服务端的角度考虑，即接受请求，处理请求并返回响应的一端
service UpFileService {
  // 文件下载远程调用方法定义
  rpc UpLoad(stream UpFileRequest) returns (UpFileResponse) {}
}

service ChatService {
  // 文件下载远程调用方法定义,文字聊天
  rpc TextChat(stream TextRequest) returns (stream TextResponse) {}
}

// 包含文件名的文件请求消息
message FileRequest {
  // 每个字典=最后序号1为唯一的标识号，必填
  string name = 1;
}

// 包含消息内容的响应消息
message FileResponse {
  string name = 1;
  bytes content = 2;
}

// 包含文件名的文件请求消息
message UpFileRequest {
  string name = 1;
  bytes content = 2;
}

// 包含文件名的文件请求消息
message UpFileResponse {
  string state = 1;
}

// 包含文件名的文件请求消息
message TextRequest {
  // 每个字典=最后序号1为唯一的标识号，必填
  string message = 1;
}

// 包含消息内容的响应消息
message TextResponse {
  string message = 1;
}

在命令行中执行如下操作用于生成go的代码文件

# 进入simple目录
cd src/stream
# 执行protoc命令
protoc --go_out=. --go-grpc_out=. proto/file.proto

创建both_stream_server.go实现服务端文件下载

package main

import (
	"fmt"
	"google.golang.org/grpc"
	"grpc-demo/src/stream/proto"
	"log"
	"net"
)

type ChatService struct {
	proto.UnimplementedChatServiceServer
}

func (ChatService) TextChat(stream proto.ChatService_TextChatServer) error {
	for i := 0; i < 10; i++ {
		req, _ := stream.Recv()
		fmt.Println(req)
		stream.Send(&proto.TextResponse{
			Message: fmt.Sprintf("server send world to client!i=%d", i),
		})
	}
	return nil
}

func main() {
	listen, _ := net.Listen("tcp", ":7070")
	// 创建grpc服务
	grpcServer := grpc.NewServer()
	// 注册服务
	proto.RegisterChatServiceServer(grpcServer, &ChatService{})
	// 启动服务
	err := grpcServer.Serve(listen)
	if err != nil {
		log.Fatal("服务启动失败：", err)
		return
	}
}

创建both_stream_client.go实现客户端文件下载

package main

import (
	"context"
	"fmt"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
	"grpc-demo/src/stream/proto"
	"log"
)

func main() {
	conn, err := grpc.Dial("127.0.0.1:7070", grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		log.Fatal("连接失败:", err)
		return
	}
	defer conn.Close()

	// 建立连接
	client := proto.NewChatServiceClient(conn)
	// 执行rpc调用
	stream, err := client.TextChat(context.Background())
	if err != nil {
		log.Fatalln("获取流出错", err)
	}

	for i := 0; i < 10; i++ {
		stream.SendMsg(&proto.TextRequest{
			Message: fmt.Sprintf("client send hello to server!i=%d", i),
		})
		resp, err := stream.Recv()
		fmt.Println(resp, err)
	}

}

运行服务端和客户端，最终按照预期实现双方文字聊天消息发送

• 本人博客网站IT小神 www.itxiaoshen.com