张欢的个人网站
电子书/Go 语言高级编程

4 / 7

第4章 RPC和Protobuf

约 1.3 万字 · 1 阅读

RPC入门

RPC 是远程过程调用的简称，是分布式系统中不同节点间流行的通信方式。在互联网时代，RPC 已经和 IPC 一样成为一个不可或缺的基础构件。因此 Go 语言的标准库也提供了一个简单的 RPC 实现，我们将以此为入口学习 RPC 的各种用法。

4.1.1 RPC 版 “Hello, World”

Go 语言的 RPC 包的路径为 net/rpc，也就是放在了 net 包目录下面。因此我们可以猜测该 RPC 包是建立在 net 包基础之上的。在第一章 “Hello, World” 革命一节最后，我们基于 http 实现了一个打印例子。下面我们尝试基于 rpc 实现一个类似的例子。

我们先构造一个 HelloService 类型，其中的 Hello 方法用于实现打印功能：

type HelloService struct {}

func (p *HelloService) Hello(request string, reply *string) error {
    *reply = "hello:" + request
    return nil
}

其中 Hello 方法必须满足 Go 语言的 RPC 规则：方法只能有两个可序列化的参数，其中第二个参数是指针类型，并且返回一个 error 类型，同时必须是公开的方法。

然后就可以将 HelloService 类型的对象注册为一个 RPC 服务：

func main() {
    rpc.RegisterName("HelloService", new(HelloService))

    listener, err := net.Listen("tcp", ":1234")
    if err != nil {
        log.Fatal("ListenTCP error:", err)
    }

    conn, err := listener.Accept()
    if err != nil {
        log.Fatal("Accept error:", err)
    }

    rpc.ServeConn(conn)
}

其中 rpc.Register 函数调用会将对象类型中所有满足 RPC 规则的对象方法注册为 RPC 函数，所有注册的方法会放在 “HelloService” 服务空间之下。然后我们建立一个唯一的 TCP 连接，并且通过 rpc.ServeConn 函数在该 TCP 连接上为对方提供 RPC 服务。

下面是客户端请求 HelloService 服务的代码：

func main() {
    client, err := rpc.Dial("tcp", "localhost:1234")
    if err != nil {
        log.Fatal("dialing:", err)
    }

    var reply string
    err = client.Call("HelloService.Hello", "hello", &reply)
    if err != nil {
        log.Fatal(err)
    }

    fmt.Println(reply)
}

首先是通过 rpc.Dial 拨号 RPC 服务，然后通过 client.Call 调用具体的 RPC 方法。在调用 client.Call 时，第一个参数是用点号连接的 RPC 服务名字和方法名字，第二和第三个参数分别我们定义 RPC 方法的两个参数。

由这个例子可以看出 RPC 的使用其实非常简单。

4.1.2 更安全的 RPC 接口

在涉及 RPC 的应用中，作为开发人员一般至少有三种角色：首先是服务端实现 RPC 方法的开发人员，其次是客户端调用 RPC 方法的人员，最后也是最重要的是制定服务端和客户端 RPC 接口规范的设计人员。在前面的例子中我们为了简化将以上几种角色的工作全部放到了一起，虽然看似实现简单，但是不利于后期的维护和工作的切割。

如果要重构 HelloService 服务，第一步需要明确服务的名字和接口：

const HelloServiceName = "path/to/pkg.HelloService"

type HelloServiceInterface interface {
    Hello(request string, reply *string) error
}

func RegisterHelloService(svc HelloServiceInterface) error {
    return rpc.RegisterName(HelloServiceName, svc)
}

我们将 RPC 服务的接口规范分为三个部分：首先是服务的名字，然后是服务要实现的详细方法列表，最后是注册该类型服务的函数。为了避免名字冲突，我们在 RPC 服务的名字中增加了包路径前缀（这个是 RPC 服务抽象的包路径，并非完全等价 Go 语言的包路径）。RegisterHelloService 注册服务时，编译器会要求传入的对象满足 HelloServiceInterface 接口。

在定义了 RPC 服务接口规范之后，客户端就可以根据规范编写 RPC 调用的代码了：

func main() {
    client, err := rpc.Dial("tcp", "localhost:1234")
    if err != nil {
        log.Fatal("dialing:", err)
    }

    var reply string
    err = client.Call(HelloServiceName+".Hello", "hello", &reply)
    if err != nil {
        log.Fatal(err)
    }
    fmt.Println(reply)
}

其中唯一的变化是 client.Call 的第一个参数用 HelloServiceName+".Hello" 代替了 "HelloService.Hello"。然而通过 client.Call 函数调用 RPC 方法依然比较繁琐，同时参数的类型依然无法得到编译器提供的安全保障。

为了简化客户端用户调用 RPC 函数，我们在可以在接口规范部分增加对客户端的简单包装：

type HelloServiceClient struct {
    *rpc.Client
}

var _ HelloServiceInterface = (*HelloServiceClient)(nil)

func DialHelloService(network, address string) (*HelloServiceClient, error) {
    c, err := rpc.Dial(network, address)
    if err != nil {
        return nil, err
    }
    return &HelloServiceClient{Client: c}, nil
}

func (p *HelloServiceClient) Hello(request string, reply *string) error {
    return p.Client.Call(HelloServiceName+".Hello", request, reply)
}

我们在接口规范中针对客户端新增加了 HelloServiceClient 类型，该类型也必须满足 HelloServiceInterface 接口，这样客户端用户就可以直接通过接口对应的方法调用 RPC 函数。同时提供了一个 DialHelloService 方法，直接拨号 HelloService 服务。

基于新的客户端接口，我们可以简化客户端用户的代码：

func main() {
    client, err := DialHelloService("tcp", "localhost:1234")
    if err != nil {
        log.Fatal("dialing:", err)
    }

    var reply string
    err = client.Hello("hello", &reply)
    if err != nil {
        log.Fatal(err)
    }
    fmt.Println(reply)
}

现在客户端用户不用再担心 RPC 方法名字或参数类型不匹配等低级错误的发生。

最后是基于 RPC 接口规范编写真实的服务端代码：

type HelloService struct {}

func (p *HelloService) Hello(request string, reply *string) error {
    *reply = "hello:" + request
    return nil
}

func main() {
    RegisterHelloService(new(HelloService))

    listener, err := net.Listen("tcp", ":1234")
    if err != nil {
        log.Fatal("ListenTCP error:", err)
    }

    for {
        conn, err := listener.Accept()
        if err != nil {
            log.Fatal("Accept error:", err)
        }

        go rpc.ServeConn(conn)
    }
}

在新的 RPC 服务端实现中，我们用 RegisterHelloService 函数来注册函数，这样不仅可以避免命名服务名称的工作，同时也保证了传入的服务对象满足了 RPC 接口的定义。最后我们新的服务改为支持多个 TCP 连接，然后为每个 TCP 连接提供 RPC 服务。

4.1.3 跨语言的 RPC

标准库的 RPC 默认采用 Go 语言特有的 gob 编码，因此从其它语言调用 Go 语言实现的 RPC 服务将比较困难。在互联网的微服务时代，每个 RPC 以及服务的使用者都可能采用不同的编程语言，因此跨语言是互联网时代 RPC 的一个首要条件。得益于 RPC 的框架设计，Go 语言的 RPC 其实也是很容易实现跨语言支持的。

Go 语言的 RPC 框架有两个比较有特色的设计：一个是 RPC 数据打包时可以通过插件实现自定义的编码和解码；另一个是 RPC 建立在抽象的 io.ReadWriteCloser 接口之上的，我们可以将 RPC 架设在不同的通讯协议之上。这里我们将尝试通过官方自带的 net/rpc/jsonrpc 扩展实现一个跨语言的 RPC。

首先是基于 json 编码重新实现 RPC 服务：

func main() {
    rpc.RegisterName("HelloService", new(HelloService))

    listener, err := net.Listen("tcp", ":1234")
    if err != nil {
        log.Fatal("ListenTCP error:", err)
    }

    for {
        conn, err := listener.Accept()
        if err != nil {
            log.Fatal("Accept error:", err)
        }

        go rpc.ServeCodec(jsonrpc.NewServerCodec(conn))
    }
}

代码中最大的变化是用 rpc.ServeCodec 函数替代了 rpc.ServeConn 函数，传入的参数是针对服务端的 json 编解码器。

然后是实现 json 版本的客户端：

func main() {
    conn, err := net.Dial("tcp", "localhost:1234")
    if err != nil {
        log.Fatal("net.Dial:", err)
    }

    client := rpc.NewClientWithCodec(jsonrpc.NewClientCodec(conn))

    var reply string
    err = client.Call("HelloService.Hello", "hello", &reply)
    if err != nil {
        log.Fatal(err)
    }

    fmt.Println(reply)
}

先手工调用 net.Dial 函数建立 TCP 连接，然后基于该连接建立针对客户端的 json 编解码器。

在确保客户端可以正常调用 RPC 服务的方法之后，我们用一个普通的 TCP 服务代替 Go 语言版本的 RPC 服务，这样可以查看客户端调用时发送的数据格式。比如通过 nc 命令 nc -l 1234 在同样的端口启动一个 TCP 服务。然后再次执行一次 RPC 调用将会发现 nc 输出了以下的信息：

{"method":"HelloService.Hello","params":["hello"],"id":0}

这是一个 json 编码的数据，其中 method 部分对应要调用的 rpc 服务和方法组合成的名字，params 部分的第一个元素为参数，id 是由调用端维护的一个唯一的调用编号。

请求的 json 数据对象在内部对应两个结构体：客户端是 clientRequest，服务端是 serverRequest。clientRequest 和 serverRequest 结构体的内容基本是一致的：

type clientRequest struct {
    Method string         `json:"method"`
    Params [1]interface{} `json:"params"`
    Id     uint64         `json:"id"`
}

type serverRequest struct {
    Method string           `json:"method"`
    Params *json.RawMessage `json:"params"`
    Id     *json.RawMessage `json:"id"`
}

在获取到 RPC 调用对应的 json 数据后，我们可以通过直接向架设了 RPC 服务的 TCP 服务器发送 json 数据模拟 RPC 方法调用：

$ echo -e '{"method":"HelloService.Hello","params":["hello"],"id":1}' | nc localhost 1234

返回的结果也是一个 json 格式的数据：

{"id":1,"result":"hello:hello","error":null}

其中 id 对应输入的 id 参数，result 为返回的结果，error 部分在出问题时表示错误信息。对于顺序调用来说，id 不是必须的。但是 Go 语言的 RPC 框架支持异步调用，当返回结果的顺序和调用的顺序不一致时，可以通过 id 来识别对应的调用。

返回的 json 数据也是对应内部的两个结构体：客户端是 clientResponse，服务端是 serverResponse。两个结构体的内容同样也是类似的：

type clientResponse struct {
    Id     uint64           `json:"id"`
    Result *json.RawMessage `json:"result"`
    Error  interface{}      `json:"error"`
}

type serverResponse struct {
    Id     *json.RawMessage `json:"id"`
    Result interface{}      `json:"result"`
    Error  interface{}      `json:"error"`
}

因此无论采用何种语言，只要遵循同样的 json 结构，以同样的流程就可以和 Go 语言编写的 RPC 服务进行通信。这样我们就实现了跨语言的 RPC。

4.1.4 Http 上的 RPC

Go 语言内在的 RPC 框架已经支持在 Http 协议上提供 RPC 服务。但是框架的 http 服务同样采用了内置的 gob 协议，并且没有提供采用其它协议的接口，因此从其它语言依然无法访问的。在前面的例子中，我们已经实现了在 TCP 协议之上运行 jsonrpc 服务，并且通过 nc 命令行工具成功实现了 RPC 方法调用。现在我们尝试在 http 协议上提供 jsonrpc 服务。

