grpc和protobuf

什么是rpc

RPC（Remote Procedure Call），翻译成中文叫远程过程调用。其设计思路是程序A可以像支持本地函数一样调用远程程序B的一个函数。程序A和程序B很可能不在一台机器上，这中间就需要网络通讯，一般都是基于tcp或者http协议。函数有入参和返回值这里就需要约定一致的序列化方式，比较常见的有binary，json，xml。函数的调用过程可以同步和异步。随着分布式程序近几年的大行其道，rpc作为其主要的通讯方式也越来越进入我们的视野，比如grpc，thrift。

go标准库的rpc

golang的标准库就支持rpc。

server端

package main

import (
	"errors"
	"log"
	"net"
	"net/rpc"
)
type Param struct {
	A,B int
}
type Server struct {}

func (t *Server) Multiply(args *Param, reply *int) error {
	*reply = args.A * args.B
	return nil
}
func (t *Server) Divide(args *Param, reply *int) error {
	if args.B == 0 {
		return errors.New("divide by zero")
	}
	*reply = args.A / args.B
	return nil
}

func main() {
	rpc.RegisterName("test", new(Server))
	listener, err := net.Listen("tcp", ":9700")
	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)
	}
}

client端

package main
import (
	"fmt"
	"log"
	"net/rwida/micro-server
	"time"wida/micro-server
)
type Param struct {
	A,B int
}
func main() {
	client, err := rpc.Dial("tcp", "localhost:9700")
	if err != nil {
		log.Fatal("dialing:", err)
	}
	//同步调用
	var reply int
	err = client.Call("test.Multiply", Param{34,35}, &reply)
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println(reply)
	//异步调用
	done := make(chan *rpc.Call, 1)
	client.Go("test.Divide", Param{34,17}, &reply,done)
	select {
		case d := <-done:
			fmt.Println(* d.Reply.(*int))

		case <-time.After(3e9):
			fmt.Println("time out")
	}
}

$ go run server/main.go &
$ go run client/main.go
1190
2

程序中我们没有看到参数的序列化和反序列化过程，实际上go标准库rpc使用encoding/gob做了序列化和反序列化操作，但是encoding/gob只支持go语言内部做交互，如果需要夸语言的话就不能用encoding/gob了。我们还可以使用标准库中的jsonrpc.

server

package main

import (
	"errors"
	"log"
	"net"
	"net/rpc"
	"net/rpc/jsonrpc"
)

type Param struct {
	A,B int
}
type Server struct {}

func (t *Server) Multiply(args *Param, reply *int) error {
	*reply = args.A * args.B
	return nil
}
func (t *Server) Divide(args *Param, reply *int) error {
	if args.B == 0 {
		return errors.New("divide by zero")
	}
	*reply = args.A / args.B
	return nil
}

func main() {
	rpc.RegisterName("test", new(Server))
	listener, err := net.Listen("tcp", ":9700")
	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))
	}
}

$ go run server/main.go &
$ echo -e '{"method":"test.Multiply","params":[{"A":34,"B":35}],"id":0}' | nc localhost 9700
{"id":0,"result":1190,"error":null}
$ echo -e '{"method":"test.Divide","params":[{"A":34,"B":17}],"id":1}' | nc localhost 9700
{"id":0,"result":1190,"error":null}

这个例子中我们用go写了服务端，客户端我们使用linux 命令行工具nc（natcat可能需要单独安装）直接和服务端交互。可以看到交互的序列化方式是json，因此其它语言也很容易实现和该服务端的交互。

当然我们上面演示的是基于tcp的rpc方式，标准库同时也支持http协议的rpc，感兴趣的同学可以去了解下，这边就不做展开介绍。

GRPC

上面我们发了一些篇幅介绍了标准库的rpc，主要的目的是了解rpc是什么。实际的开发中我们反而比较少用标准库的rpc，我们通常会选择grpc，grpc不仅仅在go生态里头非常流行，在其他语言生态里头同样也非常流行。

grpc是google公司开发的基于http2协议设计和protobuf开发的，高性能，跨语言的rpc框架。这边着重介绍下http2的一个重要特性当tcp多路复用功能，说得直白点就是一个在tcp链接执行多个请求，所以Service A提供N多个服务，Client B和A的所有交互都只用一个链接，这样子可以省很多的链接资源。对tcp连接复用（ connection multiplexing）感兴趣的同学可以阅读下yamux的源码。

protobuf

protobuf是google开发的一种平台中立，语言中立，可拓展的数据描述语言。类似json，xml等数据描述语言类似，proto也实现了自己的序列化和反序列化方式，对相对于json来说，protobuf序列化效率更高，体积更小。官方地址protobuf有兴趣的同学可以看看。

protobuf安装

我们可以从protobuf/releases里头下载到相应的二进制版本安装protobuf。比如

protoc-3.10.0-win32.zip
protoc-3.10.0-linux-x86_64.zip

解压文件到文件夹，然后将该文件目录添加到环境变量PATH中。

$ protoc --version
libprotoc 3.10.0

ok protobuf就安装成功了。

安装protobuf go插件

go get -u -v github.com/golang/protobuf/proto
go get -u -v github.com/golang/protobuf/protoc-gen-go

hello world

新建一个go项目

├── client
│   └── main.go
├── pb
│   ├── gen.sh
│   └── search.proto
└── server
    └── main.go

protobuf 消息定义文件为 pb/search.proto

syntax = "proto3";

package pb;

service Searcher {
    rpc Search (SearchRequest) returns (SearchReply) {}
}

