RPC in Golang(一)

[zhangxiang958]

RPC，即远程过程调用，在 grpc 等协议里面，数据的传输是通过二进制的方式进行传输的。而我们在使用 rpc 框架的时候，为了屏蔽底层的数据传输细节，像通过调用本地代码一样完成远程调用，往往会发现有一个 stub 模块。

stub 模块

什么是 stub 模块？其实就是将客户端的请求参数打包成网络信息，然后通过网络发送给服务方。服务端也有 stub 模块，同样也是封装了将网络消息解包，调用本地方法的过程。

简单的 RPC 框架

服务端

客户端传到服务端的是一个函数名，服务端需要维护一个函数名到函数具体逻辑的映射关系。具体是怎么处理映射关系呢？请看下面代码：

package main

import (
	"fmt"
	"reflect"
)

func CallMeBaby(lastName, firstName string) (string, error) {
	res := fmt.Sprintf("%s%s is my baby!", lastName, firstName)
	return res, nil
}

func main() {
	funcs := make(map[string]reflect.Value)
	funcs["CallMeBaby"] = reflect.ValueOf(CallMeBaby)

	req := []reflect.Value{
		reflect.ValueOf("xiang"),
		reflect.ValueOf("zhang"),
	}

	val := funcs["CallMeBaby"].Call(req)
	var rsp []interface{}
	for _, v := range val {
		rsp = append(rsp, v.Interface())
	}

	fmt.Println("result", rsp)
}

像上面这样的代码，将函数与函数名维护在一个 Map 里面，类似这样，如果客户端传过来函数名字，就能通过 Call 方法来将参数传进函数并执行。

类似这样，我们可以提供一个注册的方式，用服务端编写者能够将方法进行注册，将方法放进映射里面。

import (
	"reflect"
)

type Server struct {
    handlers map[string]reflect.Value
}

func (s *Server) Register(name string, f interface{}) {
    if f, exist := s.handlers[name]; exist {
        return;
    }
    
    s.handlers[name] = reflect.ValueOf(f);
}

这里网络传输是基于 TCP 协议的，我们还需要建立一个 TCP 服务来接收网络信息的传输。

func (s *Server) Serve(addr string) {
    server, err := net.Listen("tcp", addr)
	if err != nil {
		log.Fatal("serve start err", err)
	}
	fmt.Printf("serve listening on %s\n", addr)
	defer server.Close()

	for {
		conn, err := server.Accept()
		if err != nil {
			log.Fatal("serve accep err", err)
			continue
		}
		go func() {
            defer conn.Close()
            buf := make([]byte, 4096)
            for {
                n, err := conn.Read(buf)
                if err != nil || n == 0 {
                    log.Fatal("tcp handler", err)
                    break
                }
                receiveStr := strings.TrimSpace(string(buf[0:n]))
                fmt.Printf("receive: %v", receiveStr)
                conn.Write([]byte(receiveStr))
            }
            fmt.Printf("Connection from %v closed. \n", conn.RemoteAddr())
		}()
	}
}

这里需要使用 while 循环来不停地接收客户端传来的信息，golang 中没有 while 关键字，直接使用 for，省略掉循环条件就可以了。

这里需要使用协程来保证服务端的并发。假设这里只是传一个函数名过来，让服务端执行，并返回函数值，相关代码如下：

go func() {
	defer conn.Close()
	buf := make([]byte, 4096)
	for {
		n, err := conn.Read(buf)
		if err != nil || n == 0 {
			log.Fatal("tcp handler", err)
			break
		}
		receiveStr := strings.TrimSpace(string(buf[0:n]))
		fmt.Printf("receive: %v", receiveStr)
		f, RPCExist := s.handlers[receiveStr]
		if !RPCExist {
			err := fmt.Sprintf("rpc %s is not exist", receiveStr)
			conn.Write([]byte(err))
			continue
		}
		req := []reflect.Value{
			reflect.ValueOf("rpc"),
			reflect.ValueOf("name"),
		}
		result := f.Call(req)
		var rsp []interface{}
		for _, v := range result {
			rsp = append(rsp, v.Interface())
		}
		fmt.Println(rsp)
		sss := rsp[0].(string)
		conn.Write([]byte(sss))
	}
	fmt.Printf("Connection from %v closed. \n", conn.RemoteAddr())
}()

此时，服务端编写者已有简易版的 rpc 框架的写法了：

package main

import (
	"fmt"

	rpc "rpc-lite"
)

func test(n, m string) string {
	res := fmt.Sprintln("testS", n, m)
	return res
}

func main() {
	server := rpc.NewServer()

	server.Register("test", test)

	server.Serve()
}

只需要客户端使用 net 模块的 Dial 方法，建立 socket 连接，并传送数据 "test"，就可以执行服务端的代码并拿到对应的返回数据了。