新的 RPC 服务其实是一个类似 REST 规范的接口，接收请求并采用相应处理流程：

func main() {
    rpc.RegisterName("HelloService", new(HelloService))

    http.HandleFunc("/jsonrpc", func(w http.ResponseWriter, r *http.Request) {
        var conn io.ReadWriteCloser = struct {
            io.Writer
            io.ReadCloser
        }{
            ReadCloser: r.Body,
            Writer:     w,
        }

        rpc.ServeRequest(jsonrpc.NewServerCodec(conn))
    })

    http.ListenAndServe(":1234", nil)
}

RPC 的服务架设在 “/jsonrpc” 路径，在处理函数中基于 http.ResponseWriter 和 http.Request 类型的参数构造一个 io.ReadWriteCloser 类型的 conn 通道。然后基于 conn 构建针对服务端的 json 编码解码器。最后通过 rpc.ServeRequest 函数为每次请求处理一次 RPC 方法调用。

模拟一次 RPC 调用的过程就是向该连接发送一个 json 字符串：

$ curl localhost:1234/jsonrpc -X POST \
    --data '{"method":"HelloService.Hello","params":["hello"],"id":0}'

返回的结果依然是 json 字符串：

{"id":0,"result":"hello:hello","error":null}

这样就可以很方便地从不同语言中访问 RPC 服务了。

Protobuf

Protobuf 是 Protocol Buffers 的简称，它是 Google 公司开发的一种数据描述语言，并于 2008 年对外开源。Protobuf 刚开源时的定位类似于 XML、JSON 等数据描述语言，通过附带工具生成代码并实现将结构化数据序列化的功能。但是我们更关注的是 Protobuf 作为接口规范的描述语言，可以作为设计安全的跨语言 PRC 接口的基础工具。

4.2.1 Protobuf 入门

对于没有用过 Protobuf 的读者，建议先从官网了解下基本用法。这里我们尝试将 Protobuf 和 RPC 结合在一起使用，通过 Protobuf 来最终保证 RPC 的接口规范和安全。Protobuf 中最基本的数据单元是 message，是类似 Go 语言中结构体的存在。在 message 中可以嵌套 message 或其它的基础数据类型的成员。

首先创建 hello.proto 文件，其中包装 HelloService 服务中用到的字符串类型：

syntax = "proto3";

package main;

message String {
    string value = 1;
}

开头的 syntax 语句表示采用 proto3 的语法。第三版的 Protobuf 对语言进行了提炼简化，所有成员均采用类似 Go 语言中的零值初始化（不再支持自定义默认值），因此消息成员也不再需要支持 required 特性。然后 package 指令指明当前是 main 包（这样可以和 Go 的包名保持一致，简化例子代码），当然用户也可以针对不同的语言定制对应的包路径和名称。最后 message 关键字定义一个新的 String 类型，在最终生成的 Go 语言代码中对应一个 String 结构体。String 类型中只有一个字符串类型的 value 成员，该成员编码时用 1 编号代替名字。

在 XML 或 JSON 等数据描述语言中，一般通过成员的名字来绑定对应的数据。但是 Protobuf 编码却是通过成员的唯一编号来绑定对应的数据，因此 Protobuf 编码后数据的体积会比较小，但是也非常不便于人类查阅。我们目前并不关注 Protobuf 的编码技术，最终生成的 Go 结构体可以自由采用 JSON 或 gob 等编码格式，因此大家可以暂时忽略 Protobuf 的成员编码部分。

Protobuf 核心的工具集是 C++ 语言开发的，在官方的 protoc 编译器中并不支持 Go 语言。要想基于上面的 hello.proto 文件生成相应的 Go 代码，需要安装相应的插件。首先是安装官方的 protoc 工具，可以从 https://github.com/google/protobuf/releases 下载。然后是安装针对 Go 语言的代码生成插件，可以通过 go get github.com/golang/protobuf/protoc-gen-go 命令安装。

然后通过以下命令生成相应的 Go 代码：

$ protoc --go_out=. hello.proto

其中 go_out 参数告知 protoc 编译器去加载对应的 protoc-gen-go 工具，然后通过该工具生成代码，生成代码放到当前目录。最后是一系列要处理的 protobuf 文件的列表。

这里只生成了一个 hello.pb.go 文件，其中 String 结构体内容如下：

type String struct {
    Value string `protobuf:"bytes,1,opt,name=value" json:"value,omitempty"`
}

func (m *String) Reset()         { *m = String{} }
func (m *String) String() string { return proto.CompactTextString(m) }
func (*String) ProtoMessage()    {}
func (*String) Descriptor() ([]byte, []int) {
    return fileDescriptor_hello_069698f99dd8f029, []int{0}
}

func (m *String) GetValue() string {
    if m != nil {
        return m.Value
    }
    return ""
}

生成的结构体中还会包含一些以 XXX_ 为名字前缀的成员，我们已经隐藏了这些成员。同时 String 类型还自动生成了一组方法，其中 ProtoMessage 方法表示这是一个实现了 proto.Message 接口的方法。此外 Protobuf 还为每个成员生成了一个 Get 方法，Get 方法不仅可以处理空指针类型，而且可以和 Protobuf 第二版的方法保持一致（第二版的自定义默认值特性依赖这类方法）。

基于新的 String 类型，我们可以重新实现 HelloService 服务：

type HelloService struct{}

func (p *HelloService) Hello(request *String, reply *String) error {
    reply.Value = "hello:" + request.GetValue()
    return nil
}

其中 Hello 方法的输入参数和输出的参数均改用 Protobuf 定义的 String 类型表示。因为新的输入参数为结构体类型，因此改用指针类型作为输入参数，函数的内部代码同时也做了相应的调整。

至此，我们初步实现了 Protobuf 和 RPC 组合工作。在启动 RPC 服务时，我们依然可以选择默认的 gob 或手工指定 json 编码，甚至可以重新基于 protobuf 编码实现一个插件。虽然做了这么多工作，但是似乎并没有看到什么收益！

回顾第一章中更安全的 RPC 接口部分的内容，当时我们花费了极大的力气去给 RPC 服务增加安全的保障。最终得到的更安全的 RPC 接口的代码本身就非常繁琐的使用手工维护，同时全部安全相关的代码只适用于 Go 语言环境！既然使用了 Protobuf 定义的输入和输出参数，那么 RPC 服务接口是否也可以通过 Protobuf 定义呢？其实用 Protobuf 定义语言无关的 RPC 服务接口才是它真正的价值所在！

下面更新 hello.proto 文件，通过 Protobuf 来定义 HelloService 服务：

service HelloService {
    rpc Hello (String) returns (String);
}

但是重新生成的 Go 代码并没有发生变化。这是因为世界上的 RPC 实现有千万种，protoc 编译器并不知道该如何为 HelloService 服务生成代码。

不过在 protoc-gen-go 内部已经集成了一个名字为 grpc 的插件，可以针对 gRPC 生成代码：

$ protoc --go_out=plugins=grpc:. hello.proto

在生成的代码中多了一些类似 HelloServiceServer、HelloServiceClient 的新类型。这些类型是为 gRPC 服务的，并不符合我们的 RPC 要求。

不过 gRPC 插件为我们提供了改进的思路，下面我们将探索如何为我们的 RPC 生成安全的代码。

4.2.2 定制代码生成插件

Protobuf 的 protoc 编译器是通过插件机制实现对不同语言的支持。比如 protoc 命令出现 --xxx_out 格式的参数，那么 protoc 将首先查询是否有内置的 xxx 插件，如果没有内置的 xxx 插件那么将继续查询当前系统中是否存在 protoc-gen-xxx 命名的可执行程序，最终通过查询到的插件生成代码。对于 Go 语言的 protoc-gen-go 插件来说，里面又实现了一层静态插件系统。比如 protoc-gen-go 内置了一个 gRPC 插件，用户可以通过 --go_out=plugins=grpc 参数来生成 gRPC 相关代码，否则只会针对 message 生成相关代码。

参考 gRPC 插件的代码，可以发现 generator.RegisterPlugin 函数可以用来注册插件。插件是一个 generator.Plugin 接口：

// A Plugin provides functionality to add to the output during
// Go code generation, such as to produce RPC stubs.
type Plugin interface {
    // Name identifies the plugin.
    Name() string
    // Init is called once after data structures are built but before
    // code generation begins.
    Init(g *Generator)
    // Generate produces the code generated by the plugin for this file,
    // except for the imports, by calling the generator's methods P, In,
    // and Out.
    Generate(file *FileDescriptor)
    // GenerateImports produces the import declarations for this file.
    // It is called after Generate.
    GenerateImports(file *FileDescriptor)
}

其中 Name 方法返回插件的名字，这是 Go 语言的 Protobuf 实现的插件体系，和 protoc 插件的名字并无关系。然后 Init 函数是通过 g 参数对插件进行初始化，g 参数中包含 Proto 文件的所有信息。最后的 Generate 和 GenerateImports 方法用于生成主体代码和对应的导入包代码。

因此我们可以设计一个 netrpcPlugin 插件，用于为标准库的 RPC 框架生成代码：

import (
    "github.com/golang/protobuf/protoc-gen-go/generator"
)

type netrpcPlugin struct{*generator.Generator}

func (p *netrpcPlugin) Name() string                { return "netrpc" }
func (p *netrpcPlugin) Init(g *generator.Generator) { p.Generator = g }

func (p *netrpcPlugin) GenerateImports(file *generator.FileDescriptor) {
    if len(file.Service) > 0 {
        p.genImportCode(file)
    }
}

func (p *netrpcPlugin) Generate(file *generator.FileDescriptor) {
    for _, svc := range file.Service {
        p.genServiceCode(svc)
    }
}

首先 Name 方法返回插件的名字。netrpcPlugin 插件内置了一个匿名的 *generator.Generator 成员，然后在 Init 初始化的时候用参数 g 进行初始化，因此插件是从 g 参数对象继承了全部的公有方法。其中 GenerateImports 方法调用自定义的 genImportCode 函数生成导入代码。Generate 方法调用自定义的 genServiceCode 方法生成每个服务的代码。

目前，自定义的 genImportCode 和 genServiceCode 方法只是输出一行简单的注释：

func (p *netrpcPlugin) genImportCode(file *generator.FileDescriptor) {
    p.P("// TODO: import code")
}

func (p *netrpcPlugin) genServiceCode(svc *descriptor.ServiceDescriptorProto) {
    p.P("// TODO: service code, Name =" + svc.GetName())
}

要使用该插件需要先通过 generator.RegisterPlugin 函数注册插件，可以在 init 函数中完成：

func init() {
    generator.RegisterPlugin(new(netrpcPlugin))
}

因为 Go 语言的包只能静态导入，我们无法向已经安装的 protoc-gen-go 添加我们新编写的插件。我们将重新克隆 protoc-gen-go 对应的 main 函数：

package main

import (
    "io/ioutil"
    "os"

    "github.com/golang/protobuf/proto"
    "github.com/golang/protobuf/protoc-gen-go/generator"
)

func main() {
    g := generator.New()

    data, err := ioutil.ReadAll(os.Stdin)
    if err != nil {
        g.Error(err, "reading input")
    }

    if err := proto.Unmarshal(data, g.Request); err != nil {
        g.Error(err, "parsing input proto")
    }

    if len(g.Request.FileToGenerate) == 0 {
        g.Fail("no files to generate")
    }

    g.CommandLineParameters(g.Request.GetParameter())

    // Create a wrapped version of the Descriptors and EnumDescriptors that
    // point to the file that defines them.
    g.WrapTypes()

    g.SetPackageNames()
    g.BuildTypeNameMap()

    g.GenerateAllFiles()

    // Send back the results.
    data, err = proto.Marshal(g.Response)
    if err != nil {
        g.Error(err, "failed to marshal output proto")
    }
    _, err = os.Stdout.Write(data)
    if err != nil {
        g.Error(err, "failed to write output proto")
    }
}

