RPC（Remote Procedure Call）远程过程调用，它可以使一台主机上的进程调用另一台主机的进程，由以访为其他若干个主机提供服务，也就是我们常说的 C/S 服务，Server 与 Client 之间通过 rpc 方式进行通信。

Go 标准包中已经提供了对 RPC 的支持，支持三个级别的 RPC：TCP、HTTP、JSONRPC。Go 的 RPC 包与传统的 RPC 系统不同，他只支持 Go 开发的服务器与客户端之间的交互，因为在内部，它们采用了 Gob 来编码。

Go RPC 的函数要满足下面的条件才能够被远程调用，不然会被忽略：

  • 函数必须是导出的，即首字母为大写。
  • 必须有两个导出类型的参数。
  • 第一个参数是接收的参数，第二个参数是返回给客户端的参数，第二个参数必须是指针类型的。
  • 函数还要有一个 error 类型返回值。方法的返回值，如果非 nil，将被作为字符串回传，在客户端看来就和 errors.New 创建的一样。如果返回了错误，回复的参数将不会被发送给客户端。

举个例子，正确的 RPC 函数格式为：

1
func (t *T) MethidName(argType T1, replyType *T2) error

T、T1 和 T2 类型都必须能被 encoding/gob 包编解码

任何 RPC 都需要通过网络来传递数据，Go RPC 可以利用 HTTP 和 TCP 来传递数据

Server 和 Client

server

Server 对象

在 Server 对象中定义了互斥锁用来保护请求数据，另外还包含请求信息和返回的信息以及注册的服务。

1
2
3
4
5
6
7
8
// Server represents an RPC Server.
type Server struct {
	serviceMap sync.Map   // map[string]*service
	reqLock    sync.Mutex // protects freeReq
	freeReq    *Request
	respLock   sync.Mutex // protects freeResp
	freeResp   *Response
}

我们可以通过 NewServer 初始化一个 Server 对象：

1
2
3
4
// NewServer returns a new Server.
func NewServer() *Server {
	return &Server{}
}

Server 对象有 8 个方法，下面进行介绍。

func (server *Server) Register(rcvr interface{}) error

Register 用来向 Server 注册 rpc 服务，rpc 服务必须满足下面五种要求：

  • 函数必须是导出的
  • 必须有两个导出类型参数
  • 第一个参数是接收参数
  • 第二个参数是返回给客户端参数，必须是指针类型
  • 函数还要有一个返回值 error

注册之后的服务，Client 可以进行远程调用。

还有一个和 Registry 类似的方法：

1
2
3
4
// RegisterName类似Register，但使用提供的name代替rcvr的具体类型名作为服务名。
func (server *Server) RegisterName(name string, rcvr interface{}) error {
	return server.register(rcvr, name, true)
}

监听器

1
2
// Accept接收监听器l获取的连接，然后服务每一个连接。Accept会阻塞，调用者应另开线程："go server.Accept(l)"
func (server *Server) Accept(lis net.Listener)

服务端处理请求的相关方法

  • ServeConn 方法

    1
    2
    3
    4
    // ServeConn在单个连接上执行server。ServeConn会阻塞，服务该连接直到客户端挂起。
    // 调用者一般应另开线程调用本函数："go server.ServeConn(conn)"。ServeConn在该连接使用gob（参见encoding/gob包）有线格式。
    // 要使用其他的编解码器，可调用ServeCodec方法。
    func (server *Server) ServeConn(conn io.ReadWriteCloser)

  • ServeCodec 方法

    1
    2
    // ServeCodec类似ServeConn，但使用指定的编解码器，以编码请求主体和解码回复主体。
    func (server *Server) ServeCodec(codec ServerCodec)

  • ServeRequest 方法

    1
    2
    // ServeRequest类似ServeCodec，但异步的服务单个请求。它不会在调用结束后关闭codec。
    func (server *Server) ServeRequest(codec ServerCodec) error

  • ServeHTTP 方法

    1
    2
    // ServeHTTP实现了回应RPC请求的http.Handler接口。
    func (server *Server) ServeHTTP(w http.ResponseWriter, req *http.Request)

  • HandleHTTP 方法

    1
    2
    3
    // HandleHTTP注册server的RPC信息HTTP处理器对应到rpcPath，注册server的debug信息HTTP处理器对应到debugPath。
    // HandleHTTP会注册到http.DefaultServeMux。之后，仍需要调用http.Serve()，一般会另开线程："go http.Serve(l, nil)"
    func (server *Server) HandleHTTP(rpcPath, debugPath string)

client

Client 对象

1
2
3
4
5
6
7
8
9
10
11
12
13
// Client类型代表RPC客户端。同一个客户端可能有多个未返回的调用，也可能被多个go程同时使用。
type Client struct {
	codec ClientCodec

	reqMutex sync.Mutex // protects following
	request  Request

	mutex    sync.Mutex // protects following
	seq      uint64
	pending  map[uint64]*Call
	closing  bool // user has called Close
	shutdown bool // server has told us to stop
}

