Golang中gRPC使用及原理探究-CSDN博客

概述

gRPC是一种进程间通信技术在微服务和云原生领域都有着广泛的应用。

gRPC的优势

1. 提供高效的进程间通信gRPC使用HTTP/2协议以及protobuf来序列化消息从而保证了高效的数据传输和编解码速度可以提供一个高性能的进程间通信。
2. 具有简单且定义良好的服务接口和模式、支持多语言gRPC使用protobuf来定义消息以及service通过proroc的编译即可生成多种语言的客户端和服务端service接口的定义。
3. 支持双工流gRPC在客户端和服务端都提供了对流的支持。
4. 比较成熟并且被广泛使用gRPC目前已经发展成熟并在许多软件中得到了应用比如etcd、kubernetes的组件中等。

如下图所示gRPC支持多种语言编写的客户端和服务端进行通信在使用之前需要使用proto文件来定义message和service然后即可生成不同语言的数据结构以及客户端和服务端接口服务端需要我们实现接口中的方法即可实现进程间的通信。

 

1、gRPC入门

gRPC官网https://grpc.io

protobufhttps://github.com/protocolbuffers/protobuf
 

1.1 protobuf安装

在使用gRPC时需要使用一种接口定义语言interface definition languageIDL来定义出服务的接口以及传递的消息类型这种IDL就是protobuf因此需要先下载protobuf相关的组件

需要下面几个程序

• protoc该程序用于解析.proto文件
• protoc-gen-go该程序用于根据proto文件中的message生成对应的go结构体类型
• protoc-gen-go-grpc该程序用于根据proto文件中的service生成gRPC客户端和服务端代码

安装protoc

下载网址https://github.com/protocolbuffers/protobuf/releases

根据不同操作系统下载对应的文件如果是windows则下载.zip压缩文件

下载后解压文件并将bin目录加入环境变量中

使用命令行查看版本信息

$ protoc --version
libprotoc 3.20.1

安装protoc-gen-go和protoc-gen-go-grpc

go install google.golang.org/protobuf/cmd/protoc-gen-go
go install google.golang.org/grpc/cmd/protoc-gen-go-grpc

 

1.2 使用gRPC

1.2.1 定义proto文件

在使用之前需要使用proto文件来定义出message类型和service接口比如下面的greeter.proto

syntax = "proto3";

option go_package = "./;greeter";

service Greeter {
  rpc SayHello(HelloRequest) returns (HelloReply);
}

message HelloRequest  {
  string data = 1;
}

message HelloReply {
  string data = 1;
}

• syntax表示proto语法版本在此使用proto3版本
• 第3行用于声明生成的go代码的package
• 第9行和第13行定义了两个消息都包含一个data的字段类型为string
• 第5行定义了一个名为Greeter的service该service包含一个rpc方法为SayHelloHelloRequest为请求类型HelloReply为响应类型

 

1.2.2 编译proto文件

使用proto文件定义了message和service后就需要使用protoc来将proto编译为不同语言的代码在此就编译为go的代码。

使用下面的命令将proto文件编译为go代码

$ protoc --go_out=. --go-grpc_out=. *.proto

• protoc用于解析proto文件中的内容生成语法树。
• –go-out和**–go-grpc_out用于指定其它的插件这两个插件的名称为protoc-gen-go和proto-gen-go-grpc**。
• 这两个插件可以从标准输入中读取到protoc解析的结构然后生成相应的go代码。
• protoc-gen-go用于根据message生成对应的go结构体类型以及相应的方法
• protoc-gen-go-grpc用于根据service生成对应go的客户端代码和服务端service接口定义

编译之后会生成两个文件分别是greeter.pb.go和greeter_grpc.pb.go

在greeter.pb.go中生成了相应的结构体


在greeter_grpc.pb.go中生成了客户端和服务端代码


 

1.2.3 编写客户端和服务端代码

在上面生成代码后就可以编写服务端和客户端的代码了。对于客户端我们可以直接使用生成的代码来调用服务而对于服务端来说我们需要实现GreeterServer中定义的方法。

目录结构如下

grpc_test                  
|-- client
|   |-- main.go
|-- server
|   |-- main.go
|-- proto
	|-- greeter.pb.go
|   |-- greeter.proto
|   |-- greeter_grpc.pb.go
|-- go.mod
|-- go.sum   

服务端代码server.go如下

package main

import (
	"context"
	"fmt"
	"net"

	greeter "code/grpc_test/proto"
	"google.golang.org/grpc"
)

// 声明GreeterImpl结构体并实现SayHello方法
type GreeterImpl struct {
    // 将greeterUnimplementedGreeterServer内置
	greeter.UnimplementedGreeterServer
}

// 实现SayHello方法
func (g *GreeterImpl) SayHello(ctx context.Context, req *greeter.HelloRequest) (*greeter.HelloReply, error) {
	fmt.Println("req:", req.Data)
	return &greeter.HelloReply{Data: "hello client"}, nil
}

func main() {
    // 创建listener
	listener, err := net.Listen("tcp", ":8080")
	if err != nil {
		panic(err)
	}
    
	// 创建grpc server
	server := grpc.NewServer()
	
    // 注册Greeter服务到server中
	greeter.RegisterGreeterServer(server, &GreeterImpl{})
	
    // 启动服务
	if err = server.Serve(listener); err != nil {
		fmt.Println("Server:", err.Error())
	}
}