为了避免对 protoc-gen-go 插件造成干扰，我们将我们的可执行程序命名为 protoc-gen-go-netrpc，表示包含了 netrpc 插件。然后用以下命令重新编译 hello.proto 文件：

$ protoc --go-netrpc_out=plugins=netrpc:. hello.proto

其中 --go-netrpc_out 参数告知 protoc 编译器加载名为 protoc-gen-go-netrpc 的插件，插件中的 plugins=netrpc 指示启用内部唯一的名为 netrpc 的 netrpcPlugin 插件。在新生成的 hello.pb.go 文件中将包含增加的注释代码。

至此，手工定制的 Protobuf 代码生成插件终于可以工作了。

4.2.3 自动生成完整的 RPC 代码

在前面的例子中我们已经构建了最小化的 netrpcPlugin 插件，并且通过克隆 protoc-gen-go 的主程序创建了新的 protoc-gen-go-netrpc 的插件程序。现在开始继续完善 netrpcPlugin 插件，最终目标是生成 RPC 安全接口。

首先是自定义的 genImportCode 方法中生成导入包的代码：

func (p *netrpcPlugin) genImportCode(file *generator.FileDescriptor) {
    p.P(`import "net/rpc"`)
}

然后要在自定义的 genServiceCode 方法中为每个服务生成相关的代码。分析可以发现每个服务最重要的是服务的名字，然后每个服务有一组方法。而对于服务定义的方法，最重要的是方法的名字，还有输入参数和输出参数类型的名字。

为此我们定义了一个 ServiceSpec 类型，用于描述服务的元信息：

type ServiceSpec struct {
    ServiceName string
    MethodList  []ServiceMethodSpec
}

type ServiceMethodSpec struct {
    MethodName     string
    InputTypeName  string
    OutputTypeName string
}

然后我们新建一个 buildServiceSpec 方法用来解析每个服务的 ServiceSpec 元信息：

func (p *netrpcPlugin) buildServiceSpec(
    svc *descriptor.ServiceDescriptorProto,
) *ServiceSpec {
    spec := &ServiceSpec{
        ServiceName: generator.CamelCase(svc.GetName()),
    }

    for _, m := range svc.Method {
        spec.MethodList = append(spec.MethodList, ServiceMethodSpec{
            MethodName:     generator.CamelCase(m.GetName()),
            InputTypeName:  p.TypeName(p.ObjectNamed(m.GetInputType())),
            OutputTypeName: p.TypeName(p.ObjectNamed(m.GetOutputType())),
        })
    }

    return spec
}

其中输入参数是 *descriptor.ServiceDescriptorProto 类型，完整描述了一个服务的所有信息。然后通过 svc.GetName() 就可以获取 Protobuf 文件中定义的服务的名字。Protobuf 文件中的名字转为 Go 语言的名字后，需要通过 generator.CamelCase 函数进行一次转换。类似的，在 for 循环中我们通过 m.GetName() 获取方法的名字，然后再转为 Go 语言中对应的名字。比较复杂的是对输入和输出参数名字的解析：首先需要通过 m.GetInputType() 获取输入参数的类型，然后通过 p.ObjectNamed 获取类型对应的类对象信息，最后获取类对象的名字。

然后我们就可以基于 buildServiceSpec 方法构造的服务的元信息生成服务的代码：

func (p *netrpcPlugin) genServiceCode(svc *descriptor.ServiceDescriptorProto) {
    spec := p.buildServiceSpec(svc)

    var buf bytes.Buffer
    t := template.Must(template.New("").Parse(tmplService))
    err := t.Execute(&buf, spec)
    if err != nil {
        log.Fatal(err)
    }

    p.P(buf.String())
}

为了便于维护，我们基于 Go 语言的模板来生成服务代码，其中 tmplService 是服务的模板。

在编写模板之前，我们先查看下我们期望生成的最终代码大概是什么样子：

type HelloServiceInterface interface {
    Hello(in String, out *String) error
}

func RegisterHelloService(srv *rpc.Server, x HelloService) error {
    if err := srv.RegisterName("HelloService", x); err != nil {
        return err
    }
    return nil
}

type HelloServiceClient struct {
    *rpc.Client
}

var _ HelloServiceInterface = (*HelloServiceClient)(nil)

func DialHelloService(network, address string) (*HelloServiceClient, error) {
    c, err := rpc.Dial(network, address)
    if err != nil {
        return nil, err
    }
    return &HelloServiceClient{Client: c}, nil
}

func (p *HelloServiceClient) Hello(in String, out *String) error {
    return p.Client.Call("HelloService.Hello", in, out)
}

其中 HelloService 是服务名字，同时还有一系列的方法相关的名字。

参考最终要生成的代码可以构建如下模板：

const tmplService = `
{{$root := .}}

type {{.ServiceName}}Interface interface {
    {{- range $_, $m := .MethodList}}
    {{$m.MethodName}}(*{{$m.InputTypeName}}, *{{$m.OutputTypeName}}) error
    {{- end}}
}

func Register{{.ServiceName}}(
    srv *rpc.Server, x {{.ServiceName}}Interface,
) error {
    if err := srv.RegisterName("{{.ServiceName}}", x); err != nil {
        return err
    }
    return nil
}

type {{.ServiceName}}Client struct {
    *rpc.Client
}

var _ {{.ServiceName}}Interface = (*{{.ServiceName}}Client)(nil)

func Dial{{.ServiceName}}(network, address string) (
    *{{.ServiceName}}Client, error,
) {
    c, err := rpc.Dial(network, address)
    if err != nil {
        return nil, err
    }
    return &{{.ServiceName}}Client{Client: c}, nil
}

{{range $_, $m := .MethodList}}
func (p *{{$root.ServiceName}}Client) {{$m.MethodName}}(
    in *{{$m.InputTypeName}}, out *{{$m.OutputTypeName}},
) error {
    return p.Client.Call("{{$root.ServiceName}}.{{$m.MethodName}}", in, out)
}
{{end}}
`

当 Protobuf 的插件定制工作完成后，每次 hello.proto 文件中 RPC 服务的变化都可以自动生成代码。也可以通过更新插件的模板，调整或增加生成代码的内容。在掌握了定制 Protobuf 插件技术后，你将彻底拥有这个技术。

玩转RPC

在不同的场景中 RPC 有着不同的需求，因此开源的社区就诞生了各种 RPC 框架。本节我们将尝试 Go 内置 RPC 框架在一些比较特殊场景的用法。

4.3.1 客户端 RPC 的实现原理

Go 语言的 RPC 库最简单的使用方式是通过 Client.Call 方法进行同步阻塞调用，该方法的实现如下：

func (client *Client) Call(
    serviceMethod string, args interface{},
    reply interface{},
) error {
    call := <-client.Go(serviceMethod, args, reply, make(chan *Call, 1)).Done
    return call.Error
}

首先通过 Client.Go 方法进行一次异步调用，返回一个表示这次调用的 Call 结构体。然后等待 Call 结构体的 Done 管道返回调用结果。

我们也可以通过 Client.Go 方法异步调用前面的 HelloService 服务：

func doClientWork(client *rpc.Client) {
    helloCall := client.Go("HelloService.Hello", "hello", new(string), nil)

    // do some thing

    helloCall = <-helloCall.Done
    if err := helloCall.Error; err != nil {
        log.Fatal(err)
    }

    args := helloCall.Args.(string)
    reply := helloCall.Reply.(*string)
    fmt.Println(args, *reply)
}

在异步调用命令发出后，一般会执行其他的任务，因此异步调用的输入参数和返回值可以通过返回的 Call 变量进行获取。

执行异步调用的 Client.Go 方法实现如下：

func (client *Client) Go(
    serviceMethod string, args interface{},
    reply interface{},
    done chan *Call,
) *Call {
    call := new(Call)
    call.ServiceMethod = serviceMethod
    call.Args = args
    call.Reply = reply
    call.Done = make(chan *Call, 10) // buffered.

    client.send(call)
    return call
}

首先是构造一个表示当前调用的 call 变量，然后通过 client.send 将 call 的完整参数发送到 RPC 框架。client.send 方法调用是线程安全的，因此可以从多个 Goroutine 同时向同一个 RPC 连接发送调用指令。

当调用完成或者发生错误时，将调用 call.done 方法通知完成：

func (call *Call) done() {
    select {
    case call.Done <- call:
        // ok
    default:
        // We don't want to block here. It is the caller's responsibility to make
        // sure the channel has enough buffer space. See comment in Go().
    }
}

从 Call.done 方法的实现可以得知 call.Done 管道会将处理后的 call 返回。

4.3.2 基于 RPC 实现 Watch 功能

在很多系统中都提供了 Watch 监视功能的接口，当系统满足某种条件时 Watch 方法返回监控的结果。在这里我们可以尝试通过 RPC 框架实现一个基本的 Watch 功能。如前文所描述，因为 client.send 是线程安全的，我们也可以通过在不同的 Goroutine 中同时并发阻塞调用 RPC 方法。通过在一个独立的 Goroutine 中调用 Watch 函数进行监控。

为了便于演示，我们计划通过 RPC 构造一个简单的内存 KV 数据库。首先定义服务如下：

type KVStoreService struct {
    m      map[string]string
    filter map[string]func(key string)
    mu     sync.Mutex
}

func NewKVStoreService() *KVStoreService {
    return &KVStoreService{
        m:      make(map[string]string),
        filter: make(map[string]func(key string)),
    }
}

其中 m 成员是一个 map 类型，用于存储 KV 数据。filter 成员对应每个 Watch 调用时定义的过滤器函数列表。而 mu 成员为互斥锁，用于在多个 Goroutine 访问或修改时对其它成员提供保护。

然后就是 Get 和 Set 方法：

func (p *KVStoreService) Get(key string, value *string) error {
    p.mu.Lock()
    defer p.mu.Unlock()

    if v, ok := p.m[key]; ok {
        *value = v
        return nil
    }

    return fmt.Errorf("not found")
}

func (p *KVStoreService) Set(kv [2]string, reply *struct{}) error {
    p.mu.Lock()
    defer p.mu.Unlock()

    key, value := kv[0], kv[1]

    if oldValue := p.m[key]; oldValue != value {
        for _, fn := range p.filter {
            fn(key)
        }
    }

    p.m[key] = value
    return nil
}

在 Set 方法中，输入参数是 key 和 value 组成的数组，用一个匿名的空结构体表示忽略了输出参数。当修改某个 key 对应的值时会调用每一个过滤器函数。

而过滤器列表在 Watch 方法中提供：

func (p *KVStoreService) Watch(timeoutSecond int, keyChanged *string) error {
    id := fmt.Sprintf("watch-%s-%03d", time.Now(), rand.Int())
    ch := make(chan string, 10) // buffered

    p.mu.Lock()
    p.filter[id] = func(key string) { ch <- key }
    p.mu.Unlock()

    select {
    case <-time.After(time.Duration(timeoutSecond) * time.Second):
        return fmt.Errorf("timeout")
    case key := <-ch:
        *keyChanged = key
        return nil
    }

    return nil
}

Watch 方法的输入参数是超时的秒数。当有 key 变化时将 key 作为返回值返回。如果超过时间后依然没有 key 被修改，则返回超时的错误。Watch 的实现中，用唯一的 id 表示每个 Watch 调用，然后根据 id 将自身对应的过滤器函数注册到 p.filter 列表。

KVStoreService 服务的注册和启动过程我们不再赘述。下面我们看看如何从客户端使用 Watch 方法：

func doClientWork(client *rpc.Client) {
    go func() {
        var keyChanged string
        err := client.Call("KVStoreService.Watch", 30, &keyChanged)
        if err != nil {
            log.Fatal(err)
        }
        fmt.Println("watch:", keyChanged)
    } ()

    err := client.Call(
        "KVStoreService.Set", [2]string{"abc", "abc-value"},
        new(struct{}),
    )
    if err != nil {
        log.Fatal(err)
    }

    time.Sleep(time.Second*3)
}

