chuck-lua中的RPC

chuck-lua內置了基於coroutine的RPC支持,全部的遠程方法調用都跟調用本地方法同樣簡單.下面先來看一個簡單的示例。

rpcserver.lua

local Task   = require("distri.uthread.task")
local Distri = require("distri.distri")
local Redis  = require("distri.redis")
local Socket = require("distri.socket")
local chuck  = require("chuck")
local RPC    = require("distri.rpc")
local Packet = chuck.packet

local config = RPC.Config(
function (data)                            --encoder
    local wpk = Packet.wpacket(512)
    wpk:WriteTab(data)
    return wpk
end,
function (packet)                          --decoder
    return packet:ReadTab()
end)

local rpcServer = RPC.Server(config)
rpcServer:RegService("hello",function ()
    return "world"
end)

local server = Socket.stream.Listen("127.0.0.1",8010,function (s,errno)
    if s then
        s:Ok(4096,Socket.stream.decoder.rpacket(4096),function (_,msg,errno)
            if msg then
                rpcServer:ProcessRPC(s,msg)
            else
                s:Close()
                s = nil
            end
        end)
    end
end)

if server then
    Distri.Signal(chuck.signal.SIGINT,Distri.Stop)
    Distri.Run()
end

rpcclient.lua

local Task   = require("distri.uthread.task")
local Distri = require("distri.distri")
local Redis  = require("distri.redis")
local Socket = require("distri.socket")
local chuck  = require("chuck")
local RPC    = require("distri.rpc")
local Packet = chuck.packet
local Sche  = require("distri.uthread.sche")

local config = RPC.Config(
function (data)                           --encoder                        
    local wpk = Packet.wpacket(512)
    wpk:WriteTab(data)
    return wpk
end,
function (packet)                         --decoder
    return packet:ReadTab()
end)

local rpcClient = RPC.Client(config)

local c = 0
if Socket.stream.Connect("127.0.0.1",8010,function (s,errno)
    if s then
        if not s:Ok(4096,Socket.stream.decoder.rpacket(4096),
                    function (_,msg,errno)
                        if msg then
                            c = c + 1
                            RPC.OnRPCResponse(config,s,msg)
                        else
                            print("close")
                            s:Close()
                            s = nil
                        end
                    end) 
        then
            return
        end
        rpcClient:Connect(s)
        for i = 1,100 do
            Task.New(function ()
                while true do
                    local err,ret = rpcClient:Call("hello")
                    if ret ~= "world" then
                        print("err")
                        break
                    end
                end
            end)
        end
    end
end) then
    Distri.Signal(chuck.signal.SIGINT,Distri.Stop)
    Distri.Run()
end

首先須要瞭解的是RPC.Config,它是RPC的配置對象:

local rpc_config = {}

function rpc_config:new(encoder,decoder,serializer,unserializer)
    if not encoder or not decoder then
        return nil
    end
    local o        = {}   
    o.__index      = rpc_config
    setmetatable(o, o)
    o.decoder      = decoder        --將數據從packet中取出
    o.encoder      = encoder        --使用特定封包格式將數據封裝到packet中      
    o.serializer   = serializer     --將lua table序列化成傳輸格式,能夠爲空
    o.unserializer = unserializer   --將傳輸格式的數據反序列化成lua table,能夠爲空
    return o
end

其中提供了4個重要的方法,這個對象將會被傳遞給RPC.Client和RPC.Server供他們將調用的參數。

在上面的示例中,我只提供了decoder和encoder.下面簡單介紹下這4個方法的做用.

• serializer: rpc調用的參數和返回值都經過luatable包裹.這個函數用於將這個被包裹的table序列化成一種公共傳輸格式,例如json.其輸出對象將會被提供給encoder.

• unserializer: 與serializer的做用正好相反,將公共傳輸格式反序列化成luatable.其輸出對象將會被提供給decoder.若是不提供這兩個函數,則直接將luatable提供給encoder/decoder.

• encoder: 將傳輸數據封裝成網絡包,供Send發送.

• decoder: 從網絡包中提取出傳輸的數據.

在上面的示例中,我使用了內置的一種packet.直接將luatable寫入到packet中.

示例中須要注意的另一點是,rpcClient:Call("hello")必須在coroutine上下文中使用,不然將會返回錯誤.而 rpcServer:ProcessRPC(s,msg)則不是必須的,但若是在RPC請求的處理函數中須要執行一些阻塞調用,例如請求別的遠程方法,或請求redis數據.則必須將rpcServer:ProcessRPC(s,msg)放在coroutine下執行.

若是在一條鏈路上只涉及到RPC請求,那麼徹底能夠對rpcclient作一個簡單的封裝,免得每次都手動執行全部的初始化步驟,例如:

local WrapRPCClient = {}

function WrapRPCClient:new(config)
  local o = {}
  o.__index = actor      
  setmetatable(o,o)
  o.rpc    = RPC.Client(config)
  return o
end

function WrapRPCClient:Connect(host,port,on_success)
    local config = self.rpc.config
    if Socket.stream.Connect("127.0.0.1",8010,function (s,errno)
        if s then
            if not s:Ok(4096,Socket.stream.decoder.rpacket(4096),
                        function (_,msg,errno)
                            if msg then
                                RPC.OnRPCResponse(config,s,msg)
                            else
                                print("close")
                                s:Close()
                                s = nil
                            end
                        end) 
            then
                return
            end
            rpcClient:Connect(s)
            on_success()
        end
    end) then
        return true
    end 
end

function WrapRPCClient:Call(func,...)
    return self.rpc:Call(func,...)
end

有了這個封裝以後,只須要調用Connect,當on_success回調以後就能夠直接使用Call調用遠程方法了.

https://github.com/sniperHW/chuck