在服务端中声明了一个GreeterImpl的结构体用于实现GreeterServer中的方法。将greeter.UnimplementedGreeterServer内置在了GreeterImpl中因为GreeterServer生成了一个未导出的方法mustEmbedUnimplementedGreeterServer()而将UnimplementedGreeterServer内置我们就不需要来实现这个方法只需要实现SayHello方法来覆盖UnimplementedGreeterServer中的SayHello即可。

在main函数中首先创建了一个listener用于监听网络连接然后创建了一个grpc server接着将服务注册到server中并启动服务。至此一个简单的grpc 服务就启动起来了可以随时接收客户端的调用了。

 

客户端代码client.go如下

package main

func main() {
    // 1. 使用Dial连接服务端
	conn, err := grpc.Dial("127.0.0.1:8080",
        // 1.1 创建不安全的证书
		grpc.WithTransportCredentials(insecure.NewCredentials()),
        // 1.2 阻塞直到连接建立成功                   
		grpc.WithBlock(),
	)
	if err != nil {
		panic(err)
	}
	defer conn.Close()

    // 创建出greeter Service
	greeterClient := greeter.NewGreeterClient(conn)
	
    // 调用SayHello方法
	reply, err := greeterClient.SayHello(context.Background(), &greeter.HelloRequest{Data: "hello, server"})
	if err != nil {
		slog.Error("call SayHello", "error", err)
		return 
	}
	
	fmt.Println("server reply ", reply.Data)
	
}

客户端的代码比较简单因为客户端的调用代码已经由生成的代码实现我们只需要创建出客户端来调用即可。

 

2、gRPC的通信模式

gRPC有四种通信模式分别是

1. 一元RPC模式
2. 客户端流RPC模式
3. 服务端流RPC模式
4. 双向流RPC模式
    

2.1 一元RPC模式

在上面定义的SayHello就是一元RPC模式该模式是基于请求响应的客户端发送请求来调用某个方法服务端收到请求调用指定方法并返回响应。

 

2.2 客户端流RPC模式

在客户端流模式中可以在客户端和服务端直接创建一个单向流客户端可以向流中发送多个请求服务端读取流并对所有请求聚合处理后返回给客户端一个响应。

 

下面的proto文件用于生成客户端流模式的方法

syntax = "proto3";

option go_package = "./;greeter";

service Greeter {
  // 在请求参数前添加一个stream	
  rpc ClientStreamSayHello(stream HelloRequest) returns (HelloReply);
}

message HelloRequest  {
  string data = 1;
}

message HelloReply {
  string data = 1;
}

生成的代码如下

 

服务端代码

func (g *GreeterImpl) ClientStreamSayHello(stream greeter.Greeter_ClientStreamSayHelloServer) error {
	for {
        // 从流中读取消息
		req, err := stream.Recv()
        // 读到EOF说明数据读取完毕
		if err == io.EOF {
			break
		} else if err != nil {
			return err
		}

		fmt.Printf("From Client:%s\n", req.GetData())
	}

    // 发送响应并关闭流
	stream.SendAndClose(&greeter.HelloReply{Data: "Receive over"})
	return nil
}

客户端代码

func CallClientStreamSayHello(client greeter.GreeterClient) {
    // 获取流
	stream, err := client.ClientStreamSayHello(context.TODO())
	if err != nil {
		panic(fmt.Errorf("Get stream error:%s", err.Error()))
	}
    // 发送请求到流中
	for i := 0; i < 10; i++ {
		stream.Send(&greeter.HelloRequest{Data: fmt.Sprintf("Hello %d", i)})
		time.Sleep(time.Second)
	}
    
    // 关闭流并读取响应
	reply, err := stream.CloseAndRecv()
	if err != nil {
		fmt.Println("Recv error:", err)
	}

	fmt.Println("Call rpc success, reply: ", reply.Data)
}

2.3 服务端流RPC模式

与客户端流类似在服务端流模式中可以在客户端和服务端直接创建一个单向流客户端发送一个请求给服务端服务端处理请求并通过流返回多个响应。通过这种方式也可以实现服务端推送。

下面的proto文件用于生成客户端流模式的方法

syntax = "proto3";

option go_package = "./;greeter";

service Greeter {
  // 在响应参数前添加一个stream	
  rpc ServerStreamSayHello(HelloRequest) returns (stream HelloReply);
}

message HelloRequest  {
  string data = 1;
}

message HelloReply {
  string data = 1;
}

生成的代码如下

 

服务端代码

func (g *GreeterImpl) ServerStreamSayHello(req *greeter.HelloRequest, stream greeter.Greeter_ServerStreamSayHelloServer) error {
	fmt.Println("Recv:", req.GetData())

	for i := 0; i < 10; i++ {
        // 向流中发送响应
		stream.Send(&greeter.HelloReply{Data: fmt.Sprintf("Hello %d", i)})
		time.Sleep(time.Second)
	}

	return nil
}

客户端代码

func CallServerStreamSayHello(client greeter.GreeterClient) {
    // 发送请求并获取流
	stream, err := client.ServerStreamSayHello(context.TODO(), &greeter.HelloRequest{Data: "hello, server"})
	if err != nil {
		fmt.Println("Get stream error:", err)
	}

	for {
        // 循环从流中读取响应
		reply, err := stream.Recv()
		if err != nil && err != io.EOF {
			fmt.Println("Recv error:", err)
		} else if err == io.EOF {
			fmt.Println("Recv EOF")
			break
		}

		fmt.Println("Recv:", reply.GetData())
	}
}

 

2.4双向流RPC模式

在双向流模式中客户端和服务端之间建立一个双向的流客户端和服务都可以往流中发送消息通过双向流可以实现请求响应模型也可以实现服务端主动推送消息。

 