首先启动一个独立的 Goroutine 监控 key 的变化。同步的 watch 调用会阻塞，直到有 key 发生变化或者超时。然后在通过 Set 方法修改 KV 值时，服务器会将变化的 key 通过 Watch 方法返回。这样我们就可以实现对某些状态的监控。

4.3.3 反向 RPC

通常的 RPC 是基于 C/S 结构，RPC 的服务端对应网络的服务器，RPC 的客户端也对应网络客户端。但是对于一些特殊场景，比如在公司内网提供一个 RPC 服务，但是在外网无法连接到内网的服务器。这种时候我们可以参考类似反向代理的技术，首先从内网主动连接到外网的 TCP 服务器，然后基于 TCP 连接向外网提供 RPC 服务。

以下是启动反向 RPC 服务的代码：

func main() {
    rpc.Register(new(HelloService))

    for {
        conn, _ := net.Dial("tcp", "localhost:1234")
        if conn == nil {
            time.Sleep(time.Second)
            continue
        }

        rpc.ServeConn(conn)
        conn.Close()
    }
}

反向 RPC 的内网服务将不再主动提供 TCP 监听服务，而是首先主动连接到对方的 TCP 服务器。然后基于每个建立的 TCP 连接向对方提供 RPC 服务。

而 RPC 客户端则需要在一个公共的地址提供一个 TCP 服务，用于接受 RPC 服务器的连接请求：

func main() {
    listener, err := net.Listen("tcp", ":1234")
    if err != nil {
        log.Fatal("ListenTCP error:", err)
    }

    clientChan := make(chan *rpc.Client)

    go func() {
        for {
            conn, err := listener.Accept()
            if err != nil {
                log.Fatal("Accept error:", err)
            }

            clientChan <- rpc.NewClient(conn)
        }
    }()

    doClientWork(clientChan)
}

当每个连接建立后，基于网络连接构造 RPC 客户端对象并发送到 clientChan 管道。

客户端执行 RPC 调用的操作在 doClientWork 函数完成：

func doClientWork(clientChan <-chan *rpc.Client) {
    client := <-clientChan
    defer client.Close()

    var reply string
    err := client.Call("HelloService.Hello", "hello", &reply)
    if err != nil {
        log.Fatal(err)
    }

    fmt.Println(reply)
}

首先从管道去取一个 RPC 客户端对象，并且通过 defer 语句指定在函数退出前关闭客户端。然后是执行正常的 RPC 调用。

4.3.4 上下文信息

基于上下文我们可以针对不同客户端提供定制化的 RPC 服务。我们可以通过为每个连接提供独立的 RPC 服务来实现对上下文特性的支持。

首先改造 HelloService，里面增加了对应连接的 conn 成员：

type HelloService struct {
    conn net.Conn
}

然后为每个连接启动独立的 RPC 服务：

func main() {
    listener, err := net.Listen("tcp", ":1234")
    if err != nil {
        log.Fatal("ListenTCP error:", err)
    }

    for {
        conn, err := listener.Accept()
        if err != nil {
            log.Fatal("Accept error:", err)
        }

        go func() {
            defer conn.Close()

            p := rpc.NewServer()
            p.Register(&HelloService{conn: conn})
            p.ServeConn(conn)
        } ()
    }
}

Hello 方法中就可以根据 conn 成员识别不同连接的 RPC 调用：

func (p *HelloService) Hello(request string, reply *string) error {
    *reply = "hello:" + request + ", from" + p.conn.RemoteAddr().String()
    return nil
}

基于上下文信息，我们可以方便地为 RPC 服务增加简单的登陆状态的验证：

type HelloService struct {
    conn    net.Conn
    isLogin bool
}

func (p *HelloService) Login(request string, reply *string) error {
    if request != "user:password" {
        return fmt.Errorf("auth failed")
    }
    log.Println("login ok")
    p.isLogin = true
    return nil
}

func (p *HelloService) Hello(request string, reply *string) error {
    if !p.isLogin {
        return fmt.Errorf("please login")
    }
    *reply = "hello:" + request + ", from" + p.conn.RemoteAddr().String()
    return nil
}

这样可以要求在客户端连接 RPC 服务时，首先要执行登陆操作，登陆成功后才能正常执行其他的服务。

gRPC入门

gRPC 是 Google 公司基于 Protobuf 开发的跨语言的开源 RPC 框架。gRPC 基于 HTTP/2 协议设计，可以基于一个 HTTP/2 连接提供多个服务，对于移动设备更加友好。本节将讲述 gRPC 的简单用法。

4.4.1 gRPC 技术栈

Go 语言的 gRPC 技术栈如图 4-1 所示：

图 4-1 gRPC 技术栈

最底层为 TCP 或 Unix Socket 协议，在此之上是 HTTP/2 协议的实现，然后在 HTTP/2 协议之上又构建了针对 Go 语言的 gRPC 核心库。应用程序通过 gRPC 插件生产的 Stub 代码和 gRPC 核心库通信，也可以直接和 gRPC 核心库通信。

4.4.2 gRPC 入门

如果从 Protobuf 的角度看，gRPC 只不过是一个针对 service 接口生成代码的生成器。我们在本章的第二节中手工实现了一个简单的 Protobuf 代码生成器插件，只不过当时生成的代码是适配标准库的 RPC 框架的。现在我们将学习 gRPC 的用法。

创建 hello.proto 文件，定义 HelloService 接口：

syntax = "proto3";

package main;

message String {
    string value = 1;
}

service HelloService {
    rpc Hello (String) returns (String);
}

使用 protoc-gen-go 内置的 gRPC 插件生成 gRPC 代码：

$ protoc --go_out=plugins=grpc:. hello.proto

gRPC 插件会为服务端和客户端生成不同的接口：

type HelloServiceServer interface {
    Hello(context.Context, *String) (*String, error)
}

type HelloServiceClient interface {
    Hello(context.Context, *String, ...grpc.CallOption) (*String, error)
}

gRPC 通过 context.Context 参数，为每个方法调用提供了上下文支持。客户端在调用方法的时候，可以通过可选的 grpc.CallOption 类型的参数提供额外的上下文信息。

基于服务端的 HelloServiceServer 接口可以重新实现 HelloService 服务：

type HelloServiceImpl struct{}

func (p *HelloServiceImpl) Hello(
    ctx context.Context, args *String,
) (*String, error) {
    reply := &String{Value: "hello:" + args.GetValue()}
    return reply, nil
}

gRPC 服务的启动流程和标准库的 RPC 服务启动流程类似：

func main() {
    grpcServer := grpc.NewServer()
    RegisterHelloServiceServer(grpcServer, new(HelloServiceImpl))

    lis, err := net.Listen("tcp", ":1234")
    if err != nil {
        log.Fatal(err)
    }
    grpcServer.Serve(lis)
}

首先是通过 grpc.NewServer() 构造一个 gRPC 服务对象，然后通过 gRPC 插件生成的 RegisterHelloServiceServer 函数注册我们实现的 HelloServiceImpl 服务。然后通过 grpcServer.Serve(lis) 在一个监听端口上提供 gRPC 服务。

然后就可以通过客户端连接 gRPC 服务了：

func main() {
    conn, err := grpc.Dial("localhost:1234", grpc.WithInsecure())
    if err != nil {
        log.Fatal(err)
    }
    defer conn.Close()

    client := NewHelloServiceClient(conn)
    reply, err := client.Hello(context.Background(), &String{Value: "hello"})
    if err != nil {
        log.Fatal(err)
    }
    fmt.Println(reply.GetValue())
}

其中 grpc.Dial 负责和 gRPC 服务建立连接，然后 NewHelloServiceClient 函数基于已经建立的连接构造 HelloServiceClient 对象。返回的 client 其实是一个 HelloServiceClient 接口对象，通过接口定义的方法就可以调用服务端对应的 gRPC 服务提供的方法。

gRPC 和标准库的 RPC 框架有一个区别，gRPC 生成的接口并不支持异步调用。不过我们可以在多个 Goroutine 之间安全地共享 gRPC 底层的 HTTP/2 连接，因此可以通过在另一个 Goroutine 阻塞调用的方式模拟异步调用。

4.4.3 gRPC 流

RPC 是远程函数调用，因此每次调用的函数参数和返回值不能太大，否则将严重影响每次调用的响应时间。因此传统的 RPC 方法调用对于上传和下载较大数据量场景并不适合。同时传统 RPC 模式也不适用于对时间不确定的订阅和发布模式。为此，gRPC 框架针对服务器端和客户端分别提供了流特性。

服务端或客户端的单向流是双向流的特例，我们在 HelloService 增加一个支持双向流的 Channel 方法：

service HelloService {
    rpc Hello (String) returns (String);

    rpc Channel (stream String) returns (stream String);
}

关键字 stream 指定启用流特性，参数部分是接收客户端参数的流，返回值是返回给客户端的流。

重新生成代码可以看到接口中新增加的 Channel 方法的定义：

type HelloServiceServer interface {
    Hello(context.Context, *String) (*String, error)
    Channel(HelloService_ChannelServer) error
}
type HelloServiceClient interface {
    Hello(ctx context.Context, in *String, opts ...grpc.CallOption) (
        *String, error,
    )
    Channel(ctx context.Context, opts ...grpc.CallOption) (
        HelloService_ChannelClient, error,
    )
}

在服务端的 Channel 方法参数是一个新的 HelloService_ChannelServer 类型的参数，可以用于和客户端双向通信。客户端的 Channel 方法返回一个 HelloService_ChannelClient 类型的返回值，可以用于和服务端进行双向通信。

HelloService_ChannelServer 和 HelloService_ChannelClient 均为接口类型：

type HelloService_ChannelServer interface {
    Send(*String) error
    Recv() (*String, error)
    grpc.ServerStream
}

type HelloService_ChannelClient interface {
    Send(*String) error
    Recv() (*String, error)
    grpc.ClientStream
}

可以发现服务端和客户端的流辅助接口均定义了 Send 和 Recv 方法用于流数据的双向通信。

现在我们可以实现流服务：

func (p *HelloServiceImpl) Channel(stream HelloService_ChannelServer) error {
    for {
        args, err := stream.Recv()
        if err != nil {
            if err == io.EOF {
                return nil
            }
            return err
        }

        reply := &String{Value: "hello:" + args.GetValue()}

        err = stream.Send(reply)
        if err != nil {
            return err
        }
    }
}

服务端在循环中接收客户端发来的数据，如果遇到 io.EOF 表示客户端流被关闭，如果函数退出表示服务端流关闭。生成返回的数据通过流发送给客户端，双向流数据的发送和接收都是完全独立的行为。需要注意的是，发送和接收的操作并不需要一一对应，用户可以根据真实场景进行组织代码。

客户端需要先调用 Channel 方法获取返回的流对象：

stream, err := client.Channel(context.Background())
if err != nil {
    log.Fatal(err)
}

在客户端我们将发送和接收操作放到两个独立的 Goroutine。首先是向服务端发送数据：

go func() {
    for {
        if err := stream.Send(&String{Value: "hi"}); err != nil {
            log.Fatal(err)
        }
        time.Sleep(time.Second)
    }
}()

然后在循环中接收服务端返回的数据：

for {
    reply, err := stream.Recv()
    if err != nil {
        if err == io.EOF {
            break
        }
        log.Fatal(err)
    }
    fmt.Println(reply.GetValue())
}

这样就完成了完整的流接收和发送支持。

4.4.4 发布和订阅模式

