Golang——RPC

一. RPC简介

• 远程过程调用(Remote Procedure Call，RPC)是一个计算机通信协议。
• 该协议运行于一台计算机的程序调用另外一台计算机的子程序，而程序员无需额外的为这个交互作用编程。
• 如果涉及的软件采用面向对象编程，那么远程过程调用亦可称作远程调用或远程方法调用。
• 远程过程调用是一个分布式计算的客户端-服务器(Client/Server)的例子。
• 远程过程调用总是由客户端对服务器发出一个执行若干过程请求，并由客户端提供参数，执行结果将返回客户端。由于存在各式各样的变体和差异，对应的派生了各式远程过程调用协议，而他们并不互相兼容。
• RPC本身是client-server模型，也是一种request-response协议。有些实现扩展了远程调用的模型。实现了双向的服务调用，但是不管怎么样，调用过程还是由一个客户端发起，服务端提供响应。
• 调用过程为：
  • client调用client stub，这是一次本地过程调用
  • client stub将参数打包成一个消息，然后发送这个消息
  • client所在的系统将消息发送给server
  • server的系统将收到的包传给server stub
  • server stub解包得到参数
  • 最后server stub调用服务过程，返回结果按照相反的步骤传给client

二. golang中如何实现RPC

• golang中实现RPC非常简单，官方提供了封装好的库，还有一些第三方的库。
• golang官方的net/rpc库使用encoding/gob进行编解码，支持tcp和http数据传输方式。由于其它语言不支持gob编解码，所以golang的RPC只支持golang开发的服务器与客户端之间的交互。
• 官方还提供了net/rpc/jsonrpc库实现RPC方法，jsonrpc采用json进行数据编解码，因而支持跨语言调用，目前jsonrpc库是基于tcp协议实现的，暂不支持http传输方式。

例题：golang实现RPC程序，实现求矩形面积和周长。

服务器：

        注册服务，监听请求，处理请求。

package main

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

// 接收参数
type Params struct {
	Length int
	Heigth int
}

// 用于注册
type Rect struct{}

func (r *Rect) Area(p Params, ret *int) error {
	fmt.Println("参数：", p.Heigth, p.Length)
	*ret = p.Heigth * p.Length
	return nil
}

func (r *Rect) Perimeter(p Params, ret *int) error {
	fmt.Println("参数：", p.Heigth, p.Length)
	*ret = 2 * (p.Heigth + p.Length)
	return nil
}

func main() {
	//1. 注册服务
	rect := new(Rect)
	rpc.Register(rect)

	//2. 服务器处理绑定到http协议上
	rpc.HandleHTTP()
	//3. 监听服务
	err := http.ListenAndServe(":8080", nil)
	if err != nil {
		log.Panicln(err)
	}
}

客户端：

        连接rpc服务，发送请求调用方法。

package main

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

// 参数字段要和服务器的一样
type Params struct {
	Length int
	Heigth int
}

func main() {
	conn, err := rpc.DialHTTP("tcp", ":8080")
	if err != nil {
		log.Fatal(err)
	}

	ret := 0
	p := Params{
		20,
		30,
	}
	//调用服务器的方法
	err = conn.Call("Rect.Area", p, &ret)
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println("面积", ret)

	err = conn.Call("Rect.Perimeter", p, &ret)
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println("周长", ret)
}

• golang写RPC程序必须符合4个基本条件，不然RPC用不了
  • 结构体字段首字母必须大写，可以别人调用
  • 函数名必须首字母大写
  • 函数(服务方法)第一参数是接受参数，第二个参数是返回客户端的参数，必须是指针类型
  • 函数(服务方法)还必须有一个返回值error

另外，net/rpc/jsonrpc库通过json格式编解码，支持跨语言调用。

服务端：

package main

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

// 接收参数
type Params struct {
	A int
	B int
}

type Rect struct{}

func (r *Rect) Area(p Params, ret *int) error {
	*ret = p.A * p.B
	return nil
}

func (r *Rect) Perimeter(p Params, ret *int) error {
	*ret = 2 * (p.A + p.B)
	return nil
}

func main() {
	//注册服务
	rpc.Register(new(Rect))
	//监听连接
	l, err := net.Listen("tcp", ":8080")
	if err != nil {
		log.Fatal(err)
	}

	for {
		//接收客户端连接
		conn, err := l.Accept()
		if err != nil {
			continue
		}
		//创建协程处理请求
		go func(conn net.Conn) {
			fmt.Println("get a new client")
			//只是处理服务，没有监听和接收请求
			jsonrpc.ServeConn(conn)
		}(conn)
	}
}

客户端：

package main

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