下面的proto文件用于生成客户端流模式的方法

syntax = "proto3";

option go_package = "./;greeter";

service Greeter {
  // 在请求和响应参数前都添加一个stream	
  rpc StreamSayHello(stream HelloRequest) returns (stream HelloReply);
}

message HelloRequest  {
  string data = 1;
}

message HelloReply {
  string data = 1;
}

生成的代码如下

 

服务端代码

func (g *GreeterImpl) StreamSayHello(stream greeter.Greeter_StreamSayHelloServer) error {
	for {
        // 从流中读取客户端消息
		request, err := stream.Recv()
		if err != nil && err != io.EOF {
			fmt.Println("Recv error:", err)
		} else if err == io.EOF {
			fmt.Println("Recv EOF")
			break
		}
		fmt.Println("Recv:", request.GetData())
		
        // 向流中发送响应
		stream.Send(&greeter.HelloReply{Data: "Hello client"})
	}

	return nil
}

客户端代码

func CallStreamSayHello(client greeter.GreeterClient) {
    // 获取流
	stream, err := client.StreamSayHello(context.Background())
	if err != nil {
		panic(fmt.Errorf("get stream error:", err))
	}

	for i := 0; i < 10; i++ {
        // 通过流发送消息
		err = stream.Send(&greeter.HelloRequest{Data: "Hello server"})
		if err != nil {
			fmt.Println("Send error:", err)
			continue
		}
        // 读取服务响应
		reply, err := stream.Recv()
		if err != nil && err != io.EOF {
			fmt.Println("Recv error:", err)
		} else if err == io.EOF {
			fmt.Println("Recv EOF")
			break
		}
		fmt.Println("Recv:", reply.GetData())
		time.Sleep(time.Second)
	}

	_ = stream.CloseSend()
}

上面是一个简单的demo由于客户端和服务端都是可以任意的收发数据的因此对于读取的接收我们也可以启动多个goroutine来处理。

 

3、 gRPC的底层原理探究

gRPC的底层传输协议走的协议是HTTP/2采用的序列化协议是protobuf因此gRPC的传输效率和编解码效率都是比较高的。接下来将会使用抓包的方法来探究gRPC以及HTPP/2协议的底层原理。

 

3.1 HTPP/2协议介绍及探究

由于gRPC是基于HTTP/2的因此了解HTTP/2是非常有必要的。

之前也看过HTTP/2相关的八股文包括它相比于HTTP/1.1版本多了哪些特性比如头部压缩、二进制帧等等等解决了HTTP/1.1的队头阻塞问题

也了解到其中的一些概念比如流、帧。。。

但是是真的不理解什么叫流数据帧看的是一头雾水虽然记住了概念但是没用一点用除了稍微能应付一下面试。

但是当我真正使用抓包软件抓取HTTP/2的包以及深入源码才真正的理解了什么是流、数据帧为什么可以实现并发传输为什么HTTP/2的性能要比HTTP/1.1强。

 

3.1.1 HTTP/2简介及为什么需要HTTP/2

首先在讲解HTTP/2协议之前了解一下HTTP/1.1协议的一些缺点

1.明文传输、头部冗余

HTTP/1.1采用明文传输的方式请求分为请求行、请求头以及请求体组成。请求行和请求头中的内容都是以字符串方式的明文传输并且没有进行压缩。在每次请求和响应时都会传输大量的header并且很多header在多次请求响应时都是相同的因此就存在大量的冗余头部造成网络传输效率降低。

 
2.基于请求响应模型效率低下

HTTP/1.1是基于请求响应模型的客户端发送请求服务端处理并响应请求。在第一个请求的响应到达之前客户端不能再次发送请求。

虽然HTTP/1.1有pipeline管道传输但是这个特性也存在一些坑因此很少被使用。

管道传输是指客户端可以发送多个请求但是服务端必须要按照顺序来响应。因为HTTP/1.1协议的报文中中并没有标识请求对应的响应因此如果服务端不按照客户端的请求顺序来响应数据那么就无法分辨出哪个响应对应于哪个请求。

因此如果第一个请求处理的速度很慢就会导致后面的请求阻塞。就算采用并行的方式处理这个三个请求也需要按顺序回复响应效率低下。

看过go的http标准库实现代码的就可以知道对于每个客户端只会启动一个goroutine来处理处理的流程如下图

因此在处理前面的请求时其它的请求数据是会保存在socket的接收缓冲区中的而缓冲区的大小是有限的如果处理前面的请求需要很久的时间就会导致后面请求阻塞甚至会填满缓冲区导致不能再接收数据。

因此很多浏览器为了解决这个问题就会创建多个tcp连接从而并行获取数据比如chrome对于一个域名最多可以创建6个tcp连接。

 
HTTP/2协议简介

HTTP/2协议简称h2于2015年正式发布大多数主流的浏览器都已经提供了支持。

HTTP/2协议出来的目的是为了改善HTTP的性能因此它可以兼容HTTP/1.1它并没有再URI中引入新的协议名仍然使用http://表示http协议https://表示基于tls、ssl的http协议只在应用层做了改变依然使用TCP传输层协议。

HTTP/2把HTTP分解成了「语义」和「语法」两个部分「语义」层不做改动与 HTTP/1.1 完全一致比如请求方法、状态码、头字段等规则保留不变。

但是HTTP/2 在「语法」层面做了很多改造基本改变了 HTTP 报文的传输格式。

 

3.1.2 HTTP/2特性介绍

接下来介绍HTTP/2中的一些关键特性

