[TOC]

golang如何使用原生RPC及微服务简述

微服务

1. 微服务是什么

• 使用一套小服务来开发单个应用的方式，每个服务运行在独立的进程里，一般采用轻量级的通讯机制互联，并且它们可以通过自动化的方式部署

微服务是设计思想，不是量的体现

• 专一的功能
• 代码量并不少
• 架构变复杂

2. 特点是啥

• 专一的职责，例如专注于权限管理
• 轻量级的通信，通信与平台和语言无关，例如http是轻量的
• 隔离性，数据隔离
• 有自己的数据
• 技术多样

3. 微服务架构的优势

• 独立性
• 使用者容易理解
• 技术栈灵活
• 高效团队

4. 微服务架构的不足

• 额外的工作，服务的拆分
• 保证数据一致性
• 增加了沟通成本

微服务生态

1. 硬件层

• 用docker+k8s去解决

2. 通信层

• 网络传输，用RPC（远程过程调用）

  • HTTP传输，GET POST PUT DELETE
  • 基于TCP，更靠底层，RPC基于TCP，Dubbo（18年底改成支持各种语言），Grpc，Thrift

• 需要知道调用谁，用服务注册和发现

  • 需要分布式数据同步：etcd，consul，zk
• 数据传递这里面可能是各种语言，各种技术，各种传递

数据传输协议选型建议

1、对于公司间的系统调用，如果性能要求在100ms以上的服务，基于XML的SOAP协议是一个值得考虑的方案。

2、对于调试环境比较恶劣的场景，采用JSON或XML能够极大的提高调试效率，降低系统开发成本。

3、当对性能和简洁性有极高要求的场景，Protobuf，Thrift，Avro之间具有一定的竞争关系。

4、对于T级别的数据的持久化应用场景，Protobuf和Avro是首要选择。如果持久化后的数据存储在Hadoop子项目里，Avro会是更好的选择。

5、如果需要提供一个完整的RPC解决方案，Thrift是一个好的选择。

6、如果序列化之后需要支持不同的传输层协议，或者需要跨防火墙访问的高性能场景，Protobuf可以优先考虑。

RPC 机制和实现过程

1. RPC机制

服务间通过轻量级的远程过程调用，一般使用HTTP，RPC

• HTTP调用应用层协议，结构相对固定
• RPC的网络协议就相对灵活，并且可以定制

RPC远程过程调用，一般采用C/S 模式，客户端服务器模式，客户端进程，调用服务端进程的程序，服务端进程执行结果返回给客户端，客户端从阻塞状态被唤醒，接收数据，提取数据。

上述过程中，客户端调用服务器的函数，来执行任务，它不知道操作是在本地操作系统进行，还是通过远程过程调用进行的，全程无感。

RPC的基本通信如下：

RPC远程过程调用，需要考虑的问题有如下四点：

• 参数传递
• 通信协议机制
• 出错处理
• 超时处理

2. 参数传递

• 值传递

一般默认是值传递，只需要将参数中的值复制到网络消息中的数据中即可

• 引用传递

比较困难，单纯传递参数的引用是完全没有用意义的，因为引用的地址给到远端的服务器，服务器上的该内存地址完全不是客户端想要的数据，若非要这样处理，客户端还必须把数据的副本传递给到远端服务器，并将它们放到远端服务器内存中，服务器复制引用的地址后，即可进行数据的读取。

可是上述做法很麻烦，且很容易出错，一般RPC不支持直接传递引用

• 数据格式统一问题

需要有一个标准来对所有数据类型进行编解码 ，数据格式可以有隐式类型和显式类型

• 隐式类型

只传递值，不传递变量的名称或 类型

• 显式类型

传递字段的类型和值

常见的传输数据格式有：

• ISO标准的ASN.1
• JSON
• PROTOBUF
• XML

3. 通信协议机制

广义上的协议栈分为共有协议和私有协议

• 共有协议