// 参数字段要和服务器的一样
type Params struct {
	A int
	B int
}

func main() {
	//远程连接rpc服务
	conn, err := jsonrpc.Dial("tcp", ":8080")
	if err != nil {
		log.Fatal(err)
	}
	//调用方法
	ret := 0
	err = conn.Call("Rect.Area", Params{10, 20}, &ret)
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println("ret=", ret)

	err = conn.Call("Rect.Perimeter", Params{10, 20}, &ret)
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println("ret=", ret)
}

三.  RPC调用流程

• 微服务架构下数据交互一般是对内RPC，对外REST
• 将业务按功能模块拆分到各个微服务，具有提高项目协作效率，降低模块耦合度，提高系统可用性等优点
• 一般情况下，我们将功能代码在本地直接调用，微服务框架下，我们需要将这个函数作为单独的服务运行，客户端通过网络调用。

我理解的微服务(RPC)，将项目中必要的功能注册成一个服务，客户端可以直接调用。

四. 网络传输数据格式

• 两端要约定好数据包的格式
• 成熟的RPC框架会有自定义传输协议，这里网络传输格式定义如下，前面是固定长度的消息头，后面是变长消息体

• 自己定义数据格式的读写

package rpc

import (
	"encoding/binary"
	"fmt"
	"io"
	"net"
)

type Session struct {
	Conn net.Conn
}

func (s *Session) Write(data []byte) error {
	//定义写数据格式
	//4字节包头 + 变长数据
	buf := make([]byte, 4+len(data))
	//写入头部
	binary.BigEndian.PutUint32(buf[:4], uint32(len(data)))
	//将有效数据写到头部后面
	copy(buf[4:], data)
	//发送
	_, err := s.Conn.Write(buf)
	if err != nil {
		return err
	}
	return nil
}

func (s *Session) Read() ([]byte, error) {
	//读取头部
	header := make([]byte, 4)
	//按长度读取数据
	_, err := io.ReadFull(s.Conn, header)
	if err != nil {
		return nil, err
	}

	//读取数据，报头保存的是有效数据长度
	dataLen := binary.BigEndian.Uint32(header)
	fmt.Println(dataLen)
	data := make([]byte, dataLen)

	_, err = io.ReadFull(s.Conn, data)
	if err != nil {
		return nil, err
	}
	return data, nil
}

测试代码：

package rpc

import (
	"errors"
	"fmt"
	"net"
	"sync"
	"testing"
)

func TestSession_ReadWrite(t *testing.T) {
	//地址和数据
	addr := "127.0.0.1:8080"
	my_data := "hello"

	//等待组
	wg := sync.WaitGroup{}
	wg.Add(2)
	go func() {
		//监听
		defer wg.Done()
		l, err := net.Listen("tcp", addr)
		if err != nil {
			t.Fatal(err)
		}
		//接收连接
		conn, err := l.Accept()
		if err != nil {
			t.Fatal(err)
		}
		//写数据
		s := Session{Conn: conn}
		err = s.Write([]byte(my_data))
		if err != nil {
			t.Fatal(err)
		}
	}()

	go func() {
		defer wg.Done()
		conn, err := net.Dial("tcp", addr)
		if err != nil {
			t.Fatal(err)
		}

		s := Session{Conn: conn}
		data, err := s.Read()
		if err != nil {
			t.Fatal(err)
		}

		//校验
		if string(data) != my_data {
			t.Fatal(errors.New("数据错误"))
		}
		fmt.Println(string(data))
	}()
	wg.Wait()
}

 编码和解码：

func Encode(data RPCData) ([]byte, error) {
	//得到字节数组编码器
	var buf bytes.Buffer
	buffEnc := gob.NewEncoder(&buf)
	//编码器对数据编码
	if err := buffEnc.Encode(data); err != nil {
		return nil, err
	}

	return buf.Bytes(), nil
}

func Decode(data []byte) (RPCData, error) {
	buff := bytes.NewBuffer(data)
	//得到字节数组解码器
	buffDec := gob.NewDecoder(buff)
	//解码器对数据解码
	var res RPCData
	if err := buffDec.Decode(&res); err != nil {
		return res, err
	}

	return res, nil
}

五. 实现RPC服务端

        要实现的一个功能为，接收到客户端发过来需要调用的函数和参数，实现对应函数的调用，将结果返回。

• 服务端接收的数据包括
  • 调用的函数名，参数，返回值
• 服务器端需要解决的问题
  • 需要维护一个map，来保存调用的函数
• 服务器的核心功能
  • 维护函数map
  • 接收到客户端的数据，并解析
  • 找到调用的函数，将返回值打包穿给客户端

package main