①并发传输、多路复用、二进制分帧

首先介绍三个概念

流Stream在一个已建立的连接上的双向字节流。一个流可以携带一条或多条消息。

帧FrameHTTP/2中最小的传输单元。每一帧都包含一个帧头它至少要标记该帧所属的流。

消息Message完整的帧序列映射为一条逻辑上的HTTP消息由一帧或多帧组成。这样的话允许消息进行多路复用客户端和服务端能够将消息分解成独立的帧交叉发送它们然后在另一端重新组合。

        在HTTP/1.1中是基于请求/响应模型的客户端在发出请求到接收到响应之间是不能再次发出请求的。如果要并发请求多个资源就需要通过建立多条TCP连接来实现。

        而HTTP/2则采用了多路复用的方式在一条TCP连接上可以存在多个流Stream每个Stream都会分配一个Stream ID多个Stream之间可以乱序传输并且可以并行处理。

        并且这个数据流是双向的服务端无需按照流的发送顺序来响应同时服务端也可以主动向客户端推送数据。当客户端请求服务器时分配的流ID都是奇数而服务端主动推送数据时分配的流ID则是偶数通过这样的方式客户端便可以知道接收到的数据是服务端的响应还是服务端主动推送的。

如下图所示在一个TCP连接中可以存在多个流在不同流中传输的帧是可以乱序发送的。比如两个HTTP请求分别处于不同的流中每个请求都由Header帧和Data帧组成那么我可以按多种顺序发送H1,h2,D1,D2或者H1,H2,D2,D1等等等

但是在同一个流中的帧是不能乱序发送的比如一个HTTP消息由Header帧和Data帧组成那么在发送的时候必须先发送Header帧再发送Data帧。

流是一个逻辑的概念通过流ID来区分不同的流同一个连接中的流是不能复用的并且流是顺序递增的。

 

相比于HTTP/1.1明文传输的方式HTTP/2采用了二进制帧的方式传输。

一条HTTP消息由1个或多个帧组成帧需要在流中传输因此一个帧有所属的流

每个帧又由两部分组成帧头和负载如下图所示

帧头部分占9个字节总共包含4部分内容分别是

1. 长度用于存放payload的长度。因为TCP是流式协议需要我们自己来区分数据的边界因此在读取帧的时候先读取固定大小的帧头就可以解析出payload的长度然后再次读取该长度的内容即可。

2. 帧类型用于表示该帧的类型总共十种类型如下图。最常用的帧类型有HEADERS和DATA帧。

3. 标志位用于携带简单的控制信息。比如

   END_HEADERS 表示头数据结束标志相当于 HTTP/1 里头后的空行“\r\n”

   END_Stream 表示单方向数据发送结束后续不会再有数据帧。

   PRIORITY 表示流的优先级

4. 流标识符占31位表示帧所属的流的ID

如下图为使用WireShark抓包工具抓到的一个HTTP HEADERS帧

前面部分为帧头包含长度、帧类型、标志位以及流ID后面部分为payload部分payload中的内容为HTTP头信息

下面图为gRPC的一次调用

上面方框为请求可以看到一次HTTP请求被分为了两个数据帧分别发送包含HEADERS帧和DATA帧两个帧是处于不同的TCP报文中的

而服务端的响应则被放在了一条TCP报文中包含了三个帧HEADERS、DATA和HEADERS

 

接下来将从源码的角度来将为什么多个流之间是可以并行处理的。

在前面介绍了在go标准库的HTTP/1.1的实现中一个连接只会使用一个goroutine来处理。

而在2.0的实现中会启动两个goroutine一个goroutine用来处理tcp连接上的数据读取另一个goroutine用来处理数据发送调度以及接收到的数据解析如下图所示。

readFrames goroutine在读取到数据帧后会发送到readFrameCh中由另一个goroutine来处理

在处理Frame时会判断Frame的类型如果要处理的帧是HEADERS类型的则会启动一个新的goroutine来负责DATA帧的读取和服务

在每一个连接中都会维护一个流ID和一个http2stream的结构体当读取到HEADERS帧时就会创建出一个该结构或者使用已经创建过的。并且会启动一个goroutine来处理这个流goroutine会从stream的pipeline中读取数据。这个pipeline就相当于一个生产者消费者类型的缓存运行runHandler的goroutine会从pipeline中读取数据直到接收到flag为END_Stream的帧当pipeline中没有数据时就会使用条件变量阻塞当负责从TCP连接中读取数据的goroutine接收到DATA帧时会查询到对应的流的结构然后将数据写入pipeline中然后使用条件变量通知消费者。

从go的HTTP2的代码实现中可以了解到多个流的处理会启动多个goroutine因此多个流之间是可以并行处理的但是通常也会设置最大的并行处理的流的数量。

 

②头部压缩

HTTP/2的另一个特性就是头部压缩。在HTTP/1.1中每次发送请求都会携带大量的字符串类型的header并且在多次请求之间这些大部分的header甚至是相同的。那么为什么不能将那些固定的、不常变化的header使用一个代号标识呢这样的话就可以大大减少传输的数据量。

静态表和动态表

HTTP/2正是这样做的HTTP/2定义了一个静态表和动态表。

静态表中包含了最常用的61个header将它们使用一个字节的数字来表示比如:method:GET这个header就可以用2这个数字来表示。客户端在序列化header时可以从静态表中查询header对应的编号在传输时只需要传输一个字节的编号即可当服务端收到消息后可以根据编号从静态表中查询到对应的header。