在前一节中，我们基于 Go 内置的 RPC 库实现了一个简化版的 Watch 方法。基于 Watch 的思路虽然也可以构造发布和订阅系统，但是因为 RPC 缺乏流机制导致每次只能返回一个结果。在发布和订阅模式中，由调用者主动发起的发布行为类似一个普通函数调用，而被动的订阅者则类似 gRPC 客户端单向流中的接收者。现在我们可以尝试基于 gRPC 的流特性构造一个发布和订阅系统。

发布订阅是一个常见的设计模式，开源社区中已经存在很多该模式的实现。其中 docker 项目中提供了一个 pubsub 的极简实现，下面是基于 pubsub 包实现的本地发布订阅代码：

import (
    "github.com/moby/pubsub"
)

func main() {
    p := pubsub.NewPublisher(100*time.Millisecond, 10)

    golang := p.SubscribeTopic(func(v interface{}) bool {
        if key, ok := v.(string); ok {
            if strings.HasPrefix(key, "golang:") {
                return true
            }
        }
        return false
    })
    docker := p.SubscribeTopic(func(v interface{}) bool {
        if key, ok := v.(string); ok {
            if strings.HasPrefix(key, "docker:") {
                return true
            }
        }
        return false
    })

    go p.Publish("hi")
    go p.Publish("golang: https://golang.org")
    go p.Publish("docker: https://www.docker.com/")
    time.Sleep(1)

    go func() {
        fmt.Println("golang topic:", <-golang)
    }()
    go func() {
        fmt.Println("docker topic:", <-docker)
    }()

    <-make(chan bool)
}

其中 pubsub.NewPublisher 构造一个发布对象，p.SubscribeTopic() 可以通过函数筛选感兴趣的主题进行订阅。

现在尝试基于 gRPC 和 pubsub 包，提供一个跨网络的发布和订阅系统。首先通过 Protobuf 定义一个发布订阅服务接口：

service PubsubService {
    rpc Publish (String) returns (String);
    rpc Subscribe (String) returns (stream String);
}

其中 Publish 是普通的 RPC 方法，Subscribe 则是一个单向的流服务。然后 gRPC 插件会为服务端和客户端生成对应的接口：

type PubsubServiceServer interface {
    Publish(context.Context, *String) (*String, error)
    Subscribe(*String, PubsubService_SubscribeServer) error
}
type PubsubServiceClient interface {
    Publish(context.Context, *String, ...grpc.CallOption) (*String, error)
    Subscribe(context.Context, *String, ...grpc.CallOption) (
        PubsubService_SubscribeClient, error,
    )
}

type PubsubService_SubscribeServer interface {
    Send(*String) error
    grpc.ServerStream
}

因为 Subscribe 是服务端的单向流，因此生成的 PubsubService_SubscribeServer 接口中只有 Send 方法。

然后就可以实现发布和订阅服务了：

type PubsubService struct {
    pub *pubsub.Publisher
}

func NewPubsubService() *PubsubService {
    return &PubsubService{
        pub: pubsub.NewPublisher(100*time.Millisecond, 10),
    }
}

然后是实现发布方法和订阅方法：

func (p *PubsubService) Publish(
    ctx context.Context, arg *String,
) (*String, error) {
    p.pub.Publish(arg.GetValue())
    return &String{}, nil
}

func (p *PubsubService) Subscribe(
    arg *String, stream PubsubService_SubscribeServer,
) error {
    ch := p.pub.SubscribeTopic(func(v interface{}) bool {
        if key, ok := v.(string); ok {
            if strings.HasPrefix(key,arg.GetValue()) {
                return true
            }
        }
        return false
    })

    for v := range ch {
        if err := stream.Send(&String{Value: v.(string)}); err != nil {
            return err
        }
    }

    return nil
}

这样就可以从客户端向服务器发布信息了：

func main() {
    conn, err := grpc.Dial("localhost:1234", grpc.WithInsecure())
    if err != nil {
        log.Fatal(err)
    }
    defer conn.Close()

    client := NewPubsubServiceClient(conn)

    _, err = client.Publish(
        context.Background(), &String{Value: "golang: hello Go"},
    )
    if err != nil {
        log.Fatal(err)
    }
    _, err = client.Publish(
        context.Background(), &String{Value: "docker: hello Docker"},
    )
    if err != nil {
        log.Fatal(err)
    }
}

然后就可以在另一个客户端进行订阅信息了：

func main() {
    conn, err := grpc.Dial("localhost:1234", grpc.WithInsecure())
    if err != nil {
        log.Fatal(err)
    }
    defer conn.Close()

    client := NewPubsubServiceClient(conn)
    stream, err := client.Subscribe(
        context.Background(), &String{Value: "golang:"},
    )
    if err != nil {
        log.Fatal(err)
    }

    for {
        reply, err := stream.Recv()
        if err != nil {
            if err == io.EOF {
                break
            }
            log.Fatal(err)
        }

        fmt.Println(reply.GetValue())
    }
}

到此我们就基于 gRPC 简单实现了一个跨网络的发布和订阅服务。

gRPC进阶

作为一个基础的 RPC 框架，安全和扩展是经常遇到的问题。本节将简单介绍如何对 gRPC 进行安全认证。然后介绍通过 gRPC 的截取器特性，以及如何通过截取器优雅地实现 Token 认证、调用跟踪以及 Panic 捕获等特性。最后介绍了 gRPC 服务如何和其他 Web 服务共存。

4.5.1 证书认证

gRPC 建立在 HTTP/2 协议之上，对 TLS 提供了很好的支持。我们前面章节中 gRPC 的服务都没有提供证书支持，因此客户端在连接服务器中通过 grpc.WithInsecure() 选项跳过了对服务器证书的验证。没有启用证书的 gRPC 服务在和客户端进行的是明文通讯，信息面临被任何第三方监听的风险。为了保障 gRPC 通信不被第三方监听篡改或伪造，我们可以对服务器启动 TLS 加密特性。

可以用以下命令为服务器和客户端分别生成私钥和证书：

$ openssl genrsa -out server.key 2048
$ openssl req -new -x509 -days 3650 \
    -subj "/C=GB/L=China/O=grpc-server/CN=server.grpc.io" \
    -key server.key -out server.crt

$ openssl genrsa -out client.key 2048
$ openssl req -new -x509 -days 3650 \
    -subj "/C=GB/L=China/O=grpc-client/CN=client.grpc.io" \
    -key client.key -out client.crt

以上命令将生成 server.key、server.crt、client.key 和 client.crt 四个文件。其中以. key 为后缀名的是私钥文件，需要妥善保管。以. crt 为后缀名是证书文件，也可以简单理解为公钥文件，并不需要秘密保存。在 subj 参数中的 /CN=server.grpc.io 表示服务器的名字为 server.grpc.io，在验证服务器的证书时需要用到该信息。

有了证书之后，我们就可以在启动 gRPC 服务时传入证书选项参数：

func main() {
    creds, err := credentials.NewServerTLSFromFile("server.crt", "server.key")
    if err != nil {
        log.Fatal(err)
    }

    server := grpc.NewServer(grpc.Creds(creds))

    ...
}

其中 credentials.NewServerTLSFromFile 函数是从文件为服务器构造证书对象，然后通过 grpc.Creds(creds) 函数将证书包装为选项后作为参数传入 grpc.NewServer 函数。

在客户端基于服务器的证书和服务器名字就可以对服务器进行验证：

func main() {
    creds, err := credentials.NewClientTLSFromFile(
        "server.crt", "server.grpc.io",
    )
    if err != nil {
        log.Fatal(err)
    }

    conn, err := grpc.Dial("localhost:5000",
        grpc.WithTransportCredentials(creds),
    )
    if err != nil {
        log.Fatal(err)
    }
    defer conn.Close()

    ...
}

其中 credentials.NewClientTLSFromFile 是构造客户端用的证书对象，第一个参数是服务器的证书文件，第二个参数是签发证书的服务器的名字。然后通过 grpc.WithTransportCredentials(creds) 将证书对象转为参数选项传入 grpc.Dial 函数。

以上这种方式，需要提前将服务器的证书告知客户端，这样客户端在连接服务器时才能进行对服务器证书认证。在复杂的网络环境中，服务器证书的传输本身也是一个非常危险的问题。如果在中间某个环节，服务器证书被监听或替换那么对服务器的认证也将不再可靠。

为了避免证书的传递过程中被篡改，可以通过一个安全可靠的根证书分别对服务器和客户端的证书进行签名。这样客户端或服务器在收到对方的证书后可以通过根证书进行验证证书的有效性。

根证书的生成方式和自签名证书的生成方式类似：

$ openssl genrsa -out ca.key 2048
$ openssl req -new -x509 -days 3650 \
    -subj "/C=GB/L=China/O=gobook/CN=github.com" \
    -key ca.key -out ca.crt

然后是重新对服务器端证书进行签名：

$ openssl req -new \
    -subj "/C=GB/L=China/O=server/CN=server.io" \
    -key server.key \
    -out server.csr
$ openssl x509 -req -sha256 \
    -CA ca.crt -CAkey ca.key -CAcreateserial -days 3650 \
    -in server.csr \
    -out server.crt

签名的过程中引入了一个新的以. csr 为后缀名的文件，它表示证书签名请求文件。在证书签名完成之后可以删除. csr 文件。

然后在客户端就可以基于 CA 证书对服务器进行证书验证：

func main() {
    certificate, err := tls.LoadX509KeyPair("client.crt", "client.key")
    if err != nil {
        log.Fatal(err)
    }

    certPool := x509.NewCertPool()
    ca, err := ioutil.ReadFile("ca.crt")
    if err != nil {
        log.Fatal(err)
    }
    if ok := certPool.AppendCertsFromPEM(ca); !ok {
        log.Fatal("failed to append ca certs")
    }

    creds := credentials.NewTLS(&tls.Config{
        Certificates:       []tls.Certificate{certificate},
        ServerName:         tlsServerName, // NOTE: this is required!
        RootCAs:            certPool,
    })

    conn, err := grpc.Dial(
        "localhost:5000", grpc.WithTransportCredentials(creds),
    )
    if err != nil {
        log.Fatal(err)
    }
    defer conn.Close()

    ...
}

在新的客户端代码中，我们不再直接依赖服务器端证书文件。在 credentials.NewTLS 函数调用中，客户端通过引入一个 CA 根证书和服务器的名字来实现对服务器进行验证。客户端在连接服务器时会首先请求服务器的证书，然后使用 CA 根证书对收到的服务器端证书进行验证。

如果客户端的证书也采用 CA 根证书签名的话，服务器端也可以对客户端进行证书认证。我们用 CA 根证书对客户端证书签名：

$ openssl req -new \
    -subj "/C=GB/L=China/O=client/CN=client.io" \
    -key client.key \
    -out client.csr
$ openssl x509 -req -sha256 \
    -CA ca.crt -CAkey ca.key -CAcreateserial -days 3650 \
    -in client.csr \
    -out client.crt

因为引入了 CA 根证书签名，在启动服务器时同样要配置根证书：

func main() {
    certificate, err := tls.LoadX509KeyPair("server.crt", "server.key")
    if err != nil {
        log.Fatal(err)
    }

    certPool := x509.NewCertPool()
    ca, err := ioutil.ReadFile("ca.crt")
    if err != nil {
        log.Fatal(err)
    }
    if ok := certPool.AppendCertsFromPEM(ca); !ok {
        log.Fatal("failed to append certs")
    }

    creds := credentials.NewTLS(&tls.Config{
        Certificates: []tls.Certificate{certificate},
        ClientAuth:   tls.RequireAndVerifyClientCert, // NOTE: this is optional!
        ClientCAs:    certPool,
    })

    server := grpc.NewServer(grpc.Creds(creds))
    ...
}

服务器端同样改用 credentials.NewTLS 函数生成证书，通过 ClientCAs 选择 CA 根证书，并通过 ClientAuth 选项启用对客户端进行验证。