import (
	"bytes"
	"encoding/binary"
	"encoding/gob"
	"errors"
	"fmt"
	"io"
	"log"
	"net"
	"reflect"
)

// 定义交互的数据结构
type RPCData struct {
	//访问的函数
	Name string
	//参数
	Args []interface{}
}

func encode(data RPCData) ([]byte, error) {
	//得到字节数组编码器
	var buf bytes.Buffer
	gob.Register(User{})
	buffEnc := gob.NewEncoder(&buf)
	//编码器对数据编码
	if err := buffEnc.Encode(data); err != nil {
		return nil, err
	}

	return buf.Bytes(), nil
}

func decode(data []byte) (RPCData, error) {
	buff := bytes.NewBuffer(data)
	//得到字节数组解码器
	gob.Register(User{})
	buffDec := gob.NewDecoder(buff)
	//解码器对数据解码
	var res RPCData
	if err := buffDec.Decode(&res); err != nil {
		return res, err
	}

	return res, nil
}

// 发送和接收数据结构
type Session struct {
	Conn net.Conn
}

func NewSession(conn net.Conn) *Session {
	return &Session{Conn: conn}
}

func (s *Session) Write(data []byte) error {
	//定义写数据格式
	//4字节包头 + 变长数据
	buf := make([]byte, 4+len(data))
	//写入头部
	binary.BigEndian.PutUint32(buf[:4], uint32(len(data)))
	//将有效数据写到头部后面
	copy(buf[4:], data)
	//发送
	_, err := s.Conn.Write(buf)
	if err != nil {
		return err
	}
	return nil
}

func (s *Session) Read() ([]byte, error) {
	//读取头部
	header := make([]byte, 4)
	//按长度读取数据
	_, err := io.ReadFull(s.Conn, header)
	if err != nil {
		return nil, err
	}

	//读取数据，报头保存的是有效数据长度
	dataLen := binary.BigEndian.Uint32(header)
	fmt.Println(dataLen)
	data := make([]byte, dataLen)

	_, err = io.ReadFull(s.Conn, data)
	if err != nil {
		return nil, err
	}
	return data, nil
}

// 服务器
type Server struct {
	//访问地址
	addr string
	//维护函数map
	funcs map[string]reflect.Value
}

func NewServer(addr string) *Server {
	return &Server{addr: addr, funcs: make(map[string]reflect.Value)}
}

func (s *Server) Register(key string, value interface{}) error {
	if _, ok := s.funcs[key]; ok {
		return errors.New(key + "已存在")
	}

	s.funcs[key] = reflect.ValueOf(value)
	return nil
}