同样的为了支持用户自定义的header还引入了一个动态表动态表中的编号占据62及之后的编号。客户端在第一次发送用户自定义的header时依然需要传输整个header然后客户端和服务端双方会将其添加到动态表中那么客户端在后续的通信中只需要传输对应的编号即可。

 

如下面分别为HTTP/2的两次通信抓包可以看到两次通信处于不同的流中。并且第一次通信时content-type: application/grpc这个header占用了13个字节。而在第二次通信时只占用了1个字节。这便是通过动态表进行的优化。

 
哈夫曼编码

在上面的第一次通信的抓包图中我们可以看到content-type: application/grpc在第一次传输时并没有采用明文的字符串进行传输因为HTTP/2还采用了另一个特性来压缩头部就是哈夫曼编码。在传输头部时如果无法使用到静态表则会使用哈夫曼编码对字符串进行压缩。至于哈夫曼编码的原理这里就略过。

 

3.2 protobuf简介

protobuf是一种序列化协议我们可以使用proto文件来定义出message一个message就相当于go的一个结构体。只是message中只包含一些字段每个字段需要给予一个唯一的ID在序列化时就可以使用ID来唯一标识一个字段

syntax = "proto3";

package ecommerce;

message Product {
	string id = 1;
	string name = 2;
	string desc = 3;
	float price = 4;
}

message ProductID {
	string value = 1;
}

在定义了proto文件后就可以编译为go的代码每个message对应于一个go的结构体。并且可以使用proto库来对序列化和反序列化消息。

如下分别使用proto和json来序列化同一个对象查看序列化后的内容和数据长度

func TestPB(t *testing.T) {
	product := &Product{
		Id:    "1",
		Name:  "phone",
		Desc:  "mobile phone",
		Price: 8000,
	}

	// 使用proto进行序列化
	b1, err := proto.Marshal(product)
	if err != nil {
		slog.Error("marshal proto", "error", err)
		return
	}

	// 使用json进行序列化
	b2, err := json.Marshal(product)
	if err != nil {
		slog.Error("marshal json", "error", err)
		return
	}

	fmt.Println(string(b1))
	fmt.Println(string(b2))
	fmt.Println("proto:", len(b1))
	fmt.Println("json:", len(b2))
}

运行程序结果如下

可以看到相比于jsonproto序列化后的数据大小是json的一半。

除此之外proto的序列化速度也要比json、xml等序列化协议要更优秀。

然而由于proto序列化后的数据是二进制的因此它的可读性较差。

 

3.3 基于HTTP/2的gRPC

gRPC是基于HTTP/2的因此gRPC在传输数据时会按照HTTP/2的协议格式。

3.3.1 请求消息

gRPC请求消息用于初始化远程调用。在gRPC中一个客户端请求包含两个部分请求头信息和数据。它们会处于不同的帧中。

当我们调用SayHello()时客户端会通过发送下面的请求头信息来初始化调用

:method = POST      
:scheme = http
:path = /Greeter/SayHello
:authority = 127.0.0.1:8080
content-type = application/grpc
te = trailers
...

在HTTP/2协议中没用了HTTP/1.1的请求行因此原本请求行中的内容都移到了请求头中。

对用上面的请求头含义如下

1. 定义HTTP方法。对于gRPC来说:method头信息始终为POST
2. 定义HTTP模式如果使用TLS则为https否则为http
3. 定义端点路径。对于gRPC来说这个值的构造为/{服务名}/{方法名}
4. 定义目标URI的虚拟主机名
5. 定义content-type对于gRPC来说该值应该以application/grpc开头否则gRPC会给出HTTP状态为415不支持的媒体类型的响应
6. 定义对不兼容代理的检测。在gRPC中这个值必须为trailers

注意名称以":"开头的头信息叫做保留头信息HTTP/2要求保留头信息出现在其它头信息之前。

当完成对服务器端的初始化之后客户端会发送HTTP/2的数据帧到服务端。数据帧中包含了我们的请求数据如果一个帧无法存放所有数据那么它可以跨多个数据帧但是这些数据帧会属于同一个stream而最后一个数据帧会携带flag为END_STREAM的标志。

如下图所示数据部分只由一个帧组成。在帧的payload部分存放的是以长度为前缀的proto消息。

Message Len表示的是序列化后的proto消息的长度后面的数据则为proto buffers数据。

 

3.3.2 响应消息

响应消息由服务端生成服务端根据客户端的请求调用相应的服务然后发送数据响应客户端的请求。与请求类似在大多数场景中响应消息包含三个主要部分响应头信息、以长度为前缀的消息、trailer

如下图为服务端的响应抓包包含了三个帧分别是响应头、消息和trailer

响应头包含下面的内容

:status: 200 OK
content-type: application/grpc

第一个header表示HTTP请求的状态

第二个定义content-type与请求中的类似

 

与请求类似在数据帧中同样也包含以长度为前缀的消息。但是END_STREAM并不会同数据帧一起发送而是作为单独的头信息来发送名为trailer。

最后通过发送trailer来提醒客户端响应消息已经发送。trailer还会携带状态码以及请求的状态信息

grpc-status: 0
grpc-message: xxxx

grpc-status定义了gRPC的状态码。gRPC会使用一组定义良好的状态码。这些状态码的定义可以在gRPC官网文档中找到。

grpc-message则是对错误的描述信息。这是可选的只有在处理请求出现错误时才会进行设置。

 

3.4 gRPC实现架构

如图所示gRPC的实现可以分为多层。

最基础的是gRPC核心层它为其上的层抽象了所有网络操作使得应用程序开发人员可以很容易地通过网络发送RPC调用。