新建一个 Client

  • 初始化一个 Client

    1
    2
    // NewClient返回一个新的Client，以管理对连接另一端的服务的请求。它添加缓冲到连接的写入侧，以便将回复的头域和有效负载作为一个单元发送。
    func NewClient(conn io.ReadWriteCloser) *Client

  • 初始化一个 Client 并指定编码器

    1
    2
    // 另外还有一个NewClientWithCodec方法，NewClientWithCodec类似NewClient，但使用指定的编解码器，以编码请求主体和解码回复主体。
    func NewClientWithCodec(codec ClientCodec) *Client

连接服务端

  • 通过指定的网络和地址与 RPC 服务端连接。

    1
    func Dial(network, address string) (*Client, error)

  • 通过指定的网络和地址与在默认 HTTP RPC 路径监听的 HTTP RPC 服务端连接。

    1
    func DialHTTP(network, address string) (*Client, error)

  • 通过在指定的网络、地址和路径与 HTTP RPC 服务端连接。

    1
    func DialHTTPPath(network, address, path string) (*Client, error)

调用服务端的服务

  • Call 方法

    1
    2
    // Call调用指定的方法，等待调用返回，将结果写入reply，然后返回执行的错误状态。
    func (client *Client) Call(serviceMethod string, args interface{}, reply interface{}) error

  • 异步调用（Go 方法）

    1
    2
    3
    4
    // Go异步的调用函数。
    // 本方法Call结构体类型指针的返回值代表该次远程调用。通道类型的参数done会在本次调用完成时发出信号（通过返回本次Go方法的返回值）。
    // 如果done为nil，Go会申请一个新的通道（写入返回值的Done字段）；如果done非nil，done必须有缓冲，否则Go方法会故意崩溃。
    func (client *Client) Go(serviceMethod string, args interface{}, reply interface{}, done chan *Call) *Call

关闭 Client

1
func (client *Client) Close() error

go 语言标准库中的 JSON-RPC

JSON-RPC，是一个无状态且轻量级的远程过程调用（RPC）传送协议，其传递内容通过 JSON 为主。相较于一般的 REST 通过网址（如 GET /user）调用远程服务器，JSON-RPC 直接在内容中定义了欲调用的函数名称（如 {“method”: “getUser”}），这也令开发者不会陷于该使用 PUT 或者 PATCH 的问题之中。

更多 JSON-RPC 约定参见：https://zh.wikipedia.org/wiki/JSON-RPC

连接 Server 端

1
2
// Dial在指定的网络和地址连接一个JSON-RPC服务端。
func Dial(network, address string) (*rpc.Client, error)

创建 CLient

  • NewClient

    1
    2
    // NewClient返回一个新的rpc.Client，以管理对连接另一端的服务的请求。
    func NewClient(conn io.ReadWriteCloser) *rpc.Client

  • NewClientCodec

    1
    2
    // NewClientCodec返回一个在连接上使用JSON-RPC的rpc.ClientCodec。
    func NewClientCodec(conn io.ReadWriteCloser) rpc.ClientCodec

  • NewServerCodec

    1
    2
    // NewServerCodec返回一个在连接上使用JSON-RPC的rpc. ServerCodec。
    func NewServerCodec(conn io.ReadWriteCloser) rpc.ServerCodec

处理客户端连接请求

1
2
3
// ServeConn在单个连接上执行DefaultServer。ServeConn会阻塞，服务该连接直到客户端挂起。
// 调用者一般应另开线程调用本函数："go serveConn(conn)"。ServeConn在该连接使用JSON编解码格式。
func ServeConn(conn io.ReadWriteCloser)

Demo：rpc 通信的三种方式

使用 rpc 方式通信需要通过下面几步才能完成：

Server 端：

  • 初始化一个 Server 对象
  • 注册服务
  • 绑定处理器
  • 监听服务

Client 端：

  • 初始化一个 Client 对象
  • 连接 RPC 服务端
  • 发送请求
  • 接收返回值

下面介绍 net/rpc 的三种连接方式。

1. Http 方式

Server 端：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
package main

import (
	"fmt"
	"log"
	"net/http"
	"net/rpc"
)

type Student struct {
	Name   string
	School string
}
type RpcServer struct{}

func (r *RpcServer) Introduce(student Student, words *string) error {
	fmt.Println("student: ", student)
	*words = fmt.Sprintf("Hello everyone, my name is %s, and I am from %s", student.Name, student.School)
	return nil
}

func main() {
	rpcServer := new(RpcServer)
	// 注册rpc服务
	_ = rpc.Register(rpcServer)
	//把服务处理绑定到http协议上
	rpc.HandleHTTP()
	log.Println("http rpc service start success addr:8080")
	err := http.ListenAndServe(":8080", nil)
	if err != nil {
		log.Fatal(err)
	}
}

Client 端：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
package main

import (
	"fmt"
	"net/rpc"
)