例如HTTP，SMPP，WEBSERVICE都是共有协议，拥有通用型上，公网传输的能力上 有优势

• 私有协议

内部约定而成的协议，弊端多，但是可以高度的定制化，提升性能，降低成本，提高灵活性和效率。企业内部往往采用私有协议开发

对于协议的制定需要考虑如下5个方面：

• 协议设计

需要考虑哪些问题

• 私有协议的编解码

需要有业务针对性的编解码方式方法，如下有案例

• 命令的定义和命令处理器的选择

协议的过程一般会有2种

1. 负载命令

传输业务具体的数据，如请求参数，响应结果的命令

1. 控制命令

一般为功能管理命令，如心跳命令等

• 命令的协议

一般是使用序列化协议，不同的协议在编码效率和传输效率上都不相同，如

• 通信模式

1. oneway -- 不关心响应，请求线程不会被阻塞
2. sync -- 调用会被阻塞，知道返回结果为止
3. future -- 调用时不会阻塞县线程，获取结果的时候会阻塞线程
4. callback -- 异步调用，不会阻塞线程

出错处理和超时处理

远程过程调用相对本地过程调用出错的概率更大，因此需要考虑到调用失败的各种场景：

• 服务端出错，需要如何处理
• 客户端请求服务时候出现错误或者超时，需要设置合适的重试机制

4. 简易GO语言原生RPC

大概分为如下4个步骤：

• 设计数据结构和方法
• 实现方法
• 注册服务

• 客户端连接服务端，调用服务端的方法

  往下看有golang如何使用原生rpc的案例

rpc调用和服务监控

• RPC相关内容

  • 数据传输：JSON Protobuf thrift
  • 负载：随机算法 轮询 一致性hash 加权
  • 异常容错：健康检测 熔断 限流

• 服务监控

  • 日志收集
  • 打点采样

1. RPC简介

• 远程过程调用（Remote Procedure Call，RPC）是一个计算机通信协议
• 该协议允许运行于一台计算机的程序调用另一台计算机的子程序，而程序员无需额外地为这个交互作用编程
• 如果涉及的软件采用面向对象编程，那么远程过程调用亦可称作远程调用或远程方法调用

2. RPC调用流程

一般情况下，我们会将功能代码在本地直接调用，微服务架构下，我们需要将这个函数作为单独的服务运行，客户端通过网络调用

• 微服务架构下数据交互一般是对内 RPC，对外 REST

• 将业务按功能模块拆分到各个微服务，具有如下优点

  • 提高项目协作效率
  • 降低模块耦合度
  • 提高系统可用性

• 有如下缺点：

  • 开发门槛比较高，比如 RPC 框架的使用、后期的服务监控等工作

3.rpc golang 原生处理方式

最简单的golang原生rpc的使用

golang官方的net/rpc库使用encoding/gob进行编解码，支持tcp和http数据传输方式

server1.go

package main

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

type Happy struct{}

// 计算happy
func (r *Happy) CalHappy(num int, ret *int) error {
   *ret = num * 10
   return nil
}

// 主函数
func main() {
   // new一个服务
   ha := new(Happy)
   // 注册一个Happy的服务
   rpc.Register(ha)
   // 服务处理绑定到http协议上
   rpc.HandleHTTP()
   // 监听服务
   err := http.ListenAndServe(":9999", nil)
   if err != nil {
      log.Panicln(err)
   }
}

client1.go

package main

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

// 主函数
func main() {
   //连接远程rpc服务
   conn, err := rpc.DialHTTP("tcp", ":9999")
   if err != nil {
      log.Fatal(err)
   }
   // 调用服务器方法
   ret := 0
   err2 := conn.Call("Happy.CalHappy", 10, &ret)
   if err2 != nil {
      log.Fatal(err2)
   }
   fmt.Println("开心指数：", ret)
}

结果

golang使用jsonrpc

jsonrpc采用JSON进行数据编解码，支持跨语言调用，jsonrpc库是基于tcp协议实现的，暂不支持http传输方式