到此我们就实现了一个服务器和客户端进行双向证书验证的通信可靠的 gRPC 系统。

4.5.2 Token 认证

前面讲述的基于证书的认证是针对每个 gRPC 连接的认证。gRPC 还为每个 gRPC 方法调用提供了认证支持，这样就基于用户 Token 对不同的方法访问进行权限管理。

要实现对每个 gRPC 方法进行认证，需要实现 grpc.PerRPCCredentials 接口：

type PerRPCCredentials interface {
    // GetRequestMetadata gets the current request metadata, refreshing
    // tokens if required. This should be called by the transport layer on
    // each request, and the data should be populated in headers or other
    // context. If a status code is returned, it will be used as the status
    // for the RPC. uri is the URI of the entry point for the request.
    // When supported by the underlying implementation, ctx can be used for
    // timeout and cancellation.
    // TODO(zhaoq): Define the set of the qualified keys instead of leaving
    // it as an arbitrary string.
    GetRequestMetadata(ctx context.Context, uri ...string) (
        map[string]string,  error,
    )
    // RequireTransportSecurity indicates whether the credentials requires
    // transport security.
    RequireTransportSecurity() bool
}

在 GetRequestMetadata 方法中返回认证需要的必要信息。RequireTransportSecurity 方法表示是否要求底层使用安全连接。在真实的环境中建议必须要求底层启用安全的连接，否则认证信息有泄露和被篡改的风险。

我们可以创建一个 Authentication 类型，用于实现用户名和密码的认证：

type Authentication struct {
    User     string
    Password string
}

func (a *Authentication) GetRequestMetadata(context.Context, ...string) (
    map[string]string, error,
) {
    return map[string]string{"user":a.User, "password": a.Password}, nil
}
func (a *Authentication) RequireTransportSecurity() bool {
    return false
}

在 GetRequestMetadata 方法中，我们返回的认证信息包装 user 和 password 两个信息。为了演示代码简单，RequireTransportSecurity 方法表示不要求底层使用安全连接。

然后在每次请求 gRPC 服务时就可以将 Token 信息作为参数选项传入：

func main() {
    auth := Authentication{
        User:    "gopher",
        Password: "password",
    }

    conn, err := grpc.Dial("localhost"+port, grpc.WithInsecure(), grpc.WithPerRPCCredentials(&auth))
    if err != nil {
        log.Fatal(err)
    }
    defer conn.Close()

    ...
}

通过 grpc.WithPerRPCCredentials 函数将 Authentication 对象转为 grpc.Dial 参数。因为这里没有启用安全连接，需要传入 grpc.WithInsecure() 表示忽略证书认证。

然后在 gRPC 服务端的每个方法中通过 Authentication 类型的 Auth 方法进行身份认证：

type grpcServer struct {auth *Authentication}

func (p *grpcServer) SomeMethod(
    ctx context.Context, in *HelloRequest,
) (*HelloReply, error) {
    if err := p.auth.Auth(ctx); err != nil {
        return nil, err
    }

    return &HelloReply{Message: "Hello" + in.Name}, nil
}

func (a *Authentication) Auth(ctx context.Context) error {
    md, ok := metadata.FromIncomingContext(ctx)
    if !ok {
        return fmt.Errorf("missing credentials")
    }

    var appid string
    var appkey string

    if val, ok := md["user"]; ok { appid = val[0] }
    if val, ok := md["password"]; ok { appkey = val[0] }

    if appid != a.User || appkey != a.Password {
        return grpc.Errorf(codes.Unauthenticated, "invalid token")
    }

    return nil
}

详细的认证工作主要在 Authentication.Auth 方法中完成。首先通过 metadata.FromIncomingContext 从 ctx 上下文中获取元信息，然后取出相应的认证信息进行认证。如果认证失败，则返回一个 codes.Unauthenticated 类型的错误。

4.5.3 截取器

gRPC 中的 grpc.UnaryInterceptor 和 grpc.StreamInterceptor 分别对普通方法和流方法提供了截取器的支持。我们这里简单介绍普通方法的截取器用法。

要实现普通方法的截取器，需要为 grpc.UnaryInterceptor 的参数实现一个函数：

func filter(ctx context.Context,
    req interface{}, info *grpc.UnaryServerInfo,
    handler grpc.UnaryHandler,
) (resp interface{}, err error) {
    log.Println("filter:", info)
    return handler(ctx, req)
}

函数的 ctx 和 req 参数就是每个普通的 RPC 方法的前两个参数。第三个 info 参数表示当前是对应的那个 gRPC 方法，第四个 handler 参数对应当前的 gRPC 方法函数。上面的函数中首先是日志输出 info 参数，然后调用 handler 对应的 gRPC 方法函数。

要使用 filter 截取器函数，只需要在启动 gRPC 服务时作为参数输入即可：

server := grpc.NewServer(grpc.UnaryInterceptor(filter))

然后服务器在收到每个 gRPC 方法调用之前，会首先输出一行日志，然后再调用对方的方法。

如果截取器函数返回了错误，那么该次 gRPC 方法调用将被视作失败处理。因此，我们可以在截取器中对输入的参数做一些简单的验证工作。同样，也可以对 handler 返回的结果做一些验证工作。截取器也非常适合前面对 Token 认证工作。

下面是截取器增加了对 gRPC 方法异常的捕获：

func filter(
    ctx context.Context, req interface{},
    info *grpc.UnaryServerInfo,
    handler grpc.UnaryHandler,
) (resp interface{}, err error) {
    log.Println("filter:", info)

    defer func() {
        if r := recover(); r != nil {
            err = fmt.Errorf("panic: %v", r)
        }
    }()

    return handler(ctx, req)
}

不过 gRPC 框架中只能为每个服务设置一个截取器，因此所有的截取工作只能在一个函数中完成。开源的 grpc-ecosystem 项目中的 go-grpc-middleware 包已经基于 gRPC 对截取器实现了链式截取器的支持。

以下是 go-grpc-middleware 包中链式截取器的简单用法

import "github.com/grpc-ecosystem/go-grpc-middleware"

myServer := grpc.NewServer(
    grpc.UnaryInterceptor(grpc_middleware.ChainUnaryServer(
        filter1, filter2, ...
    )),
    grpc.StreamInterceptor(grpc_middleware.ChainStreamServer(
        filter1, filter2, ...
    )),
)

感兴趣的同学可以参考 go-grpc-middleware 包的代码。

4.5.4 和 Web 服务共存

gRPC 构建在 HTTP/2 协议之上，因此我们可以将 gRPC 服务和普通的 Web 服务架设在同一个端口之上。

对于没有启动 TLS 协议的服务则需要对 HTTP/2 特性做适当的调整：

func main() {
    mux := http.NewServeMux()

    h2Handler := h2c.NewHandler(mux, &http2.Server{})
    server = &http.Server{Addr: ":3999", Handler: h2Handler}
    server.ListenAndServe()
}

启用普通的 https 服务器则非常简单：

func main() {
    mux := http.NewServeMux()
    mux.HandleFunc("/", func(w http.ResponseWriter, req *http.Request) {
        fmt.Fprintln(w, "hello")
    })

    http.ListenAndServeTLS(port, "server.crt", "server.key",
        http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
            mux.ServeHTTP(w, r)
            return
        }),
    )
}

而单独启用带证书的 gRPC 服务也是同样的简单：

func main() {
    creds, err := credentials.NewServerTLSFromFile("server.crt", "server.key")
    if err != nil {
        log.Fatal(err)
    }

    grpcServer := grpc.NewServer(grpc.Creds(creds))

    ...
}

因为 gRPC 服务已经实现了 ServeHTTP 方法，可以直接作为 Web 路由处理对象。如果将 gRPC 和 Web 服务放在一起，会导致 gRPC 和 Web 路径的冲突，在处理时我们需要区分两类服务。

我们可以通过以下方式生成同时支持 Web 和 gRPC 协议的路由处理函数：

func main() {
    ...

    http.ListenAndServeTLS(port, "server.crt", "server.key",
        http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
            if r.ProtoMajor != 2 {
                mux.ServeHTTP(w, r)
                return
            }
            if strings.Contains(
                r.Header.Get("Content-Type"), "application/grpc",
            ) {
                grpcServer.ServeHTTP(w, r) // gRPC Server
                return
            }

            mux.ServeHTTP(w, r)
            return
        }),
    )
}

首先 gRPC 是建立在 HTTP/2 版本之上，如果 HTTP 不是 HTTP/2 协议则必然无法提供 gRPC 支持。同时，每个 gRPC 调用请求的 Content-Type 类型会被标注为 "application/grpc" 类型。

这样我们就可以在 gRPC 端口上同时提供 Web 服务了。

gRPC和Protobuf扩展

目前开源社区已经围绕 Protobuf 和 gRPC 开发出众多扩展，形成了庞大的生态。本节我们将简单介绍验证器和 REST 接口扩展。

4.6.1 验证器

到目前为止，我们接触的全部是第三版的 Protobuf 语法。第二版的 Protobuf 有个默认值特性，可以为字符串或数值类型的成员定义默认值。

我们采用第二版的 Protobuf 语法创建文件：

syntax = "proto2";

package main;

message Message {
    optional string name = 1 [default = "gopher"];
    optional int32 age = 2 [default = 10];
}

内置的默认值语法其实是通过 Protobuf 的扩展选项特性实现。在第三版的 Protobuf 中不再支持默认值特性，但是我们可以通过扩展选项自己模拟默认值特性。

下面是用 proto3 语法的扩展特性重新改写上述的 proto 文件：

syntax = "proto3";

package main;

import "google/protobuf/descriptor.proto";

extend google.protobuf.FieldOptions {
    string default_string = 50000;
    int32 default_int = 50001;
}

message Message {
    string name = 1 [(default_string) = "gopher"];
    int32 age = 2[(default_int) = 10];
}

其中成员后面的方括号内部的就是扩展语法。重新生成 Go 语言代码，里面会包含扩展选项相关的元信息：

var E_DefaultString = &proto.ExtensionDesc{
    ExtendedType:  (*descriptor.FieldOptions)(nil),
    ExtensionType: (*string)(nil),
    Field:         50000,
    Name:          "main.default_string",
    Tag:           "bytes,50000,opt,name=default_string,json=defaultString",
    Filename:      "helloworld.proto",
}

var E_DefaultInt = &proto.ExtensionDesc{
    ExtendedType:  (*descriptor.FieldOptions)(nil),
    ExtensionType: (*int32)(nil),
    Field:         50001,
    Name:          "main.default_int",
    Tag:           "varint,50001,opt,name=default_int,json=defaultInt",
    Filename:      "helloworld.proto",
}

我们可以在运行时通过类似反射的技术解析出 Message 每个成员定义的扩展选项，然后从每个扩展的相关联的信息中解析出我们定义的默认值。

在开源社区中，github.com/mwitkow/go-proto-validators 已经基于 Protobuf 的扩展特性实现了功能较为强大的验证器功能。要使用该验证器首先需要下载其提供的代码生成插件：

$ go get github.com/mwitkow/go-proto-validators/protoc-gen-govalidators

然后基于 go-proto-validators 验证器的规则为 Message 成员增加验证规则：

syntax = "proto3";

package main;

import "github.com/mwitkow/go-proto-validators/validator.proto";

message Message {
    string important_string = 1 [
        (validator.field) = {regex: "^[a-z]{2,5}$"}
    ];
    int32 age = 2 [
        (validator.field) = {int_gt: 0, int_lt: 100}
    ];
}

在方括弧表示的成员扩展中，validator.field 表示扩展是 validator 包中定义的名为 field 扩展选项。validator.field 的类型是 FieldValidator 结构体，在导入的 validator.proto 文件中定义。

所有的验证规则都由 validator.proto 文件中的 FieldValidator 定义：

syntax = "proto2";
package validator;

