用go从零构建写一个RPC（仿gRPC，tRPC）--- 版本1

希望借助手写这个go的中间件项目，能够理解go语言的特性以及用go写中间件的优势之处，同时也是为了更好的使用和优化公司用到的trpc，并且作者之前也使用过grpc并有一定的兴趣，所以打算从0构建一个rpc系统，对于生产环境已经投入使用的项目抽丝剥茧后，再从0构建，从而更好的理解这个项目和做一个RPC需要注意的地方

打算分为多个版本，从最基本的功能，到逐渐加入新功能和新特性，不断的完善。其中也有一些作者本身的思考优化，其中不足和有误之处还望大家指正

代码地址（目前已经有两个版本）： https://github.com/karatttt/MyRPC

Server端

rpc首先有多个service，每一个service对应多个方法，当请求到来时再正确路由到对应的方法，通过server端处理后返回client端。所以server端主要做的就是一：注册service和对应的method，二：解析配置文件启动Server，三：能够正确路由到来的请求并返回client。

service和Method的注册

grpc和trpc都是使用protobuf作为序列化格式，这里我们的项目也用protobuf格式进行序列化，成熟的rpc项目正常会有对应的工具，我们写好proto文件和对应的service的实现类后，使用自动化构建工具可以生成桩代码，包括以下部分：

消息类（Message Struct）：把你 .proto 里面定义的请求、响应对象变成对应的语言结构体，比如 UserRequest、UserReply
服务接口（Service Interface）：把你 .proto 里面定义的方法变成一组接口或基类，供你实现，比如 GetUser(ctx, req)
客户端 Stub ：客户端可以直接用来调用远程方法的代码（自动封装了序列化、网络传输、重试等逻辑），类似于java的动态代理
服务端 Stub ：服务端接收到请求后，自动反序列化，然后回调你实现的业务逻辑，也类似于java的动态代理

这里我们尝试通过一个proto文件，自己实现一个server端的桩代码

syntax = "proto3";

package myrpc.helloworld;
option go_package="/pb";

service Greeter {
  rpc Hello (HelloRequest) returns (HelloReply) {}
}

message HelloRequest {
  string msg = 1;
}

message HelloReply {
  string msg = 1;
}

第一步，根据填的方法写一个接口：

// 具体方法接口
type HelloServer interface {
	// SayHello is the method that will be called by the client.
	Hello(req *HelloRequest) (*HelloReply, error)
}

第二步，我们对每一个方法写一个handler，写实际上的处理逻辑，即如何反序列化，然后回调实际写的业务逻辑，再返回结构体。具体的如何序列化反序列化实现我们后面再看

func HelloServer_Hello_Handler(srv interface{}, req []byte) (interface{}, error) {
	
	// 这里的srv是HelloServer的实现类，我们自己写的
	// 通过类型断言将srv转换为HelloServer类型
	helloServer, ok := srv.(HelloServer)
	if !ok {
		return nil, fmt.Errorf("HelloServer_Hello_Handler: %v", "type assertion failed")
	}
	// 调用HelloServer的Hello方法
	// 将req反序列化
	reqBody := &HelloRequest{}
	err := codec.Unmarshal(req, reqBody)
	if err != nil {
		return nil, fmt.Errorf("HelloServer_Hello_Handler: %v", err)
	}
	// 调用实际我们写的业务逻辑
	reply, err := helloServer.Hello(reqBody)
	
	if err != nil {
		fmt.Printf("HelloServer_Hello_Handler: %v", err)
		return nil, err
	}
	return reply, nil
}

第三步，我们写了Handler，当然要让server端能够路由到这个Handler，所以这个Handler需要绑定一个方法名和服务名，作为key保存再server端的一个map里，这样就可以正确路由。所以我们可以写一个方法，将这个映射关系注册到server里。

这个server的Register方法，我们后面再来实现。

// 映射关系
var HelloServer_ServiceDesc = server.ServiceDesc{
		ServiceName: "helloworld",
		HandlerType: (*HelloServer)(nil),
		Methods: []server.MethodDesc{
			{
				MethodName: "Hello",
				// 当接受到客户端调用的Hello方法时，server将会调用这个方法
				Func:    HelloServer_Hello_Handler,
			},
		},
	}

// 绑定方法
func RegisterHelloServer(s *server.Server, svr interface{}) error {
	if err := s.Register(&HelloServer_ServiceDesc, svr); err != nil {
		panic(fmt.Sprintf("Greeter register error:%v", err))
	}
	return nil
}

Server端的启动

Server启动的时候，需要根据我们写的配置文件以得知每一个service的name，以及他们对应的ip和端口号（当然后续还有其他的配置），正常多个service的ip和端口号是一样的，也就是说serve启动的时候，统一暴露一个端口用于rpc调用。
所以server启动的流程是：一：读取配置，二：根据配置名创建多个service并保存