server2.go

package main

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

type Happy struct{}

// 计算happy
func (r *Happy) CalHappy(num int, ret *int) error {
   *ret = num * 10
   return nil
}

// 主函数
func main() {
   // new一个服务
   ha := new(Happy)
   // 注册一个Happy的服务
   rpc.Register(ha)
   // 监听服务
   listen, err := net.Listen("tcp", ":9999")
   if err != nil {
      log.Panicln(err)
   }
   // 处理请求
   for {
      con, err := listen.Accept()
      if err != nil {
         continue
      }

      // 专门开一个协程处理相应请求
      go func(con net.Conn) {
         fmt.Println("process new client")
         jsonrpc.ServeConn(con)
      }(con)
   }
}

client2.go

package main

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

// 主函数
func main() {
   //连接远程rpc服务
   conn, err := jsonrpc.Dial("tcp", ":9999")
   if err != nil {
      log.Fatal(err)
   }
   // 调用服务器方法
   ret := 0
   err2 := conn.Call("Happy.CalHappy", 10, &ret)
   if err2 != nil {
      log.Fatal(err2)
   }
   fmt.Println("开心指数：", ret)
}

golang原生rpc自定义协议

例如我们自定义协议，一段数据，前2个字节是数据头，后面的得为真实的数据，如：

• 既然自定义了协议，那么我们发送数据和读取数据的时候就需要遵守我们的协议规定，否则会出问题
• 那么我们做数据传输的时候就会涉及到编码和解码，我们也需要自己封装好编码和解码的函数

写入数据和读取数据的函数封装

// 写入数据
func MyWriteData(con net.Conn, data []byte) (int, error) {
    if con == nil {
        log.Fatal("con is nil")
    }

    buf := make([]byte, 2+len(data))
    // 先写入头部,把真实数据的长度写到头里面
    binary.BigEndian.PutUint16(buf[:2], uint16(len(data)))
    // 再写入数据
    copy(buf[2:], data)
    n, err := con.Write(buf)
    if err != nil {
        log.Fatal("Write error", err)
    }
    return n, nil
}

//读取数据
func MyReadData(con net.Conn) ([]byte, error) {
    if con == nil {
        log.Fatal("con is nil")
    }
    // 协议头2个字节
    myheader := make([]byte, 2)
    // 读取2个字节的协议头
    _, err := io.ReadFull(con, myheader)
    if err != nil {
        log.Fatal("ReadFull error", err)
    }
    //读取真实数据
    // 从头里面读取真实数据的长度
    len := binary.BigEndian.Uint16(myheader)
    data := make([]byte, len)
    _, err = io.ReadFull(con, data)
    if err != nil {
        log.Fatal("ReadFull error", err)
    }
    return data, nil
}

编写编码和解码的函数封装

我们设计成字符串命令 与 具体调用的函数做绑定的方式，这样为接下来的server3.go rpc的实现，打好基础

// 具体的数据结构体
type MyData struct {
    Name   string
    MyArgs []interface{} // 参数列表
}

// 加密
func MyEncode(data *MyData) ([]byte, error) {
    if data == nil {
        log.Fatal("con is nil")
    }
    var bb bytes.Buffer
    buf := gob.NewEncoder(&bb)
    if err := buf.Encode(data); err != nil {
        log.Fatal("Encode error ", err)
    }
    return bb.Bytes(), nil
}

// 解密
func MyDecode(data []byte) (MyData, error) {
    if data == nil {
        log.Fatal("con is nil")
    }
    buf := bytes.NewBuffer(data)
    myDe := gob.NewDecoder(buf)
    var res MyData
    if err := myDe.Decode(&res); err != nil {
        log.Fatal("Decode error ", err)
    }
    return res, nil
}

综合上述功能server端的实现 my_server.go：

package main

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