import "google/protobuf/descriptor.proto";

extend google.protobuf.FieldOptions {
    optional FieldValidator field = 65020;
}

message FieldValidator {
    // Uses a Golang RE2-syntax regex to match the field contents.
    optional string regex = 1;
    // Field value of integer strictly greater than this value.
    optional int64 int_gt = 2;
    // Field value of integer strictly smaller than this value.
    optional int64 int_lt = 3;

    // ... more ...
}

从 FieldValidator 定义的注释中我们可以看到验证器扩展的一些语法：其中 regex 表示用于字符串验证的正则表达式，int_gt 和 int_lt 表示数值的范围。

然后采用以下的命令生成验证函数代码：

protoc  \
    --proto_path=${GOPATH}/src \
    --proto_path=${GOPATH}/src/github.com/google/protobuf/src \
    --proto_path=. \
    --govalidators_out=. --go_out=plugins=grpc:.\
    hello.proto

windows: 替换 ${GOPATH} 为 %GOPATH% 即可.

以上的命令会调用 protoc-gen-govalidators 程序，生成一个独立的名为 hello.validator.pb.go 的文件：

var _regex_Message_ImportantString = regexp.MustCompile("^[a-z]{2,5}$")

func (this *Message) Validate() error {
    if !_regex_Message_ImportantString.MatchString(this.ImportantString) {
        return go_proto_validators.FieldError("ImportantString", fmt.Errorf(
            `value '%v' must be a string conforming to regex "^[a-z]{2,5}$"`,
            this.ImportantString,
        ))
    }
    if !(this.Age> 0) {
        return go_proto_validators.FieldError("Age", fmt.Errorf(
            `value '%v' must be greater than '0'`, this.Age,
        ))
    }
    if !(this.Age < 100) {
        return go_proto_validators.FieldError("Age", fmt.Errorf(
            `value '%v' must be less than '100'`, this.Age,
        ))
    }
    return nil
}

生成的代码为 Message 结构体增加了一个 Validate 方法，用于验证该成员是否满足 Protobuf 中定义的条件约束。无论采用何种类型，所有的 Validate 方法都用相同的签名，因此可以满足相同的验证接口。

通过生成的验证函数，并结合 gRPC 的截取器，我们可以很容易为每个方法的输入参数和返回值进行验证。

4.6.2 REST 接口

gRPC 服务一般用于集群内部通信，如果需要对外暴露服务一般会提供等价的 REST 接口。通过 REST 接口比较方便前端 JavaScript 和后端交互。开源社区中的 grpc-gateway 项目就实现了将 gRPC 服务转为 REST 服务的能力。

grpc-gateway 的工作原理如下图：

图 4-2 gRPC-Gateway 工作流程

通过在 Protobuf 文件中添加路由相关的元信息，通过自定义的代码插件生成路由相关的处理代码，最终将 REST 请求转给更后端的 gRPC 服务处理。

路由扩展元信息也是通过 Protobuf 的元数据扩展用法提供：

syntax = "proto3";

package main;

import "google/api/annotations.proto";

message StringMessage {
  string value = 1;
}

service RestService {
    rpc Get(StringMessage) returns (StringMessage) {
        option (google.api.http) = {
            get: "/get/{value}"
        };
    }
    rpc Post(StringMessage) returns (StringMessage) {
        option (google.api.http) = {
            post: "/post"
            body: "*"
        };
    }
}

我们首先为 gRPC 定义了 Get 和 Post 方法，然后通过元扩展语法在对应的方法后添加路由信息。其中 “/get/{value}” 路径对应的是 Get 方法，{value} 部分对应参数中的 value 成员，结果通过 json 格式返回。Post 方法对应 “/post” 路径，body 中包含 json 格式的请求信息。

然后通过以下命令安装 protoc-gen-grpc-gateway 插件：

go get -u github.com/grpc-ecosystem/grpc-gateway/protoc-gen-grpc-gateway

再通过插件生成 grpc-gateway 必须的路由处理代码：

$ protoc -I/usr/local/include -I. \
    -I$GOPATH/src \
    -I$GOPATH/src/github.com/grpc-ecosystem/grpc-gateway/third_party/googleapis \
    --grpc-gateway_out=. --go_out=plugins=grpc:.\
    hello.proto

windows: 替换 ${GOPATH} 为 %GOPATH% 即可.

插件会为 RestService 服务生成对应的 RegisterRestServiceHandlerFromEndpoint 函数：

func RegisterRestServiceHandlerFromEndpoint(
    ctx context.Context, mux *runtime.ServeMux, endpoint string,
    opts []grpc.DialOption,
) (err error) {
    ...
}

RegisterRestServiceHandlerFromEndpoint 函数用于将定义了 Rest 接口的请求转发到真正的 gRPC 服务。注册路由处理函数之后就可以启动 Web 服务了：

func main() {
    ctx := context.Background()
    ctx, cancel := context.WithCancel(ctx)
    defer cancel()

    mux := runtime.NewServeMux()

    err := RegisterRestServiceHandlerFromEndpoint(
        ctx, mux, "localhost:5000",
        []grpc.DialOption{grpc.WithInsecure()},
    )
    if err != nil {
        log.Fatal(err)
    }

    http.ListenAndServe(":8080", mux)
}

启动 grpc 服务 , 端口 5000

type RestServiceImpl struct{}

func (r *RestServiceImpl) Get(ctx context.Context, message *StringMessage) (*StringMessage, error) {
    return &StringMessage{Value: "Get hi:" + message.Value + "#"}, nil
}

func (r *RestServiceImpl) Post(ctx context.Context, message *StringMessage) (*StringMessage, error) {
    return &StringMessage{Value: "Post hi:" + message.Value + "@"}, nil
}
func main() {
    grpcServer := grpc.NewServer()
    RegisterRestServiceServer(grpcServer, new(RestServiceImpl))
    lis, _ := net.Listen("tcp", ":5000")
    grpcServer.Serve(lis)
}

首先通过 runtime.NewServeMux() 函数创建路由处理器，然后通过 RegisterRestServiceHandlerFromEndpoint 函数将 RestService 服务相关的 REST 接口中转到后面的 gRPC 服务。grpc-gateway 提供的 runtime.ServeMux 类也实现了 http.Handler 接口，因此可以和标准库中的相关函数配合使用。

当 gRPC 和 REST 服务全部启动之后，就可以用 curl 请求 REST 服务了：

$ curl localhost:8080/get/gopher
{"value":"Get: gopher"}

$ curl localhost:8080/post -X POST --data '{"value":"grpc"}'
{"value":"Post: grpc"}

在对外公布 REST 接口时，我们一般还会提供一个 Swagger 格式的文件用于描述这个接口规范。

$ go get -u github.com/grpc-ecosystem/grpc-gateway/protoc-gen-swagger

$ protoc -I. \
  -I$GOPATH/src/github.com/grpc-ecosystem/grpc-gateway/third_party/googleapis \
  --swagger_out=. \
  hello.proto

然后会生成一个 hello.swagger.json 文件。这样的话就可以通过 swagger-ui 这个项目，在网页中提供 REST 接口的文档和测试等功能。

4.6.3 Nginx

最新的 Nginx 对 gRPC 提供了深度支持。可以通过 Nginx 将后端多个 gRPC 服务聚合到一个 Nginx 服务。同时 Nginx 也提供了为同一种 gRPC 服务注册多个后端的功能，这样可以轻松实现 gRPC 负载均衡的支持。Nginx 的 gRPC 扩展是一个较大的主题，感兴趣的读者可以自行参考相关文档。

pbgo: 基于Protobuf的框架

pbgo 是我们专门针对本节内容设计的较为完整的迷你框架，它基于 Protobuf 的扩展语法，通过插件自动生成 rpc 和 rest 相关代码。在本章第二节我们已经展示过如何定制一个 Protobuf 代码生成插件，并生成了 rpc 部分的代码。在本节我们将重点讲述 pbgo 中和 Protobuf 扩展语法相关的 rest 部分的工作原理。

4.7.1 Protobuf 扩展语法

目前 Protobuf 相关的很多开源项目都使用到了 Protobuf 的扩展语法。在前一节中提到的验证器就是通过给结构体成员增加扩展元信息实现验证。在 grpc-gateway 项目中，则是通过为服务的每个方法增加 Http 相关的映射规则实现对 Rest 接口的支持。pbgo 也是通过 Protobuf 的扩展语法来为 rest 接口增加元信息。

pbgo 的扩展语法在 github.com/chai2010/pbgo/pbgo.proto 文件定义：

syntax = "proto3";
package pbgo;

option go_package = "github.com/chai2010/pbgo;pbgo";

import "google/protobuf/descriptor.proto";

extend google.protobuf.MethodOptions {
    HttpRule rest_api = 20180715;
}

message HttpRule {
    string get = 1;
    string put = 2;
    string post = 3;
    string delete = 4;
    string patch = 5;
}

pbgo.proto 文件是 pbgo 框架的一个部分，需要被其他的 proto 文件导入。Protobuf 本身自有一套完整的包体系，在这里包的路径就是 pbgo。Go 语言也有自己的一套包体系，我们需要通过 go_package 的扩展语法定义 Protobuf 和 Go 语言之间包的映射关系。定义 Protobuf 和 Go 语言之间包的映射关系之后，其他导入 pbgo.ptoto 包的 Protobuf 文件在生成 Go 语言时，会生成 pbgo.proto 映射的 Go 语言包路径。

Protobuf 扩展语法有五种类型，分别是针对文件的扩展信息、针对 message 的扩展信息、针对 message 成员的扩展信息、针对 service 的扩展信息和针对 service 方法的扩展信息。在使用扩展前首先需要通过 extend 关键字定义扩展的类型和可以用于扩展的成员。扩展成员可以是基础类型，也可以是一个结构体类型。pbgo 中只定义了 service 的方法的扩展，只定义了一个名为 rest_api 的扩展成员，类型是 HttpRule 结构体。

定义好扩展之后，我们就可以从其他的 Protobuf 文件中使用 pbgo 的扩展。创建一个 hello.proto 文件：

syntax = "proto3";
package hello_pb;

import "github.com/chai2010/pbgo/pbgo.proto";

message String {
    string value = 1;
}

service HelloService {
    rpc Hello (String) returns (String) {
        option (pbgo.rest_api) = {
            get: "/hello/:value"
        };
    }
}

首先我们通过导入 github.com/chai2010/pbgo/pbgo.proto 文件引入扩展定义，然后在 HelloService 的 Hello 方法中使用了 pbgo 定义的扩展。Hello 方法扩展的信息表示该方法对应一个 REST 接口，只有一个 GET 方法对应 "/hello/:value" 路径。在 REST 方法的路径中采用了 httprouter 路由包的语法规则，":value" 表示路径中的该字段对应的是参数中同名的成员。

4.7.2 插件中读取扩展信息

在本章的第二节我们已经简单讲述过 Protobuf 插件的工作原理，并且展示了如何生成 RPC 必要的代码。插件是一个 generator.Plugin 接口：

type Plugin interface {
    // Name identifies the plugin.
    Name() string
    // Init is called once after data structures are built but before
    // code generation begins.
    Init(g *Generator)
    // Generate produces the code generated by the plugin for this file,
    // except for the imports, by calling the generator's methods P, In,
    // and Out.
    Generate(file *FileDescriptor)
    // GenerateImports produces the import declarations for this file.
    // It is called after Generate.
    GenerateImports(file *FileDescriptor)
}

我们需要在 Generate 和 GenerateImports 函数中分别生成相关的代码。而 Protobuf 文件的全部信息都在 *generator.FileDescriptor 类型函数参数中描述，因此我们需要从函数参数中提前扩展定义的元数据。

