Golang實現(xiàn)簡易的rpc調(diào)用
RPC(Remote Procedure Call Protocol)遠程過程調(diào)用協(xié)議。 一個通俗的描述是：客戶端在不知道調(diào)用細節(jié)的情況下，調(diào)用存在于遠程計算機上的某個對象，就像調(diào)用本地應用程序中的對象一樣。 比較正式的描述是：一種通過網(wǎng)絡從遠程計算機程序上請求服務，而不需要了解底層網(wǎng)絡技術(shù)的協(xié)議 從使用的方面來說，服務端和客戶端通過TCP/UDP/HTTP等通訊協(xié)議通訊，在通訊的時候客戶端指定好服務端的方法、參數(shù)等信息通過序列化傳送到服務端，服務端可以通過已有的元信息找到需要調(diào)用的方法，然后完成一次調(diào)用后序列化返回給客戶端(rpc更多的是指服務與服務之間的通信，可以使用效率更高的協(xié)議和序列化格式去進行，并且可以進行有效的負載均衡和熔斷超時等，因此跟前后端之間的web的交互概念上是有點不一樣的) 用一張簡單的圖來表示
開始
本文只實現(xiàn)一個rpc框架基本的功能，不對性能做保證，因此盡量使用go原生自帶的net/json庫等進行操作，對使用方面不做stub（偷懶，只使用簡單的json格式指定需要調(diào)用的方法），用最簡單的方式實現(xiàn)一個簡易rpc框架，也不保證超時調(diào)用和服務發(fā)現(xiàn)等集成的邏輯，服務發(fā)現(xiàn)可以參考下文 本文代碼地址(https://github.com/wuhuZhao/rpc_demo)
實現(xiàn)兩點之間的通訊(transport)
本段先實現(xiàn)兩端之間的通訊，只確保兩個端之間能互相通訊即可 server.go
package server
import (
"fmt"
"log"
"net"
)
// Server: transport底層實現(xiàn)，通過Server去接受客戶端的字節(jié)流
type Server struct {
ls net.Listener
port int
}
// NewServer: 根據(jù)端口創(chuàng)建一個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: 啟動服務端的端口監(jiān)聽，采取一個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: 關閉服務監(jiān)聽
func (s *Server) Close() error {
return s.ls.Close()
}
// Close: 關閉服務監(jiān)聽
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)
}執(zhí)行一次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
可以證明第一部分的任務已經(jīng)完成，可以實現(xiàn)兩端之間的通訊了
實現(xiàn)反射調(diào)用已注冊的方法
實現(xiàn)了雙端的通信以后，我們在internal.go里實現(xiàn)兩個方法，一個是注冊，一個是調(diào)用，因為go有運行時的反射，所以我們使用反射去注冊每一個需要調(diào)用到的方法，然后提供全局唯一的函數(shù)名，讓client端可以實現(xiàn)指定方法的調(diào)用
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")
}
// 獲取函數(shù)名
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)
}
在單測里測試一下這個注冊和調(diào)用的功能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)
}
}執(zhí)行調(diào)用
/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
可以看到這個注冊和調(diào)用的過程已經(jīng)實現(xiàn)并且達到指定方法調(diào)用的作用
設計struct完整表達一次完整的rpc調(diào)用，并且封裝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中讀取數(shù)據(jù)序列化并且反序列化到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() {
}
然后我們將服務端和客戶端的邏輯進行修改，改成通過上面兩個結(jié)構(gòu)體進行通信，然后返回一次調(diào)用 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方法的一個簡易實現(xiàn)
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
總結(jié)(自我pua)
這樣我們就實現(xiàn)了一個簡單的rpc框架了，符合最簡單的架構(gòu)圖，從client->序列化請求->transport -> 反序列化 ->server然后從server->序列化請求->transport->反序列化請求->client。當然從可用性的角度來說是差遠了，沒有實現(xiàn)stub代碼，也沒有idl的實現(xiàn)，導致所有的注冊方法都是硬編碼，可用性不高，而且沒有集成服務發(fā)現(xiàn)(可以參考我的另一篇文章去集成)和熔斷等功能，也沒用中間件(也是我的另一篇文章)和超時等豐富的功能在里面，并且最近看了不少rpc框架的源碼，感覺這個demo的設計也差遠了。不過因為時間問題和代碼的復雜性問題（單純懶），起碼算是實現(xiàn)了一個簡單的rpc框架。
推薦一些比較好的框架實現(xiàn)
- kitex
- dubbo
- grpc
- thrift
到此這篇關于Golang實現(xiàn)簡易的rpc調(diào)用的文章就介紹到這了,更多相關Golang rpc調(diào)用內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
相關文章
Go中Map是一個KV對集合,下面這篇文章主要給大家介紹了關于Golang中map探究的相關資料,文中通過實例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下2022-12-12
這篇文章主要介紹了Golang Map簡介以及底層原理的相關資料,Go語言提供的map是一種鍵值對存儲結(jié)構(gòu),支持基本操作如len、delete等,map是非線程安全的,可用sync.Mutex確保并發(fā)安全,為高效查找和插入,需要的朋友可以參考下2024-10-10
這篇文章主要為大家介紹了Go語言基礎語法和基本數(shù)據(jù)類型知識鞏固,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進步,早日升職加薪2023-11-11
在Go語言中,switch語句除了常見的用法外,還有一種不常用但有趣的寫法,這種寫法中,switch后面不跟任何表達式,而每個case后面跟的是返回bool類型的函數(shù)調(diào)用表達式,這實際上是一個等價于switch true的用法,通過從上到下逐一比較case后的表達式是否為true來決定執(zhí)行哪個分支2024-10-10
在Go的日常開發(fā)中,Context上下文對象無處不在,這篇文章小編就來帶大家深入了解一下上下文context的理解與使用,文中的示例代碼講解詳細,需要的可以參考下2023-10-10
這篇文章主要為大家詳細介紹了Go語言對文件進行讀寫操作的方法，文中的示例代碼講解詳細，對我們學習Go語言有一定的幫助，需要的可以參考一下2022-05-05
Go語言內(nèi)建函數(shù)cap的實現(xiàn)示例
cap 是一個常用的內(nèi)建函數(shù),它用于獲取某些數(shù)據(jù)結(jié)構(gòu)的容量,本文主要介紹了Go語言內(nèi)建函數(shù)cap的實現(xiàn)示例,具有一定的參考價值,感興趣的可以了解一下2024-08-08
Go語言利用aicli實現(xiàn)輕松調(diào)用DeepSeek和ChatGPT
這篇文章主要為大家介紹了一款用Go語言編寫的AI助手客戶端庫——aicli,該庫不僅支持ChatGPT,還集成了DeepSeek,感興趣的小伙伴可以了解一下2025-03-03
這篇文章主要介紹了golang官方嵌入文件到可執(zhí)行程序,本文通過示例代碼給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下2021-02-02
這篇文章主要給大家介紹了關于Golang利用自定義模板發(fā)送郵件的方法，文中通過示例代碼將實現(xiàn)的方法介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧。2017-10-10