Golang實現簡易的rpc調用
RPC(Remote Procedure Call Protocol)遠程過程調用協議。 一個通俗的描述是：客戶端在不知道調用細節的情況下，調用存在於遠程計算機上的某個對象，就像調用本地應用程序中的對象一樣。 比較正式的描述是：一種通過網絡從遠程計算機程序上請求服務，而不需要瞭解底層網絡技術的協議 從使用的方面來說，服務端和客戶端通過TCP/UDP/HTTP等通訊協議通訊，在通訊的時候客戶端指定好服務端的方法、參數等信息通過序列化傳送到服務端，服務端可以通過已有的元信息找到需要調用的方法，然後完成一次調用後序列化返回給客戶端(rpc更多的是指服務與服務之間的通信，可以使用效率更高的協議和序列化格式去進行，並且可以進行有效的負載均衡和熔斷超時等，因此跟前後端之間的web的交互概念上是有點不一樣的) 用一張簡單的圖來表示
開始
本文隻實現一個rpc框架基本的功能，不對性能做保證，因此盡量使用go原生自帶的net/json庫等進行操作，對使用方面不做stub（偷懶，隻使用簡單的json格式指定需要調用的方法），用最簡單的方式實現一個簡易rpc框架，也不保證超時調用和服務發現等集成的邏輯，服務發現可以參考下文 本文代碼地址(https://github.com/wuhuZhao/rpc_demo)
實現兩點之間的通訊(transport)
本段先實現兩端之間的通訊，隻確保兩個端之間能互相通訊即可 server.go
package server
import (
"fmt"
"log"
"net"
)
// Server: transport底層實現，通過Server去接受客戶端的字節流
type Server struct {
ls net.Listener
port int
}
// NewServer: 根據端口創建一個server
func NewServer(port int) *Server {
s := &Server{port: port}
s.init()
return s
}
// init: 初始化服務端連接
func (s *Server) init() {
l, err := net.Listen("tcp", fmt.Sprintf("127.0.0.1:%d", s.port))
if err != nil {
panic(err)
}
s.ls = l
}
// Start: 啟動服務端的端口監聽，采取一個conn一個g的模型，沒有使用reactor等高性能模型
func (s *Server) Start() {
go func() {
log.Printf("server [%s] start....", s.ls.Addr().String())
for {
conn, err := s.ls.Accept()
if err != nil {
panic(err)
}
go func() {
buf := make([]byte, 1024)
for {
idx, err := conn.Read(buf)
if err != nil {
panic(err)
}
if len(buf) == 0 {
continue
}
// todo 等序列化的信息
log.Printf("[conn: %v] get data: %v\n", conn.RemoteAddr(), string(buf[:idx]))
}
}()
}
}()
}
// Close: 關閉服務監聽
func (s *Server) Close() error {
return s.ls.Close()
}
// Close: 關閉服務監聽
func (s *Server) Close() error {
return s.ls.Close()
}
client.go
package client
import (
"fmt"
"log"
"net"
"unsafe"
)
type Client struct {
port int
conn net.Conn
}
func NewClient(port int) *Client {
c := &Client{port: port}
c.init()
return c
}
// init: initialize tcp client
func (c *Client) init() {
conn, err := net.Dial("tcp", fmt.Sprintf("127.0.0.1:%d", c.port))
if err != nil {
panic(err)
}
c.conn = conn
}
func (c *Client) Send(statement string) error {
_, err := c.conn.Write(*(*[]byte)(unsafe.Pointer(&statement)))
if err != nil {
panic(err)
}
return nil
}
// Close: use to close connection
func (c *Client) Close() error {
return c.conn.Close()
}
使用main.go做測試 main.go
package main
import (
"rpc_demo/internal/client"
"rpc_demo/internal/server"
"time"
)
func main() {
s := server.NewServer(9999)
s.Start()
time.Sleep(5 * time.Second)
c := client.NewClient(9999)
c.Send("this is a test\n")
time.Sleep(5 * time.Second)
}
執行一次main.go, go run main.go
2023/03/05 14:39:11 server [127.0.0.1:9999] start….
2023/03/05 14:39:16 [conn: 127.0.0.1:59126] get data: this is a test
可以證明第一部分的任務已經完成，可以實現兩端之間的通訊瞭
實現反射調用已註冊的方法
實現瞭雙端的通信以後，我們在internal.go裡實現兩個方法，一個是註冊，一個是調用，因為go有運行時的反射，所以我們使用反射去註冊每一個需要調用到的方法，然後提供全局唯一的函數名，讓client端可以實現指定方法的調用
internal.go
package internal
import (
"errors"
"fmt"
"reflect"
"runtime"
"strings"
)
// 全局唯一
var GlobalMethod = &Method{methods: map[string]reflect.Value{}}
type Method struct {
methods map[string]reflect.Value
}
func (m *Method) register(impl interface{}) error {
pl := reflect.ValueOf(impl)
if pl.Kind() != reflect.Func {
return errors.New("impl should be function")
}
// 獲取函數名
methodName := runtime.FuncForPC(pl.Pointer()).Name()
if len(strings.Split(methodName, ".")) < 1 {
return errors.New("invalid function name")
}
lastFuncName := strings.Split(methodName, ".")[1]
m.methods[lastFuncName] = pl
fmt.Printf("methods: %v\n", m.methods)
return nil
}
func (m *Method) call(methodName string, callParams ...interface{}) ([]interface{}, error) {
fn, ok := m.methods[methodName]
if !ok {
return nil, errors.New("impl method not found! Please Register first")
}
in := make([]reflect.Value, len(callParams))
for i := 0; i < len(callParams); i++ {
in[i] = reflect.ValueOf(callParams[i])
}
res := fn.Call(in)
out := make([]interface{}, len(res))
for i := 0; i < len(res); i++ {
out[i] = res[i].Interface()
}
return out, nil
}
func Call(methodName string, callParams ...interface{}) ([]interface{}, error) {
return GlobalMethod.call(methodName, callParams...)
}
func Register(impl interface{}) error {
return GlobalMethod.register(impl)
}
在單測裡測試一下這個註冊和調用的功能internal_test.go
package internal
import (
"testing"
)
func Sum(a, b int) int {
return a + b
}
func TestRegister(t *testing.T) {
err := Register(Sum)
if err != nil {
t.Fatalf("err: %v\n", err)
}
t.Logf("test success\n")
}
func TestCall(t *testing.T) {
TestRegister(t)
result, err := Call("Sum", 1, 2)
if err != nil {
t.Fatalf("err: %v\n", err)
}
if len(result) != 1 {
t.Fatalf("len(result) is not equal to 1\n")
}
t.Logf("Sum(1,2) = %d\n", result[0].(int))
if err := recover(); err != nil {
t.Fatalf("%v\n", err)
}
}
執行調用
/usr/local/go/bin/go test -timeout 30s -run ^TestCall$ rpc_demo/internal -v
Running tool: /usr/local/go/bin/go test -timeout 30s -run ^TestCall$ rpc_demo/internal -v
=== RUN TestCall
methods: map[Sum:<func(int, int) int Value>]
/root/go/src/juejin_demo/rpc_demo/internal/internal_test.go:15: test success
/root/go/src/juejin_demo/rpc_demo/internal/internal_test.go:27: Sum(1,2) = 3
— PASS: TestCall (0.00s)
PASS
ok rpc_demo/internal 0.002s
可以看到這個註冊和調用的過程已經實現並且達到指定方法調用的作用
設計struct完整表達一次完整的rpc調用，並且封裝json庫中的Decoder和Encoder，完成序列化和反序列化
internal.go
type RpcRequest struct {
MethodName string
Params []interface{}
}
type RpcResponses struct {
Returns []interface{}
Err error
}
transport.go考慮可以對接更多的格式，所以抽象瞭一層進行使用（demo肯定沒有更多格式瞭）
package transport
// Transport: 序列化格式的抽象層，從connection中讀取數據序列化並且反序列化到connection中
type Transport interface {
Decode(v interface{}) error
Encode(v interface{}) error
Close()
}
json_transport.go
package transport
import (
"encoding/json"
"net"
)
var _ Transport = (*JSONTransport)(nil)
type JSONTransport struct {
encoder *json.Encoder
decoder *json.Decoder
}
// NewJSONTransport: 負責讀取和寫入conn
func NewJSONTransport(conn net.Conn) *JSONTransport {
return &JSONTransport{json.NewEncoder(conn), json.NewDecoder(conn)}
}
// Decode: use json package to decode
func (t *JSONTransport) Decode(v interface{}) error {
if err := t.decoder.Decode(v); err != nil {
return err
}
return nil
}
// Encode: use json package to encode
func (t *JSONTransport) Encode(v interface{}) error {
if err := t.encoder.Encode(v); err != nil {
return err
}
return nil
}
// Close: not implement
func (dec *JSONTransport) Close() {
}
然後我們將服務端和客戶端的邏輯進行修改，改成通過上面兩個結構體進行通信，然後返回一次調用 server.go
//...
for {
conn, err := s.ls.Accept()
if err != nil {
panic(err)
}
tsp := transport.NewJSONTransport(conn)
go func() {
for {
request := &internal.RpcRequest{}
err := tsp.Decode(request)
if err != nil {
panic(err)
}
log.Printf("[server] get request: %v\n", request)
result, err := internal.Call(request.MethodName, request.Params...)
log.Printf("[server] invoke method: %v\n", result)
if err != nil {
response := &internal.RpcResponses{Returns: nil, Err: err}
tsp.Encode(response)
continue
}
response := &internal.RpcResponses{Returns: result, Err: err}
if err := tsp.Encode(response); err != nil {
log.Printf("[server] encode response err: %v\n", err)
continue
}
}
}()
}
//...
client.go
// ...
// Call: remote invoke
func (c *Client) Call(methodName string, params ...interface{}) (res *internal.RpcResponses) {
request := internal.RpcRequest{MethodName: methodName, Params: params}
log.Printf("[client] create request to invoke server: %v\n", request)
err := c.tsp.Encode(request)
if err != nil {
panic(err)
}
res = &internal.RpcResponses{}
if err := c.tsp.Decode(res); err != nil {
panic(err)
}
log.Printf("[client] get response from server: %v\n", res)
return res
}
// ...
main.go
package main
import (
"log"
"rpc_demo/internal"
"rpc_demo/internal/client"
"rpc_demo/internal/server"
"strings"
"time"
)
// Rpc方法的一個簡易實現
func Join(a ...string) string {
res := &strings.Builder{}
for i := 0; i < len(a); i++ {
res.WriteString(a[i])
}
return res.String()
}
func main() {
internal.Register(Join)
s := server.NewServer(9999)
s.Start()
time.Sleep(5 * time.Second)
c := client.NewClient(9999)
res := c.Call("Join", "aaaaa", "bbbbb", "ccccccccc", "end")
if res.Err != nil {
log.Printf("[main] get an error from server: %v\n", res.Err)
return
}
log.Printf("[main] get a response from server: %v\n", res.Returns[0].(string))
time.Sleep(5 * time.Second)
}
接下來我們運行一下main
[root@hecs-74066 rpc_demo]# go run main.go
2023/03/05 14:39:11 server [127.0.0.1:9999] start….
2023/03/05 14:39:16 [conn: 127.0.0.1:59126] get data: this is a test[root@hecs-74066 rpc_demo]# go run main.go
2023/03/05 21:53:41 server [127.0.0.1:9999] start….
2023/03/05 21:53:46 [client] create request to invoke server: {Join [aaaaa bbbbb ccccccccc end]}
2023/03/05 21:53:46 [server] get request: &{Join [aaaaa bbbbb ccccccccc end]}
2023/03/05 21:53:46 [server] invoke method: [aaaaabbbbbcccccccccend]
2023/03/05 21:53:46 [client] get response from server: &{[aaaaabbbbbcccccccccend] <nil>}
2023/03/05 21:53:46 [main] get a response from server: aaaaabbbbbcccccccccend
總結(自我pua)
這樣我們就實現瞭一個簡單的rpc框架瞭，符合最簡單的架構圖，從client->序列化請求->transport -> 反序列化 ->server然後從server->序列化請求->transport->反序列化請求->client。當然從可用性的角度來說是差遠瞭，沒有實現stub代碼，也沒有idl的實現，導致所有的註冊方法都是硬編碼，可用性不高，而且沒有集成服務發現(可以參考我的另一篇文章去集成)和熔斷等功能，也沒用中間件(也是我的另一篇文章)和超時等豐富的功能在裡面，並且最近看瞭不少rpc框架的源碼，感覺這個demo的設計也差遠瞭。不過因為時間問題和代碼的復雜性問題（單純懶），起碼算是實現瞭一個簡單的rpc框架。
推薦一些比較好的框架實現
- kitex
- dubbo
- grpc
- thrift
到此這篇關於Golang實現簡易的rpc調用的文章就介紹到這瞭,更多相關Golang rpc調用內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet！
推薦閱讀：
- golang RPC包原理和使用詳細介紹
- Go語言七篇入門教程六網絡編程
- Golang簡單實現http的server端和client端
- go語言中http超時引發的事故解決
- 一文詳解Golang中consul的基本使用