gRPC核心层还提供了对核心功能的扩展包含过滤器、安全、截止时间、取消等功能。

gRPC原生支持C/C++、Go语言和Java语言等通过protoc的编译可以生成各种语言对应的API。应用程序层处理应用程序和数据编码逻辑。

 

4、 gRPC高级功能

4.1 拦截器

在构建gRPC应用程序时无论是客户端应用程序还是服务端应用程序在远程方法执行之前或之后都可能需要执行一些通用逻辑。在gRPC中可以拦截RPC的执行来满足特点的需求比如日志记录、认证、性能度量等这会使用一种名为拦截器的扩展机制。

gRPC提供了简单的API用来在客户端和服务端的gRPC应用程序中实现并安装拦截器。

根据拦截器拦截的RPC调用的类型拦截器可以分为两类对于一元RPC可以使用一元拦截器对于流RPC则可以使用流拦截器

这些拦截器即可安装在客户端也可以安装在服务端。

 

4.1.1 服务端拦截器

当客户端调用gRPC服务的远程方法时通过使用服务端拦截器可以在执行远程方法之前执行一个通用的逻辑如下图所示

在请求时会经过拦截器1、拦截器2、…、拦截器N而在调用服务后则会从拦截器N、…、拦截器2、拦截器1中出来。

一元拦截器

一元拦截器的函数声明如下

type 
UnaryServerInterceptor 
func(ctx context.Context, req any, info *UnaryServerInfo, handler UnaryHandler) (resp any, err error)

接下来通过两个拦截器案例来说明分别是

• Recover拦截器用于捕获panic
• StatisticsServiceTime拦截器用于统计服务运行时间

代码如下

func Recovery() grpc.UnaryServerInterceptor {
	return func(ctx context.Context,
		req any,
		info *grpc.UnaryServerInfo,
		handler grpc.UnaryHandler) (resp any, err error) {
		
        // 在defer func中使用recover来捕获panic
		defer func() {
			if r := recover(); r != nil {
				fmt.Println("recovered: ", r)
			}
		}()

		fmt.Println("------- recovery start -------")
		
        // 前置处理阶段可以在调用对应的服务之前拦截消息
        
        // 服务调用前
        // 调用服务
		resp, err = handler(ctx, req)
		
        // 后置处理阶段可以在这里处理RPC响应
        
        // 服务调用后
		fmt.Println("------- recovery end -------")

		return
	}
}

func StatisticsServiceTime() grpc.UnaryServerInterceptor {
	return func(ctx context.Context,
		req any,
		info *grpc.UnaryServerInfo,
		handler grpc.UnaryHandler) (resp any, err error) {

		fmt.Println("------- statistics time start -------")
	
        // 记录调用服务前时刻
		start := time.Now()
		
        // 调用服务
		resp, err = handler(ctx, req)
		
        // 获取服务执行时间
		used := time.Now().Sub(start)

		fmt.Printf("service name: %s used time: %s\n", info.FullMethod, used.String())

		fmt.Println("------- statistics time end -------")

		return
	}
}

上面是两个拦截器的实现在使用拦截器时需要注册

在创建server时可以使用ChainUnaryInterceptor()来注册多个拦截器形成拦截器链在调用服务前会依次调用这些拦截器。

server := grpc.NewServer(
		grpc.ChainUnaryInterceptor(Recovery(), StatisticsServiceTime()),
	)

程序运行结果如下

 

拦截器的调用原理

在注册服务时实际注册到server中的方法并不是SayHello而是_Greeter_SayHello_Handler这个函数它对我们的SayHello方法又包了一层。因此最终服务端调用的函数为下面的函数

在调用时传入的参数第一个为我们对服务的具体实现对象最后一个参数则是第我们注册的第一个拦截器也就是Recovery在最后会先调用拦截器并且handler传了进入。

但是在_Greeter_SayHello_Handler中传入的拦截器并不完全是Recovery因为它又在拦截器外面包了一层当第一个拦截器被调用时也会传入一个handler如果你认为这个handler就是SayHello那你就错了。

事实上只有所有拦截器调用完毕最后一个拦截器中的handler才是真正的SayHello而其它拦截器中的handler都是它之后的拦截器只是又用闭包将其包装为了func(*ctx* context.Context, *req* any) (any, error)的函数类型

因此当我们注册了多个拦截器时除了最后一个拦截器其它拦截器中传入的handler都是它后面的拦截器。

因此如果我们把Recovery中的handler调用注释掉那么后面的拦截器以及服务方法都不会被调用。

 

流拦截器

服务端流拦截器会拦截gRPC服务端所处理的所有流RPC。

流拦截器包括前置处理阶段和流处理阶段。

流拦截器的函数声明如下

func(srv any, ss ServerStream, info *StreamServerInfo, handler StreamHandler) error

前置处理阶段需要我们实现一个流拦截器而流处理阶段则需要我们对ServerStream进行一层包装重新实现其中的RecvMsg和SendMsg方法代码如下:

// 相当于重写overrideServerSteam中的RecvMsg和SendMsg方法
type wrappedStream struct {
	grpc.ServerStream
}

func newWrappedStream(stream grpc.ServerStream) *wrappedStream {
	return &wrappedStream{
		ServerStream: stream,
	}
}

// 重写SendMsg
func (s *wrappedStream) SendMsg(m any) (err error) {
	fmt.Println("--------[wrappedStream] send msg start ------------")
	err = s.ServerStream.SendMsg(m)
	fmt.Println("--------[wrappedStream] send msg end ------------")
	return err
}