pbgo 框架中的插件对象是 pbgoPlugin，在 Generate 方法中首先需要遍历 Protobuf 文件中定义的全部服务，然后再遍历每个服务的每个方法。在得到方法结构之后再通过自定义的 getServiceMethodOption 方法提取 rest 扩展信息：

func (p *pbgoPlugin) Generate(file *generator.FileDescriptor) {
    for _, svc := range file.Service {
        for _, m := range svc.Method {
            httpRule := p.getServiceMethodOption(m)
            ...
        }
    }
}

在讲述 getServiceMethodOption 方法之前我们先回顾下方法扩展的定义：

extend google.protobuf.MethodOptions {
    HttpRule rest_api = 20180715;
}

pbgo 为服务的方法定义了一个 rest_api 名字的扩展，在最终生成的 Go 语言代码中会包含一个 pbgo.E_RestApi 全局变量，通过该全局变量可以获取用户定义的扩展信息。

下面是 getServiceMethodOption 方法的实现：

func (p *pbgoPlugin) getServiceMethodOption(
    m *descriptor.MethodDescriptorProto,
) *pbgo.HttpRule {
    if m.Options != nil && proto.HasExtension(m.Options, pbgo.E_RestApi) {
        ext, _ := proto.GetExtension(m.Options, pbgo.E_RestApi)
        if ext != nil {
            if x, _ := ext.(*pbgo.HttpRule); x != nil {
                return x
            }
        }
    }
    return nil
}

首先通过 proto.HasExtension 函数判断每个方法是否定义了扩展，然后通过 proto.GetExtension 函数获取用户定义的扩展信息。在获取到扩展信息之后，我们再将扩展转型为 pbgo.HttpRule 类型。

有了扩展信息之后，我们就可以参考第二节中生成 RPC 代码的方式生成 REST 相关的代码。

4.7.3 生成 REST 代码

pbgo 框架同时也提供了一个插件用于生成 REST 代码。不过我们的目的是学习 pbgo 框架的设计过程，因此我们先尝试手写 Hello 方法对应的 REST 代码，然后插件再根据手写的代码构造模板自动生成代码。

HelloService 只有一个 Hello 方法，Hello 方法只定义了一个 GET 方式的 REST 接口：

message String {
    string value = 1;
}

service HelloService {
    rpc Hello (String) returns (String) {
        option (pbgo.rest_api) = {
            get: "/hello/:value"
        };
    }
}

为了方便最终的用户，我们需要为 HelloService 构造一个路由。因此我们希望有个一个类似 HelloServiceHandler 的函数，可以基于 HelloServiceInterface 服务的接口生成一个路由处理器：

type HelloServiceInterface interface {
    Hello(in *String, out *String) error
}

func HelloServiceHandler(svc HelloServiceInterface) http.Handler {
    var router = httprouter.New()
    _handle_HelloService_Hello_get(router, svc)
    return router
}

代码中选择的是开源中比较流行的 httprouter 路由引擎。其中_handle_HelloService_Hello_get 函数用于将 Hello 方法注册到路由处理器：

func _handle_HelloService_Hello_get(
    router *httprouter.Router, svc HelloServiceInterface,
) {
    router.Handle("GET", "/hello/:value",
        func(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
            var protoReq, protoReply String

            err := pbgo.PopulateFieldFromPath(&protoReq, fieldPath, ps.ByName("value"))
            if err != nil {
                http.Error(w, err.Error(), http.StatusBadRequest)
                return
            }

            if err := svc.Hello(&protoReq, &protoReply); err != nil {
                http.Error(w, err.Error(), http.StatusInternalServerError)
                return
            }

            if err := json.NewEncoder(w).Encode(&protoReply); err != nil {
                http.Error(w, err.Error(), http.StatusInternalServerError)
                return
            }
        },
    )
}

首先通过 router.Handle 方法注册路由函数。在路由函数内部首先通过 ps.ByName("value") 从 URL 中加载 value 参数，然后通过 pbgo.PopulateFieldFromPath 辅助函数设置 value 参数对应的成员。当输入参数准备就绪之后就可以调用 HelloService 服务的 Hello 方法，最终将 Hello 方法返回的结果用 json 编码返回。

在手工构造完成最终代码的结构之后，就可以在此基础上构造插件生成代码的模板。完整的插件代码和模板在 protoc-gen-pbgo/pbgo.go 文件，读者可以自行参考。

4.7.4 启动 REST 服务

虽然从头构造 pbgo 框架的过程比较繁琐，但是使用 pbgo 构造 REST 服务却是异常简单。首先要构造一个满足 HelloServiceInterface 接口的服务对象：

import (
    "github.com/chai2010/pbgo/examples/hello.pb"
)

type HelloService struct{}

func (p *HelloService) Hello(request *hello_pb.String, reply *hello_pb.String) error {
    reply.Value = "hello:" + request.GetValue()
    return nil
}

和 RPC 代码一样，在 Hello 方法中简单返回结果。然后调用该服务对应的 HelloServiceHandler 函数生成路由处理器，并启动服务：

func main() {
    router := hello_pb.HelloServiceHandler(new(HelloService))
    log.Fatal(http.ListenAndServe(":8080", router))
}

然后在命令行测试 REST 服务：

$ curl localhost:8080/hello/vgo

这样一个超级简单的 pbgo 框架就完成了！

grpcurl工具

Protobuf 本身具有反射功能，可以在运行时获取对象的 Proto 文件。gRPC 同样也提供了一个名为 reflection 的反射包，用于为 gRPC 服务提供查询。gRPC 官方提供了一个 C++ 实现的 grpc_cli 工具，可以用于查询 gRPC 列表或调用 gRPC 方法。但是 C++ 版本的 grpc_cli 安装比较复杂，我们推荐用纯 Go 语言实现的 grpcurl 工具。本节将简要介绍 grpcurl 工具的用法。

4.8.1 启动反射服务

reflection 包中只有一个 Register 函数，用于将 grpc.Server 注册到反射服务中。reflection 包文档给出了简单的使用方法：

import (
    "google.golang.org/grpc/reflection"
)

func main() {
    s := grpc.NewServer()
    pb.RegisterYourOwnServer(s, &server{})

    // Register reflection service on gRPC server.
    reflection.Register(s)

    s.Serve(lis)
}

如果启动了 gprc 反射服务，那么就可以通过 reflection 包提供的反射服务查询 gRPC 服务或调用 gRPC 方法。

4.8.2 查看服务列表

grpcurl 是 Go 语言开源社区开发的工具，需要手工安装：

$ go get github.com/fullstorydev/grpcurl
$ go install github.com/fullstorydev/grpcurl/cmd/grpcurl

grpcurl 中最常使用的是 list 命令，用于获取服务或服务方法的列表。比如 grpcurl localhost:1234 list 命令将获取本地 1234 端口上的 grpc 服务的列表。在使用 grpcurl 时，需要通过 -cert 和 -key 参数设置公钥和私钥文件，连接启用了 tls 协议的服务。对于没有没用 tls 协议的 grpc 服务，通过 -plaintext 参数忽略 tls 证书的验证过程。如果是 Unix Socket 协议，则需要指定 -unix 参数。

如果没有配置好公钥和私钥文件，也没有忽略证书的验证过程，那么将会遇到类似以下的错误：

$ grpcurl localhost:1234 list
Failed to dial target host "localhost:1234": tls: first record does not \
look like a TLS handshake

如果 grpc 服务正常，但是服务没有启动 reflection 反射服务，将会遇到以下错误：

$ grpcurl -plaintext localhost:1234 list
Failed to list services: server does not support the reflection API

假设 grpc 服务已经启动了 reflection 反射服务，服务的 Protobuf 文件如下：

syntax = "proto3";

package HelloService;

message String {
    string value = 1;
}

service HelloService {
    rpc Hello (String) returns (String);
    rpc Channel (stream String) returns (stream String);
}

grpcurl 用 list 命令查看服务列表时将看到以下输出：

$ grpcurl -plaintext localhost:1234 list
HelloService.HelloService
grpc.reflection.v1alpha.ServerReflection

其中 HelloService.HelloService 是在 protobuf 文件定义的服务。而 ServerReflection 服务则是 reflection 包注册的反射服务。通过 ServerReflection 服务可以查询包括本身在内的全部 gRPC 服务信息。

4.8.3 服务的方法列表

继续使用 list 子命令还可以查看 HelloService 服务的方法列表：

$ grpcurl -plaintext localhost:1234 list HelloService.HelloService
Channel
Hello

从输出可以看到 HelloService 服务提供了 Channel 和 Hello 两个方法，和 Protobuf 文件的定义是一致的。

如果还想了解方法的细节，可以使用 grpcurl 提供的 describe 子命令查看更详细的描述信息：

$ grpcurl -plaintext localhost:1234 describe HelloService.HelloService
HelloService.HelloService is a service:
{
  "name": "HelloService",
  "method": [
    {
      "name": "Hello",
      "inputType": ".HelloService.String",
      "outputType": ".HelloService.String",
      "options": {

      }
    },
    {
      "name": "Channel",
      "inputType": ".HelloService.String",
      "outputType": ".HelloService.String",
      "options": {

      },
      "clientStreaming": true,
      "serverStreaming": true
    }
  ],
  "options": {

  }
}

输出列出了服务的每个方法，每个方法输入参数和返回值对应的类型。

4.8.4 获取类型信息

在获取到方法的参数和返回值类型之后，还可以继续查看类型的信息。下面是用 describe 命令查看参数 HelloService.String 类型的信息：

$ grpcurl -plaintext localhost:1234 describe HelloService.String
HelloService.String is a message:
{
  "name": "String",
  "field": [
    {
      "name": "value",
      "number": 1,
      "label": "LABEL_OPTIONAL",
      "type": "TYPE_STRING",
      "options": {

      },
      "jsonName": "value"
    }
  ],
  "options": {

  }
}

json 信息对应 HelloService.String 类型在 Protobuf 中的定义如下：

message String {
    string value = 1;
}

输出的 json 数据只不过是 Protobuf 文件的另一种表示形式。

4.8.5 调用方法

在获取 gRPC 服务的详细信息之后就可以 json 调用 gRPC 方法了。

下面命令通过 -d 参数传入一个 json 字符串作为输入参数，调用的是 HelloService 服务的 Hello 方法：

$ grpcurl -plaintext -d '{"value":"gopher"}' \
    localhost:1234 HelloService.HelloService/Hello
{
  "value": "hello:gopher"
}

如果 -d 参数是 @ 则表示从标准输入读取 json 输入参数，这一般用于比较输入复杂的 json 数据，也可以用于测试流方法。

下面命令是连接 Channel 流方法，通过从标准输入读取输入流参数：

$ grpcurl -plaintext -d @ localhost:1234 HelloService.HelloService/Channel
{"value": "gopher"}
{
  "value": "hello:gopher"
}

{"value": "wasm"}
{
  "value": "hello:wasm"
}

通过 grpcurl 工具，我们可以在没有客户端代码的环境下测试 gRPC 服务。

补充说明

4.9 补充说明

目前专门讲述 RPC 的图书比较少。目前 Protobuf 和 gRPC 的官网都提供了详细的参考资料和例子。本章重点讲述了 Go 标准库的 RPC 和基于 Protobuf 衍生的 gRPC 框架，同时也简单展示了如何自己定制一个 RPC 框架。之所以聚焦在这几个有限的主题，是因为这几个技术都是 Go 语言团队官方在进行维护，和 Go 语言契合也最为默契。不过 RPC 依然是一个庞大的主题，足以单独成书。目前开源世界也有很多富有特色的 RPC 框架，还有针对分布式系统进行深度定制的 RPC 系统，用户可以根据自己实际需求选择合适的工具。

本文转载自开源项目 Go 语言高级编程（作者 柴树杉、曹春晖），按 BSD-2-Clause 开源 许可刊载，正文未改动。

医疗、法律、政策类内容仅供参考，具体以当地部门、医生、律师的说法为准。