message SearchRequest {
    bool name = 1;
}

message SearchReply {
    string name = 1;
}

这个文件我定义了rpc的服务Searcher，里头定了一个接口Search，同时定义了入参和返回值类型。

我们写一个shell脚本gen.sh来生成go程序文件

#!/bin/bash
PROTOC=`which protoc`
$PROTOC  --go_out=plugins=grpc:. search.proto

cd 到pb文件夹执行 gen.sh 脚本

$ ./gen.sh
$ cd ../ && tree
├── client
│   └── main.go
├── pb
│   ├── gen.sh
│   ├── search.pb.go
│   └── search.proto
└── server
    └── main.go

可以看到我们多了一个 search.pb.go的文件。我们在server文件下写我们服务main.go

package main
import (
	"log"
	"net"
	"context"
	"github.com/widaT/gorpc_demo/grpc/pb"
	"google.golang.org/grpc"
)

const (
	port = ":50051"
)

type server struct{}

func (s *server) Search(ctx context.Context, in *pb.SearchRequest) (*pb.SearchReply, error) {
	return &pb.SearchReply{Name:"hello " + in.GetName()}, nil
}
func main() {
	lis, err := net.Listen("tcp", port)
	if err != nil {
		log.Fatalf("failed to listen: %v", err)
	}
	s := grpc.NewServer()
	pb.RegisterSearcherServer(s, &server{})
	s.Serve(lis)
}

我们在client文件下写我们的客户端main.go

package main

import (
	"context"
	"fmt"
	"github.com/widaT/gorpc_demo/grpc/pb"
	"google.golang.org/grpc"
	"log"
	"time"
)

const (
	address     = "localhost:50051"
)

func main() {
	// Set up a connection to the server.
	conn, err := grpc.Dial(address, grpc.WithInsecure())
	if err != nil {
		log.Fatalf("did not connect: %v", err)
	}
	defer conn.Close()
	c := pb.NewSearcherClient(conn)
	s := time.Now()
	r, err := c.Search(context.Background(), &pb.SearchRequest{Name: "world"})
	if err != nil {
		log.Fatalf("could not greet: %v", err)
	}
	fmt.Println(r , time.Now().Sub(s))
}

$ go run server/main.go &
$ go run client/main.go &
name:"hello world"  14.767013ms

grpc stream

上面的hello world程序演示了grpc的一般用法，这种方式能满足大部分的场景。grpc还提供了双向或者单向流的方式我们成为grpc stream，stream的方式一般用在有大量的数据交互或者长时间交互。

我们修改的grpc服务定义文件pb/search.proto,在service增加 rpc Search2 (stream SearchRequest) returns (stream SearchReply) {}

syntax = "proto3";
package pb;
service Searcher {
    rpc Search (SearchRequest) returns (SearchReply) {}
    rpc Search2 (stream SearchRequest) returns (stream SearchReply) {}
}
message SearchRequest {
    string name = 1;
}
message SearchReply {
    string name = 1;
}

服务端我们修改代码，添加一个func (s *server) Search2(pb.Searcher_Search2Server) error的一个实现方法。

package main

import (
	"io"
	"log"
	"net"

	"context"
	"github.com/widaT/gorpc_demo/grpc/pb"
	"google.golang.org/grpc"
)

const (
	port = ":50051"
)

type server struct{}

func (s *server) Search(ctx context.Context, in *pb.SearchRequest) (*pb.SearchReply, error) {
	return &pb.SearchReply{Name:"hello " + in.GetName()}, nil
}
func (s *server) Search2(stream pb.Searcher_Search2Server) error {
	for {
		args, err := stream.Recv()
		if err != nil {
			if err == io.EOF {
				return nil
			}
			return err
		}
		reply := &pb.SearchReply{ Name:"hello:" + args.GetName()}
		err = stream.Send(reply)
		if err != nil {
			return err
		}
	}
	return nil
}

func main() {
	lis, err := net.Listen("tcp", port)
	if err != nil {
		log.Fatalf("failed to listen: %v", err)
	}
	s := grpc.NewServer()
	pb.RegisterSearcherServer(s, &server{})
	s.Serve(lis)
}

为了方便阅读代码，client代码我们重写

package main

import (
	"context"
	"fmt"
	"github.com/widaT/gorpc_demo/grpc/pb"
	"google.golang.org/grpc"
	"io"
	"log"
	"time"
)

const (
	address = "localhost:50051"
)

func main() {
	conn, err := grpc.Dial(address, grpc.WithInsecure())
	if err != nil {
		log.Fatalf("did not connect: %v", err)
	}
	defer conn.Close()
	c := pb.NewSearcherClient(conn)

	stream, err := c.Search2(context.Background())

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

	go func() { //这边启动一个goroutine 发送请求
		for {
			if err := stream.Send(&pb.SearchRequest{Name: "world"}); err != nil {
				log.Fatal(err)
			}
			time.Sleep(time.Second)
		}
	}()

	for { //主goroutine 一直接收结果
		rep, err := stream.Recv()
		if err != nil {
			if err == io.EOF {
				break
			}
			log.Fatal(err)
		}
		fmt.Println(rep.GetName())
	}
}

运行代码

$ go run server/main.go &
$ go run client/main.go 
hello:world
hello:world
hello:world
...

总结

本文只是对golang使用grpc做了简要的介绍，grpc的使用范围很广，需要我们持续了解和学习。下面推荐个资料

参考文档

grpc go

Go一站式编程