// 写入数据
func MyWriteData(con net.Conn, data []byte) (int, error) {
    if con == nil {
        log.Fatal("con is nil")
    }

    buf := make([]byte, 2+len(data))
    // 先写入头部,把真实数据的长度写到头里面
    binary.BigEndian.PutUint16(buf[:2], uint16(len(data)))
    // 再写入数据
    copy(buf[2:], data)
    n, err := con.Write(buf)
    if err != nil {
        log.Fatal("Write error", err)
    }
    return n, nil
}

//读取数据
func MyReadData(con net.Conn) ([]byte, error) {
    if con == nil {
        log.Fatal("con is nil")
    }
    // 协议头2个字节
    myheader := make([]byte, 2)
    // 读取2个字节的协议头
    _, err := io.ReadFull(con, myheader)
    if err != nil {
        log.Fatal("ReadFull error", err)
    }
    //读取真实数据
    // 从头里面读取真实数据的长度
    len := binary.BigEndian.Uint16(myheader)
    data := make([]byte, len)
    _, err = io.ReadFull(con, data)
    if err != nil {
        log.Fatal("ReadFull error", err)
    }
    return data, nil
}

// 具体的数据结构体
type MyData struct {
    Name   string
    MyArgs []interface{} // 参数列表
}

// 加密
func MyEncode(data *MyData) ([]byte, error) {
    if data == nil {
        log.Fatal("con is nil")
    }
    var bb bytes.Buffer
    buf := gob.NewEncoder(&bb)
    if err := buf.Encode(data); err != nil {
        log.Fatal("Encode error ", err)
    }
    return bb.Bytes(), nil
}

// 解密
func MyDecode(data []byte) (MyData, error) {
    if data == nil {
        log.Fatal("con is nil")
    }
    buf := bytes.NewBuffer(data)
    myDe := gob.NewDecoder(buf)
    var res MyData
    if err := myDe.Decode(&res); err != nil {
        log.Fatal("Decode error ", err)
    }
    return res, nil
}

// 全局的一个map， 命令与函数做对应关系
var myFun = make(map[string]reflect.Value)

// 注册命令绑定函数
func MyRegister(name string, fn interface{}) {
    if _, ok := myFun[name]; ok { // 说明该命令已经绑定过函数
        return
    }
    myFun[name] = reflect.ValueOf(fn)
    log.Println("reflect.ValueOf(fn) == ", myFun[name])
}