// 重写RecvMsg
func (s *wrappedStream) RecvMsg(m any) (err error) {
	fmt.Println("--------[wrappedStream] recv msg start ------------")
	err = s.ServerStream.RecvMsg(m)
	fmt.Println("--------[wrappedStream] recv msg end ------------")
	return err
}

// 实现流拦截器
func StreamInterceptor() grpc.StreamServerInterceptor {
	return func(srv any,
		ss grpc.ServerStream,
		info *grpc.StreamServerInfo,
		handler grpc.StreamHandler) error {

		fmt.Println("---------- stream interceptor start -----------")
		
        // 构造我们wrappedStream
		stream := newWrappedStream(ss)
		
        // 传入重写的stream
		err := handler(srv, stream)

		fmt.Println("---------- stream interceptor end -----------")

		return err

	}
}

// 注册流拦截器
server := grpc.NewServer(
		grpc.StreamInterceptor(StreamInterceptor()),
	)

 

4.1.2 客户端拦截器

当客户端发起RPC来触发gRPC服务的远程方法时可以在客户端拦截这些RPC如下图所示借助客户端拦截器可以拦截一元RPC和流RPC。

一元拦截器

一元拦截器的函数声明如下

func(ctx context.Context, method string, req, reply any, cc *ClientConn, invoker UnaryInvoker, opts ...CallOption) error

示例代码

func UnaryInterceptor() grpc.UnaryClientInterceptor {
	return func(ctx context.Context,
		method string,
		req, reply any,
		cc *grpc.ClientConn,
		invoker grpc.UnaryInvoker,
		opts ...grpc.CallOption) error {

		// 前置处理阶段
		fmt.Println("method:", method)

		// 调用远程方法
		err := invoker(ctx, method, req, reply, cc, opts...)

		// 后置处理阶段已经拿到了响应
		helloReply, ok := reply.(*greeter.HelloReply)
		if err != nil || !ok {
			return err
		}

		fmt.Println("reply:", helloReply.Data)

		return err

	}
}

// 注册拦截器
conn, err := grpc.Dial("127.0.0.1:8080",
		grpc.WithTransportCredentials(insecure.NewCredentials()),
		grpc.WithBlock(),
		grpc.WithUnaryInterceptor(UnaryInterceptor()),
	)

客户端流拦截器与服务端流拦截器几乎一样因此就不再介绍。

 

4.2 截止时间和取消

在分布式计算中截止时间和超时时间是两个常用的模式。截止时间和超时可以指定客户端应用程序等待RPC完成的时间。加入服务端未能在超时时间内完成服务调用并返回我们就认为服务调用失败。

gRPC客户端在调用服务时可以指定一个超时时间或截止时间并且gRPC也支持在服务调用过程中取消。设置截止时间、超时时间和取消都是通过context包来实现的。

设置超时时间

1. 在服务端的SayHelo方法中设置一个休眠时间来模拟服务运行时间过长

func (g *GreeterImpl) SayHello(ctx context.Context, req *greeter.HelloRequest) (*greeter.HelloReply, error) {
	fmt.Println("req:", req.Data)
	// 休眠三秒
	time.Sleep(time.Second * 3)

	select {
	// 查看服务运行是否超时如果超时直接返回错误
	case <-ctx.Done():
		err := ctx.Err()
		slog.Error("SayHello", "error", err)
		return nil, err
	default:
	}

	return &greeter.HelloReply{Data: "server hello"}, nil
}

2. 客户端在调用时指定超时时间

func CallSayHello(client greeter.GreeterClient) {
    // 设置两秒的超时时间
	ctx, cancelFunc := context.WithTimeout(context.Background(), time.Second*2)
	defer cancelFunc()
	
    // 调用服务传入带有超时的ctx
	reply, err := client.SayHello(ctx, &greeter.HelloRequest{Data: "hello, server"})
	if err != nil {
		fmt.Println("SayHello error:", err)
		return
	}

	fmt.Println("Call rpc success, reply: ", reply.Data)
}

运行结果客户端

服务端

可以看到当发生超时时服务端的ctx是可以收到超时信号的。

由于是远程调用因此客户端的ctx和服务端的ctx肯定不是同一个那么可以进行一个猜出服务端在发送数据时会将超时时间发送给服务端服务端在调用服务时也会创建一个带有超时的ctx。

那么客户端服务调用返回的err是从服务端那得来的吗或者说发生超时时客户端依然会等待服务端返回数据吗

如果仔细一想应该也能想到发生超时时客户端应该不能等待服务端的响应因为发生了超时就被认为是服务调用失败了因此返回的结果也没有意义。
 
接下来通过抓包来探究一下发生超时时客户端和服务端是怎么处理的。

可以看到客户端先后发送了HEADERS帧和DATA帧来调用服务在HEADERS帧中设置了超时时间的头部。

在两秒后也就是发生超时后发送了一个RST_STREAM的帧来终止当前流并且没有收到服务端的响应。

同样的通过使用context.WithCancel也可以实现服务的取消实现的原理也是通过发送RST_STREAM来取消的。

 

4.3 错误处理

在使用本地调用时通常会在函数中返回一个error而在函数调用后会使用errors.Is或errors.As甚至是==来判断错误。

但是在使用远程调用时是无法使用这种方式的。通常会使用一个状态码以及一个描述错误消息的字符串来表示。

当发生错误时gRPC返回一个错误码并附带一条可选的错误消息该消息提供错误条件的更多细节。

状态对象由一个整型的错误码和一条字符串消息组成适用于不同语言的所有gRPC实现。