func (s *Server) Run() {
	//监听
	ls, err := net.Listen("tcp", s.addr)
	if err != nil {
		log.Printf("监听addr:%s, 失败", s.addr)
		return
	}

	for {
		conn, err := ls.Accept()
		if err != nil {
			return
		}

		//读数据
		session := NewSession(conn)
		data, err := session.Read()
		if err != nil {
			return
		}
		//解码
		rpc_data, err := decode(data)
		if err != nil {
			return
		}

		//查找访问的函数
		f, ok := s.funcs[rpc_data.Name]
		if !ok {
			log.Printf("%s不存在
", rpc_data.Name)
			continue
		}
		//参数
		args := make([]reflect.Value, 0, len(rpc_data.Args))
		for _, v := range rpc_data.Args {
			if v != nil {
				args = append(args, reflect.ValueOf(v))
			}
		}
		//反射调用方法
		//返回value类型，用于客户端传递返回结果，out是所有返回结果
		out := f.Call(args)
		outInters := make([]interface{}, 0, len(out))
		for _, v := range out {
			outInters = append(outInters, v.Interface())
		}
		fmt.Println("args: ", args, "-", len(args), "-", outInters)

		//发送给客户端
		//编码
		respRpcData := RPCData{Name: rpc_data.Name, Args: outInters}
		respdata, err := encode(respRpcData)
		if err != nil {
			log.Printf("encode fail %v
", err)
			continue
		}

		err = session.Write(respdata)
		if err != nil {
			log.Printf("Write fail %v
", err)
			continue
		}
	}

}

type User struct {
	Name string
	Age  int
}

func queryUser(uid int) (User, error) {
	users := make(map[int]User)
	users[0] = User{"zs", 10}
	users[1] = User{"ls", 20}
	users[2] = User{"ww", 25}
	users[3] = User{"lc", 18}
	fmt.Println(uid)
	if v, ok := users[uid]; ok {
		fmt.Println(v)
		return v, nil
	}
	return User{}, errors.New("Not Found")
}

func main() {
	s := NewServer("127.0.0.1:8080")

	//给服务器注册函数
	s.Register("queryUser", queryUser)

	s.Run()
}

六. 实现RPC客户端

• 客户端只有函数原型，使用reflect.MakeFunc()可以完成原型到函数的调用。
  • 即reflect.MakeFunc()函数可以将一个函数类型和函数实现结合起来，通过函数原型调用对应函数实现。

package main

import (
	"bytes"
	"encoding/binary"
	"encoding/gob"
	"fmt"
	"io"
	"log"
	"net"
	"reflect"
)

type RPCData struct {
	//访问的函数
	Name string
	//参数
	Args []interface{}
}

func encode(data RPCData) ([]byte, error) {
	//得到字节数组编码器
	var buf bytes.Buffer
	//注册接口类型
	gob.Register(User{})
	buffEnc := gob.NewEncoder(&buf)
	//编码器对数据编码
	if err := buffEnc.Encode(data); err != nil {
		return nil, err
	}

	return buf.Bytes(), nil
}

func decode(data []byte) (RPCData, error) {
	buff := bytes.NewBuffer(data)
	//得到字节数组解码器
	gob.Register(User{})
	buffDec := gob.NewDecoder(buff)
	//解码器对数据解码
	var res RPCData
	if err := buffDec.Decode(&res); err != nil {
		return res, err
	}

	return res, nil
}

type Session struct {
	Conn net.Conn
}

func NewSession(conn net.Conn) *Session {
	return &Session{Conn: conn}
}

func (s *Session) Write(data []byte) error {
	//定义写数据格式
	//4字节包头 + 变长数据
	buf := make([]byte, 4+len(data))
	//写入头部
	binary.BigEndian.PutUint32(buf[:4], uint32(len(data)))
	//将有效数据写到头部后面
	copy(buf[4:], data)
	//发送
	_, err := s.Conn.Write(buf)
	if err != nil {
		return err
	}
	return nil
}

func (s *Session) Read() ([]byte, error) {
	//读取头部
	header := make([]byte, 4)
	//按长度读取数据
	_, err := io.ReadFull(s.Conn, header)
	if err != nil {
		return nil, err
	}

	//读取数据，报头保存的是有效数据长度
	dataLen := binary.BigEndian.Uint32(header)
	fmt.Println(dataLen)
	data := make([]byte, dataLen)

	_, err = io.ReadFull(s.Conn, data)
	if err != nil {
		return nil, err
	}
	return data, nil
}

// 客户端
type Client struct {
	Conn net.Conn
}

func NewClient(conn net.Conn) *Client {
	return &Client{Conn: conn}
}

// fname为访问服务器函数名
// fptr为函数原型
func (c *Client) callRpc(fname string, fptr interface{}) {

	//获取函数原型
	fnptr := reflect.ValueOf(fptr).Elem()
	//函数实现，发送数据给服务器，收到服务端数据
	//args调用时传进来的参数
	//返回值为得到的结果
	f := func(args []reflect.Value) []reflect.Value {
		//获得参数
		argSlice := make([]interface{}, len(args))
		for _, v := range args {
			argSlice = append(argSlice, v.Interface())
		}

		session := NewSession(c.Conn)

		//发送到服务器
		rpcData := RPCData{Name: fname, Args: argSlice}
		//编码
		data, err := encode(rpcData)
		if err != nil {
			log.Println("encode fail", err)
			return nil
		}
		err = session.Write(data)
		if err != nil {
			log.Println("Write fail", err)
			return nil
		}

		//接收服务器响应
		respData, err := session.Read()
		if err != nil {
			log.Println("Read fail", err)
			return nil
		}
		//结果在Args里
		respRpcData, err := decode(respData)
		if err != nil {
			log.Println("decode fail", err)
			return nil
		}

		res := make([]reflect.Value, 0, len(respRpcData.Args))
		for i, v := range respRpcData.Args {
			if v == nil {
				//因为返回err可能为nil
				//进行nil转换
				//reflect.Zero会返回类型的零值的value
				//.out会返回函数输出的参数类型
				res = append(res, reflect.Zero(fnptr.Type().Out(i)))
			} else {
				res = append(res, reflect.ValueOf(v))
			}
		}
		return res
	}

	real_fn := reflect.MakeFunc(fnptr.Type(), f)
	//为函数原型赋值函数实现
	fnptr.Set(real_fn)
}

type User struct {
	Name string
	Age  int
}

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

	c := NewClient(conn)
	var query func(int) (User, error)
	c.callRpc("queryUser", &query)
    //进行查询
	u, err := query(2)
	if err != nil {
		log.Fatal(err)
		return
	}
	fmt.Println(u)
}