func NewServer() *Server {
	// 1. 创建一个Server实例
	server := &Server{
		services: make(map[string]Service),
	}

	// 2. 读取配置文件
	config, err := loadConfig("./rpc.yaml")
	if err != nil {
		fmt.Print("读取配置文件出错")
	}

	// 3. 创建服务
	for _, svc := range config.Server.Service {
		// 创建服务，这里创建了service实例
		service := NewService(svc.Name, WithAddress(fmt.Sprintf("%s:%d", svc.IP, svc.Port)))

		// 添加到服务映射
		server.services[svc.Name] = service
	}
	return server
}

Service类的实现

前面server端启动的时候创建了所有的service类，这里我们看看具体service应该做什么。

当请求进来时，首先找到service，再找到对应的Method，所以service应该持有method的map，以及在这里实现前面提到的Register逻辑。

同时，每一个service应该有一个serve方法，即提供服务，就是这里开始监听请求，路由和处理，这个后续会详细展开。

我们再为service实现Handler接口，赋予处理业务逻辑的能力，这个接口就是为了路由找到service里的method并调用它，这个Handler详细我们后面再看

// 定义接口，提供一些服务的注册和开启服务的功能
type Service interface {
	// Register registers a service with the server.
	// The serviceName is the name of the service, and service is the implementation of the service.
	Register(serviceDesc *ServiceDesc, service interface{}) error
	// Serve starts the server and listens for incoming connections.
	Serve(address string) error
}
// 定义一个Handler接口，service实现了这个接口
type Handler interface {
	Handle(ctx context.Context, frame []byte) (rsp []byte, err error)
}

我们先看看比较简单的regsiter方法，虽然registerMethods看起来复杂，但是实际上就是将前面桩代码的Handler作为一个函数存在map里

// 实现service的Register方法，填充service的各个属性
func (s *service) Register(serviceDesc *ServiceDesc, service interface{}) error {

	// 初始化Transport
	s.opts.Transport = transport.DefaultServerTransport

	s.registerMethods(serviceDesc.Methods, service)

	return nil
}

// 注册普通方法
func (s *service) registerMethods(methods []MethodDesc, serviceImpl interface{}) error {
	for _, method := range methods {
		if _, exists := s.handler[method.MethodName]; exists {
			return fmt.Errorf("duplicate method name: %s", method.MethodName)
		}
		s.handler[method.MethodName] = func(req []byte) (rsp interface{}, err error) {
			if fn, ok := method.Func.(func(svr interface{}, req []byte) (rsp interface{}, err error)); ok {
				// 这里调用的就是rpc.go里面的实际的handler方法
				return fn(serviceImpl, req)
			}
			return nil, fmt.Errorf("method.Func is not a valid function")
		}
	}
	return nil
}

Service类的Server方法处理请求

func (s *service) Serve(address string) error {
	fmt.Printf("Server is listening on %s\n", address)

	// 将service作为Handler传入transport，后续接收到请求，会调用service的Handle方法
	s.opts.Transport.RegisterHandler(s)
	err := s.opts.Transport.ListenAndServe(context.Background(), "tcp", address)
	if err != nil {
		return fmt.Errorf("failed to listen: %v", err)
	}
	return nil
}

这个Serve方法会在Server端启动的时候，依次触发每一个service类的这个Serve方法，意即为每一个service提供处理请求的能力
这里做了一个serverTransport主要负责网络请求，我们重点关注ListenAndServe

// ListenAndServe 监听并处理 TCP 连接
func (t *serverTransport) ListenAndServe(ctx context.Context, network, address string) error {

	ln, err := net.Listen(network, address)
	if err != nil {
		return fmt.Errorf("failed to listen: %w", err)
	}
	defer ln.Close()

	go func() {
		<-ctx.Done()
		ln.Close()
	}()

	return t.serveTCP(ctx, ln)
}

// serveTCP 处理 TCP 连接
func (t *serverTransport) serveTCP(ctx context.Context, ln net.Listener) error {
	fmt.Print("开始监听TCP连接")
	for {
		conn, err := ln.Accept()
		if err != nil {
			select {
			case <-ctx.Done():
				return nil // 退出监听
			default:
				fmt.Println("accept error:", err)
			}
			continue
		}
		go t.handleConnection(ctx, conn)
	}
}
// handleConnection 处理单个连接
func (t *serverTransport) handleConnection(ctx context.Context, conn net.Conn) {
	//TODO 这里可以做一个处理业务逻辑的协程池
	// 实际上每个连接一个协程，同时负责读取请求并直接处理业务逻辑也是可行的，读取请求时如果阻塞，Go调度器会自动切换到其他协程执行
	// 但是协程池可以限制同时处理业务逻辑的协程数量，避免请求量大时，过多协程导致的资源消耗

	// 这里是处理完一个请求就释放连接，后续可以考虑长连接
	defer conn.Close()
	fmt.Println("New connection from", conn.RemoteAddr())
	// 读取帧
	frame, err := codec.ReadFrame(conn)
	if err != nil {
		fmt.Println("read frame error:", err)
		return
	}
	// 调用service的Handler执行结果
	response, err := t.ConnHandler.Handle(ctx, frame)
	if err != nil {
		fmt.Println("handle error:", err)
		return
	}
	// 发送响应，此时已经是完整帧
	conn.Write(response)
}