// 服务端执行的方法
func MyRun(addr string) {
    listen, err := net.Listen("tcp", addr)
    if err != nil {
        log.Fatal("Listen is nil")
    }
    log.Println("启动服务端....")
    // 开始阻塞等待客户端的连接
    for {
        con, err := listen.Accept()
        if err != nil {
            log.Println("Accept is nil")
            return
        }
        // 读取数据
        b, err := MyReadData(con)
        if err != nil {
            log.Println("MyReadData error ", err)
            return
        }
        log.Println("MyReadData =============== ")
        // 解析数据
        my, err := MyDecode(b)
        if err != nil {
            log.Println("MyDecode =============== ")
            log.Println("MyDecode error ", err)
            return
        }
        f, ok := myFun[my.Name]
        if !ok {
            fmt.Printf("命令 %s 没有绑定函数n", my.Name)
            return
        }
        // 获取参数
        args := make([]reflect.Value, 0, len(my.MyArgs))
        for _, arg := range my.MyArgs {
            args = append(args, reflect.ValueOf(arg))
            log.Println("reflect.ValueOf(arg) - ", reflect.ValueOf(arg))
        }

        //反射
        res := f.Call(args)
        log.Println("f.Call(args) == ", res)
        // 包装结果数据给到客户端
        out := make([]interface{}, 0, len(res))
        for _, arg := range res {
            log.Println("arg  == ", arg)
            out = append(out, arg.Interface())
        }
        log.Println("out  == ", out)
        // 编码数据
        bb, err := MyEncode(&MyData{Name: my.Name, MyArgs: out})
        if err != nil {
            log.Println("MyEncode error ", err)
            return
        }
        // 将数据写给客户端
        _, err = MyWriteData(con, bb)
        if err != nil {
            log.Println("MyWriteData ======== ")
            log.Println("MyWriteData error ", err)
            return
        }
    }
}
// 客户端通过命令调用函数
func CallRPCFun(con net.Conn, rpcName string, args interface{}) {
    // 通过反射，获取args未初始化的函数原型
    fn := reflect.ValueOf(args).Elem()
    log.Println("fn == ", fn)
    // 需要另一个函数，作用是对第一个函数参数操作
    f := func(args []reflect.Value) []reflect.Value {
        // 处理参数
        inArgs := make([]interface{}, 0, len(args))
        for _, arg := range args {
            inArgs = append(inArgs, arg.Interface())
        }
        // 连接
        // 编码数据
        reqRPC := &MyData{Name: rpcName, MyArgs: inArgs}
        b, err := MyEncode(reqRPC)
        if err != nil {
            log.Println("MyEncode =============== ")
            log.Println("MyEncode error ", err)
        }
        // 写数据
        _, err = MyWriteData(con, b)
        if err != nil {
            log.Println("MyWriteData =============== ")
            log.Fatal("MyWriteData error ", err)
        }
        // 服务端发过来返回值，此时应该读取和解析
        respBytes, err := MyReadData(con)
        if err != nil {
            log.Fatal("MyReadData error ", err)
        }
        // 解码
        res, err := MyDecode(respBytes)
        if err != nil {
            log.Println("MyDecode =============== ")
            log.Fatal("MyDecode error ", err)
        }
        // 处理服务端返回的数据
        outArgs := make([]reflect.Value, 0, len(res.MyArgs))
        for i, arg := range res.MyArgs {
            // 必须进行nil转换
            if arg == nil {
                // reflect.Zero()会返回类型的零值的value
                // .out()会返回函数输出的参数类型
                outArgs = append(outArgs, reflect.Zero(fn.Type().Out(i)))
                continue
            }
            outArgs = append(outArgs, reflect.ValueOf(arg))
        }
        return outArgs
    }

    v := reflect.MakeFunc(fn.Type(), f)

    // 为函数f赋值
    fn.Set(v)
}

// 定义用户对象
type Data struct {
    CmdName string
    Param   string
}

// 用于测试用户查询的方法
func GetData(id int) (Data, error) {
    data := make(map[int]Data)
    // 假数据
    data[0] = Data{"PullInfo", "xiaoxiong"}
    data[1] = Data{"PutInfo", "daxiong"}

    // 查询
    if u, ok := data[id]; ok {
        return u, nil
    }
    return Data{}, fmt.Errorf("%d err", id)
}

// 主函数
func main() {
    // 简单设置log参数
    log.SetFlags(log.Lshortfile | log.LstdFlags)
    
    // rpc 服务端
    // 编码中有一个字段是interface{}时，进行注册
    gob.Register(Data{})
    addr := "127.0.0.1:9999"
    // 创建服务端
    // 将服务端方法，注册一下
    MyRegister("GetData", GetData)

    // 服务端等待调用
    go MyRun(addr)
    
    //-------------我是分割线-----------
    
    // rpc客户端获取连接
    conn, err := net.Dial("tcp", addr)
    if err != nil {
        fmt.Println("Dial err")
        return
    }
    log.Println("客户端拨号成功了，开始调用函数了...")
    // 创建客户端对象
    // 需要声明函数原型
    var getdata func(int) (Data, error)

    CallRPCFun(conn, "GetData", &getdata)
    // 得到查询结果
    u, err := getdata(1)
    if err != nil {
        fmt.Println("getdata err")
        return
    }
    log.Println(u)
    select {}
}

结果：

以上均为学习所得，若有偏误还请指正

技术是开放的，我们的心态也应如此。未来的道路上拥抱变化，勇敢前行。大家一起加油！

我是小魔童哪吒，欢迎吐槽，欢迎沟通