func main() {
	type Student struct {
		Name   string
		School string
	}
    // 连接RPC服务端 Dial会调用NewClient初始化一个Client
	client, err := rpc.DialHTTP("tcp", "127.0.0.1:8080")
	if err != nil {
		panic(err)
	}
	defer client.Close()
	// 发送请求
	var reply string
	err = client.Call("RpcServer.Introduce", &Student{Name: "random_w", School: "Secret"}, &reply)
	if err != nil {
		panic(err)
	}
	fmt.Println(reply)
}

测试：

启动服务端：

1
2
3
$ go run httpRPC.go
2020/07/15 16:28:30 http rpc service start success addr:8080
student:  {random_w Secret}

启动客户端：

1
2
$ go run httpRPCClient.go
Hello everyone, my name is random_w, and I am from Secret

从客户端的日志可以看到我们成功执行了服务端的 Introduce 方法，服务端的日志中也显示了接收到的 Student 信息。

2. TCP 方式

Server 端：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
package main

import (
	"fmt"
	"log"
	"net"
	"net/rpc"
)

type Student struct {
	Name   string
	School string
}
type RpcServer struct{}

func (r *RpcServer) Introduce(student Student, words *string) error {
	fmt.Println("student: ", student)
	*words = fmt.Sprintf("Hello everyone, my name is %s, and I am from %s", student.Name, student.School)
	return nil
}

func main() {
	rpcServer := new(RpcServer)
	// 注册rpc服务
	_ = rpc.Register(rpcServer)
	// 指定rpc模式为TCP模式，地址为127.0.0.1:8081
	tcpAddr, _ := net.ResolveTCPAddr("tcp", "127.0.0.1:8081")
	tcpListen, err := net.ListenTCP("tcp", tcpAddr)
	if err != nil {
		log.Fatal(err)
	}
	log.Println("tcp rpc service start success addr:8081")
	for {
		// 监听Client发送的请求
		conn, err3 := tcpListen.Accept()
		if err3 != nil {
			continue
		}
		// 创建一个goroutine处理请求
		go rpc.ServeConn(conn)
	}
}

Client 端：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
package main

import (
	"fmt"
	"net/rpc"
)

func main() {
	type Student struct {
		Name   string
		School string
	}
	// 连接RPC服务端 Dial会调用NewClient初始化一个Client
	client, err := rpc.Dial("tcp", "127.0.0.1:8081")
	if err != nil {
		panic(err)
	}
	defer client.Close()
	// 发送请求
	var reply string
	err = client.Call("RpcServer.Introduce", &Student{Name: "random_w", School: "Secret"}, &reply)
	if err != nil {
		panic(err)
	}
	fmt.Println(reply)
}

测试：

启动服务端：

1
2
3
$ go run tcpRPC.go
2020/07/15 16:13:21 tcp rpc service start success addr:8081
student:  {random_w Secret}

运行客户端：

1
2
$ go run tcpRPCClient.go
Hello everyone, my name is random_w, and I am from Secret

从客户端的日志可以看到我们成功执行了服务端的 Introduce 方法，服务端的日志中也显示了接收到的 Student 信息。

3. jsonrpc 方式

jsonrpc 方式支持跨语言调用。

Server 端：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
package main

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

type Student struct {
	Name   string
	School string
}

type RpcServer struct{}
func (r *RpcServer) Introduce(student Student, words *string) error {
	fmt.Println("student: ", student)
	*words = fmt.Sprintf("Hello everyone, my name is %s, and I am from %s", student.Name, student.School)
	return nil
}

func main() {
	rpcServer := new(RpcServer)
	// 注册rpc服务
	_ = rpc.Register(rpcServer)
  // jsonrpc是基于TCP协议的，现在他还不支持http协议
	tcpAddr, _ := net.ResolveTCPAddr("tcp", "127.0.0.1:8082")
	tcpListen, err := net.ListenTCP("tcp", tcpAddr)
	if err != nil {
		panic(err)
	}
	log.Println("tcp json-rpc service start success addr:8082")
	for {
		// 监听客户端请求
		conn, err3 := tcpListen.Accept()
		if err3 != nil {
			continue
		}
		go jsonrpc.ServeConn(conn)
	}
}

Client 端：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
package main

import (
	"fmt"
	"net/rpc/jsonrpc"
)

func main() {
	type Student struct {
		Name   string
		School string
	}
	client, err := jsonrpc.Dial("tcp", "127.0.0.1:8082")
	if err != nil {
		panic(err)
	}
	defer client.Close()
	var reply string
	// 发送json格式的数据
	err = client.Call("RpcServer.Introduce", &Student{
		Name:   "random_w",
		School: "Secret",
	}, &reply)
	if err != nil {
		panic(err)
	}
	fmt.Println(reply)
}

测试：

启动服务端：

1
2
3
$ go run jsonRPC.go
2020/07/15 17:15:42 tcp json-rpc service start success addr:8082
student:  {random_w Secret}

运行客户端：

1
2
$ go run jsonRPCClient.go
Hello everyone, my name is random_w, and I am from Secret

从客户端的日志可以看到我们成功执行了服务端的 Introduce 方法，服务端的日志中也显示了接收到的 Student 信息。