以上的代码简单来说就是，开启一个coonection，for循环accept请求，一旦请求到达，开启协程进行实际的业务逻辑处理
这个 t.ConnHandler.Handle(ctx, frame)，实际上就是service里的Handler方法，当transport收到请求时，回到我们的service的Handler方法执行。
对于codec.ReadFrame(conn)我们下面重点看看

Service类的Handler方法

接收到请求，我们的处理过程应该是这样：

接收codec.ReadFrame后得到原始字节流（frame）
解码frame
调用对应的业务方法 handler（其间反序列化）
把业务返回结果序列化
编码生成frame返回给调用方

首先设计Frame结构如下：
在这里插入图片描述

ReadFrame

即根据帧头读取一段完整的自定义协议数据，解决半包和粘包问题，先读16字节的帧头解析各字段，校验魔数和版本号，再根据帧头中记录的协议数据长度和消息体长度继续读取剩下的内容，最后把帧头和帧体拼成一个完整的字节数组返回。


func ReadFrame(conn net.Conn) ([]byte, error) {
	buf := bufio.NewReader(conn)
	// 读取帧头
	headerBuf := make([]byte, HeaderLength)
	n, err := io.ReadFull(buf, headerBuf)
	if err != nil {
		return nil, fmt.Errorf("read header error: %v, read %d bytes", err, n)
	}
	// 正确解析所有字段
	header := FrameHeader{
		MagicNumber:    binary.BigEndian.Uint16(headerBuf[0:2]),
		Version:        headerBuf[2],
		MessageType:    headerBuf[3],
		SequenceID:     binary.BigEndian.Uint32(headerBuf[4:8]),
		ProtocolLength: binary.BigEndian.Uint32(headerBuf[8:12]),
		BodyLength:     binary.BigEndian.Uint32(headerBuf[12:16]),
	}
	if header.MagicNumber != MagicNumber {
		return nil, fmt.Errorf("invalid magic number: %d", header.MagicNumber)
	}
	if header.Version != Version {
		return nil, fmt.Errorf("unsupported version: %d", header.Version)
	}
	// 读取协议数据 + 消息体
	frameBody := make([]byte, header.ProtocolLength+header.BodyLength)
	_, err = io.ReadFull(buf, frameBody)
	if err != nil {
		return nil, fmt.Errorf("read body error: %v", err)
	}
	// 拼接完整帧
	frame := append(headerBuf, frameBody...)
	return frame, nil
}

Decode

读取到Frame后，需要解析出其中的消息体，并将读取到的协议数据存起来

func (c *servercodec) Decode(msg internel.Message, frame []byte) ([]byte, error) {
	// 解析帧头
	header := FrameHeader{
		MagicNumber:    binary.BigEndian.Uint16(frame[0:]),
		Version:        frame[2],
		MessageType:    frame[3],
		SequenceID:     binary.BigEndian.Uint32(frame[4:]),
		ProtocolLength: binary.BigEndian.Uint32(frame[8:]),
		BodyLength:     binary.BigEndian.Uint32(frame[12:]),
	}

	// 验证魔数和版本
	if header.MagicNumber != MagicNumber {
		return nil, fmt.Errorf("invalid magic number: %d", header.MagicNumber)
	}
	if header.Version != Version {
		return nil, fmt.Errorf("unsupported version: %d", header.Version)
	}

	// 提取协议数据
	protocolData := frame[HeaderLength : HeaderLength+header.ProtocolLength]

	// 解析协议数据
	proto, err := DeserializeProtocolData(protocolData)
	if err != nil {
		return nil, fmt.Errorf("parse protocol data error: %v", err)
	}

	// 设置到消息中
	msg.WithServiceName(proto.ServiceName)
	msg.WithMethodName(proto.MethodName)

	// 返回消息体
	return frame[HeaderLength+header.ProtocolLength:], nil
}

Unmarshal

得到消息体后，还是字节数组，这个时候根据protobuf的格式，反序列化成对应的结构体（这个方法的调用在前面的桩代码的HelloServer_Hello_Handler里）

// Unmarshal 将 protobuf 字节数组反序列化为结构体
func Unmarshal(rspDataBuf []byte, rspBody interface{}) error {
	msg, ok := rspBody.(proto.Message)
	if !ok {
		return fmt.Errorf("Unmarshal: rspBody does not implement proto.Message")
	}
	return proto.Unmarshal(rspDataBuf, msg)
}