gRPC已经定义了一组专业的状态码

错误码	值	描述
OK	0	成功
Canceled	1	操作已被取消
Unknown	2	未知错误
InvalidArgument	3	非法参数
DeadlineExceeded	4	超时
NotFound	5	某些请求实体未找到
AlreadyExists	6	客户端试图创建的实体已存在
PermissionDenied	7	调用者没有权限执行特定操作
ResourceExhausted	8	某些资源已耗尽
FailedPrecondition	9	操作被拒绝
Aborted	10	操作被终止
OutOfRange	11	尝试进行的操作超出了合法范围
Unimplemented	12	服务不支持或未实现
Internal	13	内部错误
Unavailable	14	服务当前不可用
DataLoss	15	不可恢复的数据丢失或损坏
Unauthenticated	16	客户端没有进行操作的合法认证凭证

以上的错误码定义在"google.golang.org/grpc/codes"包中可以直接使用。

"google.golang.org/grpc/status"包中定义了Status类型它实现了error并且提供了一些快捷的方法因此在返回错误时可以使用status包。

比如下面的方式

// 服务端使用status包构造error
return nil, status.Error(codes.NotFound, "not found")

// 客户端使用status从error中构造Status结构
s, ok := status.FromError(err)

除了使用上面的方式外我们也可以通过自定义错误的方式来返回更加详细的错误信息。

我们可以使用proto文件来定义错误消息

比如下面定义了一个错误包含ID和Reason两个字段使用protoc编译为.go文件。

syntax = "proto3";

option go_package = "./;errpb";

message InternalError {
  uint32 id = 1;
  string reason = 2;
}

使用案例

服务端在使用时需要创建出一个Status类型的对象然后调用WithDetails()来添加自定义的错误

func (g *GreeterImpl) SayHello(ctx context.Context, req *greeter.HelloRequest) (*greeter.HelloReply, error) {
    // 业务处理...
    
    if err != nil {
        // 服务端设置error detail
		s := status.New(codes.Internal, "internal error")
		e := &errpb.InternalError{
			Id:     520,
			Reason: "I love you",
		}
		ss, _ := s.WithDetails(e)
		return nil, ss.Err()
    }
		
	return &greeter.HelloReply{Data: "server hello"}, nil
}

客户端可以使用Details()来获取到自定义的错误

func CallSayHello(client greeter.GreeterClient) {
	ctx, cancelFunc := context.WithTimeout(context.Background(), time.Second*2)
	defer cancelFunc()

	reply, err := client.SayHello(ctx, &greeter.HelloRequest{Data: "hello, server"})
	if err != nil {
        // 从err中构造Status
		s, ok := status.FromError(err)
		fmt.Println("SayHello error:", err)
		if !ok {
			return
		}
        
        // 使用Details()以及类型断言来获取自定义错误
		for _, d := range s.Details() {
			switch info := d.(type) {
			case *errpb.InternalError:
				fmt.Println(info.Id, info.Reason)
			}
		}

		return
	}

	fmt.Println("Call rpc success, reply: ", reply.Data)
}

4.4 元数据

元数据说白了就是在HTTP的Header中添加和获取我们自己的Headers

4.4.1 客户端发送和接受元数据

在客户端中想要添加header和获取响应中的header需要在调用服务前进行设置

示例代码

func CallSayHello(client greeter.GreeterClient) {
	// 1.客户端设置header
	md := metadata.New(map[string]string{
		"address":    "hangzhou",
		"university": "hdu",
	})

	ctx := metadata.NewOutgoingContext(context.Background(), md)

	// 2.客户端获取header和trailer
	var header, trailer metadata.MD
	opts := []grpc.CallOption{grpc.Header(&header), grpc.Trailer(&trailer)}
	
	reply, err := client.SayHello(ctx, &greeter.HelloRequest{Data: "hello, server"}, opts...)
	if err != nil {
		s, _ := status.FromError(err)
		fmt.Println("SayHello error:", err)
		
		return
	}

	fmt.Println("Call rpc success, reply: ", reply.Data)
}

客户端在请求中添加header需要使用metadata.NewOutgoingContext这个方法来将header附加到ctx中。但是它会覆盖之前设置的header。如果不想覆盖那么可以使用metadata.AppendToOutgoingContext这个方法来追加header。

服务端响应的header有两部分一部分位于响应头部叫做header。另一部分位于响应数据之后叫做trailer。

想要获取他们则需要在调用服务时传入options在调用服务方法时传入grpc.Header来获取header传入grpc.Trailer来获取trailer在服务调用后header和trailer就会被设置。

4.4.2 服务端发送和接受元数据

服务端想要获取和设置对客户端的header在服务方法中就可以。

实例代码如下

func (g *GreeterImpl) SayHello(ctx context.Context, req *greeter.HelloRequest) (*greeter.HelloReply, error) {
	// 服务端获取header
	md, ok := metadata.FromIncomingContext(ctx)
	if ok {
		fmt.Println(md.Get("university"))
		fmt.Println(md.Get("address"))
	}

    // 服务端设置header
	grpc.SetHeader(ctx, metadata.Pairs("server", "my-grpc"))
	// 服务端设置trailer
	grpc.SetTrailer(ctx, metadata.Pairs("trailer", "aabb"))

	return &greeter.HelloReply{Data: "server hello"}, nil
}

在服务端想要获取客户端设置的header可以通过metadata.FromIncomingContext来从ctx中获取。

使用grpc.SetHeader和grpc.SetTrailer来分别设置header和trailer

 

参考资料《gRPC与云原生应用开发》