【Skynet 入门实战练习】实现网关服务 | 用户代理 | RPC 协议

文章目录

前言
- 网关服务
- RPC 协议
- 看门狗服务
- 代理服务
- 客户端
- 逻辑梳理

前言

上两章学习了如何搭建一个项目，简单实现了几个基础模块。本章节会实现基本的客户端与服务端的通信，包括网关（gate）、看门狗（watchdog）、代理（agent）三个重要的服务，以及客户端的实现等。

网关服务

参考：websocket-gate 实现网关服务

一般客户端连接服务器选用长链接模式，skynet 支持 TCP 和 websocket，我们采用 websocket 的连接方式。

网关负责客户端的网络连接，通过 websocket 和客户端交换数据。我们可以通过普通服务创建方式来创建一个 gate 服务，但这个服务启动后，并不是马上开始工作，需要发一个 lua 消息 open，告诉 gate 监听的端口、最大连接数、延时等信息。

网关服务 service/ws_gate.lua 需要的基本接口：

local CMD = {} -- gate 服务接口
local handler = {} -- websocket 操作接口function handler.connect(fd)
end function handler.handshake(fd, header, url) 
end function handler.message(fd, msg)
end function handler.ping(fd)
endfunction handler.pong(fd)
endfunction handler.close(fd, code, reason)
endfunction handler.error(fd)
endfunction handler.warning(fd, size)
endfunction CMD.open(source, conf) 
end skynet.register_protocol {name = "client",id = skynet.PTYPE_CLIENT,
}skynet.start(function()skynet.dispatch("lua", function(session, source, cmd, ...)local f = CMD[cmd]if not f then skynet.ret(skynet.pack({ok=false}))return end if session == 0 then f(source, ...)else skynet.ret(skynet.pack(f(source, ...)))end end)skynet.register(".ws_gate")
end)

CMD 是一个服务的 lua 消息回调函数表，gate 服务会注册 (dispatch) 相关的 lua 消息，其他服务与 gate 通信，那么就会去到 CMD 中查找相关的处理函数，并根据调用方式 skynet.call 和 skynet.send 做相关的数据返回。例如 if session == 0 then 即判断是 skynet.send 调用就只需要执行即可无需返回。而是 skynet.call 调用方式，通过 skynet.ret(skynet.pack()) 进行消息打包返回给调用方。

来看具体的 CMD.open 实现：

-- call by ws_watchdog(start)
function CMD.open(source, conf) WATCHDOG = conf.watchdog or source MAXCLIENT = conf.maxclient or 1024nodelay = conf.nodelaylocal protocol = conf.protocol or "ws"local port = assert(conf.port)local address = conf.address or "0.0.0.0"local fd = socket.listen(address, port)logger.info(SERVICE_NAME, string.format("Listen websocket port: %s protocol: %s", port, protocol))socket.start(fd, function(fd, addr) logger.info(SERVICE_NAME, string.format("accept client socket_fd: %s addr: %s", fd, addr))websocket.accept(fd, handler, protocol, addr)end)
end

该方法是由 ws_watchdog 服务调用，方法中我们会获取到 watchdog 的地址，最大客户端连接数，TCP 是否延迟，通信协议 protocol，以及网关需要监听的地址 address 端口 port。

-- main.lua
-- 通知 ws_watchdog 启动服务
skynet.call(ws_watchdog, "lua", "start", {port = watchdog_port, maxclient = max_online_client,nodelay = true, protocol = ws_protocol,
})-- ws_watchdog.lua
function CMD.start(conf)-- 开启 gate 服务skynet.call(GATE, "lua", "open", conf)
end

对于服务器，通常我们需要监听一个端口，并转发某个接入连接的处理权。那么可以用如下 API ：

socket.listen(address, port) 监听一个端口，返回一个 id ，供 start 使用。
socket.start(id , accept) accept 是一个函数。每当一个监听的 id 对应的 socket 上有连接接入的时候，都会调用 accept 函数。这个函数会得到接入连接的 id 以及 ip 地址。你可以做后续操作。

socket 的 id 对于整个 skynet 节点都是公开的。也就是说，你可以把 id 这个数字通过消息发送给其它服务，其他服务也可以去操作它。任何一个服务只有在调用 socket.start(id) 之后，才可以收到这个 socket 上的数据。

handler 是一个 websocket 协议的接口表，需要有 connect / handshake / message / ping / pong / close / error这些接口方法实现。使用 websocket.accept 监听端口时，需要传入这个表，以供对上行的 socket 消息进行分别处理。该方法的源码地址：websocket.accept，每来一个连接执行一次 accept，内部使用 socket.start 获取客户端上行数据，并且执行 xpcall(resolve_accept, ...) 调用。在函数 resolve_accept 中，会先执行 connect、handshake 这两个注册在 handler 表中的方法，然后循环去读取上行的数据，对应执行其他的方法。

在这里插入图片描述

如上图，一个客户端连入，网关会监听到，并执行 handler.connect、handler.handshake 这两个方法。

来看一下 handler.connect：

function handler.connect(fd)logger.debug(SERVICE_NAME, "ws connect from: ", tostring(fd))if client_number >= MAXCLIENT then socketdriver.close(fd)return end if nodelay then socketdriver.nodelay(fd)end client_number = client_number + 1local addr = websocket.addrinfo(fd)local c = {fd = fd,ip = addr, }connection[fd] = cskynet.send(WATCHDOG, "lua", "socket", "open", fd, addr)
end

网关会控制当前连入的客户端数量，超过就不再授入连接，并且会对每个连接都设置 nodelay 属性，确保数据以小包的形式实时发送，在服务端无需额外对上行的数据进行 TCP 的拆包操作。详细的 nodelay 知识补充参考：浅谈tcp_nodelay的作用

网关服务还应该维护着客户端的连接，connection 表通过 fd 映射每一个客户端连接，同时在客户端下线，也应该清理对应的连接。

连接处理函数的最后一行还执行了一次向 ws_watchdog 服务发送的 socket消息，通知看门狗当前这个新连接的连入做相应处理。因为网关服务只是做一个通信层面的，负责客户端上行数据的转发，不做太多的逻辑处理。而客户端上行数据的逻辑处理主要就交给代理服务 ws_agent。本项目没有单独开一个登陆注册的服务，对这方面没有太复杂的需求，所以简单的登录注册逻辑就交给看门狗服务了。

客户端通信的逻辑：

client -> websocket.connect -> gate(handler.connect) -> watchdog(SOCKET.data) -> agent(login)

客户端通过 websocket.connect 连接 gate，gate 通知（open） watchdog 为该连接设置定时器 timer，在时限之内，客户端需要发送登录请求，消息被 gate 的 handler.message 收到，发现该连接没有绑定代理 agent，消息就发往 watchdog，执行登录流程，验证成功则调用 agent 的 login 方法，通知网关绑定上代理。否则消息发往 agent，执行相应逻辑处理。

下面来看 handler.message 方法：

function handler.message(fd, msg)logger.debug(SERVICE_NAME, "ws message from: ", tostring(fd), ", msg: ", msg)-- recv a package, forward itlocal c = connection[fd]local agent = c and c.agent if agent then -- msg is stringskynet.redirect(agent, c.client, "client", fd, msg)elseskynet.send(WATCHDOG, "lua", "socket", "data", fd, msg)end 
end

客户端通过 websocket 发送上来的数据 msg 是一个 lua 字符串，在 message 中进行消息分发。如果连接绑定了代理就以 client 消息类型转发 redirect 到代理处理，否则发送 socket 类型消息给看门狗。

skynet.redirect(addr, source, type, ...)：伪装成 source 地址，向 addr 发送一个消息。只有注册了 client 消息，才能使用 skynet.redirect 来发送 client 消息。

我们在 gate 服务中注册了 client 消息，专门用来将接收到的网络数据转发给 agent。不需要解包，也不需要打包。

skynet.register_protocol {name = "client",id = skynet.PTYPE_CLIENT,
}

网关服务完整代码：ws_gate.lua

RPC 协议

客户端和服务器交互的协议采用 JSON 格式，第三方工具：lua-cjson。

客户端发给服务端协议名前缀 c2s，服务端返回给客户端协议名前缀 s2c。协议的 ID 使用字段 pid 表示对应要处理的协议函数名。

例如协议 login 和协议 heartbeat：

客户端：

{pid = "c2s_heartbeat",
}{pid = "c2s_login",token = "token",acc = "account",sign = "checksum",
}

服务端：

{pid = "s2c_login",uid = "user id",msg = "Login success"
}

本项目通信设计的 RPC 协议，需要 pid 和其他参数字段。

接下来需要设计服务端和用户端的通信模块，服务端逻辑处理模块都放在 module 文件夹下，ws_agent/mng.lua 对应代理的服务逻辑处理模块，ws_watchdog/mng.lua 对应看门狗服务对应的逻辑处理模块。

这里我们先关注这两个模块的 RPC 逻辑：

ws_watchdog/mng.lua：

local _M = {} -- 模块接口
local RPC = {} -- RPC 协议接口local function check_sign(token, acc, sign)local checkstr = token .. acc local checksum = md5.sumhexa(checkstr)if checksum == sign then return true end return false 
end -- 登录协议处理函数
function RPC.c2s_login(req, fd)-- token 验证if not check_sign(req.token, req.acc, req.sign) then _M.close_fd(fd)return end -- 登录成功，分配代理，移除超时队列local res = skynet.call(AGENT, "lua", "login", req.acc, fd)noauth_fds[fd] = nil return res 
end-- 协议根据 pid 执行对应函数
function _M.handle_proto(req, fd)local f = RPC[req.pid]local res = f(req, fd)return res 
endreturn _M

上面谈及，看门狗服务我们会用来处理用户的登录逻辑。模块中 handle_proto 函数作为 RPC 协议处理的入口，根据 req.pid 对应到服务端应该执行的客户端上行协议处理函数。简单阅读上述代码，登录判定就是用 md5 对 token .. acc 进行编码与 sign 对比，成功后会通知 ws_agent 服务，并传入用户账号 acc 和客户端 fd，然后在网关绑定客户端和代理，客户端通信就移交给了代理服务。

ws_agent/mng.lua：

local _M = {} 
local RPC = {} -- c2s_heartbeat
function RPC.c2s_heartbeat(req, fd, uid)local user = online_users[uid]if not user thenreturn end user.heartbeat = skynet.time()
end function _M.handle_proto(req, fd, uid)local f = RPC[req.pid]local res = f(req, fd, uid)return res
end return _M

同理，在代理逻辑处理模块中，用于处理除了登录协议之外的其他协议，比如这里的 heartbeat。客户端登录成功，客户端设置了每 5 秒一个心跳包，服务端收到后，就会执行 user.heartbeat = skynet.time() ，维护在线用户表中用户的 heartbeat 字段。并且服务端会定时一分钟一次检测，超时踢出。

-- ws_agent/mng.luafunction _M.check_user_online(uid)local user = online_users[uid]if user then -- 心跳超时踢出if not user.heartbeat or skynet.time() - user.heartbeat >= user_alive_keep_time then _M.close_fd(user.fd)end end 
end function _M.login(acc, fd)local uid = 1 -- 数据库加载数据local user = {fd = fd, acc = acc,}online_users[uid] = user fd2uid[fd] = uid -- 通知 gate 消息由 agent 接管，绑定客户端和代理skynet.call(GATE, "lua", "forward", fd)-- 定时检查心跳local timerid = timer.timeout_repeat(60, _M.check_user_online, uid)user.timerid = timeridlocal res = {pid = "s2c_login",uid = uid, msg = "Login success",}return res
end

至此，服务端是如何与客户端通信，如何实现用户的登录逻辑，我们就已经有了一个概念。其他服务及客户端的具体实现，且继续往下看。

看门狗服务

看门狗服务主要负责 gate 的创建，agent 的创建与退出。本项目只有一个 agent，所有客户端都会绑定这个代理。如果是一个客户端对应一个代理服务，那么看门狗就可以做一个代理池，进行代理分配，代理回收等。

在主服务 main.lua 中，启动看门狗服务：

-- 开启 ws_watchdog 服务
local ws_watchdog = skynet.newservice("ws_watchdog")-- 通知 ws_watchdog 启动服务
skynet.call(ws_watchdog, "lua", "start", {port = watchdog_port, maxclient = max_online_client,nodelay = true, protocol = ws_protocol,
})

ws_watchdog.lua：

local skynet = require "skynet"
local mng = require "ws_watchdog.mng"
local cjson = require "cjson"local CMD = {} -- 服务操作接口
local SOCKET = {} -- socket 相关操作接口
local GATE -- gate 服务地址
local AGENT -- agent 服务地址function SOCKET.data(fd, msg)local req = cjson.decode(msg)if not req.pid then return end -- 判断客户端认证是否通过if not mng.check_auth(fd) then -- 没认证，且不是登录协议，踢下线if not mng.is_no_auth(req.pid) then mng.close_fd(fd)return end end -- 登录协议 or 其他协议处理local res = mng.handle_proto(req, fd)if res then skynet.call(GATE, "lua", "response", fd, cjson.encode(res))end 
endfunction CMD.start(conf)-- 开启 gate 服务skynet.call(GATE, "lua", "open", conf)
endfunction CMD.kick(fd)-- 踢客户端下线mng.close_fd(fd)
endskynet.start(function()skynet.dispatch("lua", function(session, source, cmd, subcmd, ...)if cmd == "socket" thenlocal f = SOCKET[subcmd]f(...)-- socket api don't need returnelselocal f = assert(CMD[cmd])skynet.ret(skynet.pack(f(subcmd, ...)))endend)GATE = skynet.newservice("ws_gate")AGENT = skynet.newservice("ws_agent")mng.init(GATE, AGENT)skynet.call(AGENT, "lua", "init", GATE, skynet.self())
end)

看门狗服务启动skynet.newservice("ws_watchdog")，会依次启动网关和代理服务。这里主要来看一下处理连接登录逻辑的具体步骤，先判断客户端认证是否通过（check_auth），没通过就需要限制当前看门狗服务能处理的逻辑，只能处理登录逻辑（is_no_auth），不是登录协议就不受理。

ws_watchdog/mng.lua

local skynet = require "skynet"
local timer = require "timer"
local md5 = require "md5"local _M = {} -- 模块接口
local GATE -- gate 服务地址
local AGENT -- agent 服务地址local noauth_fds = {} -- 未通过认证的服务端
local TIMEOUT_AUTH = tonumber(skynet.getenv("ws_watchdog_timeout_auth")) or 10-- 标记哪些协议不用登录就能访问
local no_auth_proto_list = {c2s_login = true, 
}function _M.is_no_auth(pid)return no_auth_proto_list[pid]
end -- 超时检测，踢掉没通过认证的客户端
local function timeout_auth(fd)local time = noauth_fds[fd]if not time then return end if skynet.time() - time < TIMEOUT_AUTH then return end _M.close_fd(fd)
endfunction _M.init(gate, agent)GATE = gate AGENT = agent
end function _M.open_fd(fd)noauth_fds[fd] = skynet.time()timer.timeout(TIMEOUT_AUTH + 1, timeout_auth, fd)
end function _M.close_fd(fd)skynet.send(GATE, "lua", "kick", fd)skynet.send(AGENT, "lua", "disconnect", fd)noauth_fds[fd] = nil 
endfunction _M.check_auth(fd)if noauth_fds[fd] then return falseend return true 
end return _M

在 ws_watchdog/mng 模块中，维护了 noauth_fds 表，还未通过认证的客户端。设定了连接登录的超时时间 TIMEOUT_AUTH ，可在配置文件中修改。

完整代码：ws_watchdog.lua、ws_watchdog/mng.lua

代理服务

代理服务主要负责接受 gate 转发的请求，处理业务，然后直接把应答响应给 gate 发到客户端。

ws_agent.lua：

skynet.register_protocol {name = "client",id = skynet.PTYPE_CLIENT,unpack = skynet.tostring, dispatch = function(fd, address, msg)skynet.ignoreret()  -- session is fd, don't call skynet.ret-- 解析消息，pid：协议idlocal req = cjson.decode(msg)if not req.pid then return end -- 登录成功会绑定 fd: uidlocal uid = mng.get_uid(fd)if not uid then mng.close_fd(fd) -- close_fd 应该实现给watchdog发消息关闭清空资源吧，end     local res = mng.handle_proto(req, fd, uid) if res then skynet.call(GATE, "lua", "response", fd, cjson.encode(res))end end
}skynet.start(function()skynet.dispatch("lua", function(_, _, command, ...)-- skynet.trace()local f = CMD[command]skynet.ret(skynet.pack(f(...)))end)skynet.register(".ws_agent")
end)

代理服务注册 skynet.PTYPE_CLIENT 类型消息，还记得网关也注册了吗？网关收到客户端网络消息，包装成 client 消息，直接发给代理，代理通过 unpack = skynet.tostring 解包消息，通过 dispatch 分发客户端上行的网络消息。

完整代码：ws_agent.lua、ws_agent/mng.lua

客户端

客户端我们也实现为一个 skynet 中的 lua 服务，需要指定配置文件启动 ./skynet/skynet etc/config.client。

etc/config.client：

include "config"thread = 2server_host = "127.0.0.1" bootstrap = "snlua bootstrap" 
start = "test/client"
logtag = "client"

主服务指定为 test/client，日志文件标识为 logtag = "client"。

下面来看 test/client.lua：

local ws_id -- websocket 连接 ID
local cmds = {} -- 命令模块，ws: ws.lua、gm: gm.lua-- 搜索加载命令模块
local function fetch_cmds()local t = utils_file.scandir("test/cmds")for _, v in pairs(t) do local cmd = utils_string.split(v, ".")[1] -- ws、gmlocal cmd_mod = "test.cmds." .. cmd cmds[cmd] = require(cmd_mod)end 
end skynet.start(function()dns.server() -- 初始化 dnsfetch_cmds()skynet.fork(websocket_main_loop)skynet.fork(console_main_loop)
end)

启动客户端，初始化设置 dns 服务器。cmds 作为命令模块，相关的 RPC 通信协议命令写在 ws.lua 模块中，GM 指令模块写在 gm.lua 模块中。fetch_cmds 会利用 file、string 工具搜索加载命令模块，存储在 cmds 中。

参考官方 wiki skynet.dns：

在 skynet 的底层，当使用域名而不是 ip 时，由于调用了系统 api getaddrinfo ，有可能阻塞住整个 socket 线程（不仅仅是阻塞当前服务，而是阻塞整个 skynet 节点的网络消息处理）。虽然大多数情况下，我们并不需要向外主动建立连接。但如果你使用了类似 httpc 这样的模块以域名形式向外请求时，一定要关注这个问题。

skynet 暂时不打算在底层实现非阻塞的域名查询。但提供了一个上层模块来辅助你解决 dns 查询时造成的线程阻塞问题。

local dns = require "skynet.dns" 加载这个模块

dns.server(ip, port) ： port 的默认值为 53 。如果不填写 ip 的话，将从 /etc/resolv.conf 中找到合适的 ip 。
dns.resolve(name, ipv6) : 查询 name 对应的 ip ，如果 ipv6 为 true 则查询 ipv6 地址，默认为 false 。如果查询失败将抛出异常，成功则返回 ip ，以及一张包含有所有 ip 的 table 。
dns.flush() : 默认情况下，模块会根据 TTL 值 cache 查询结果。在查询超时的情况下，也可能返回之前的结果。dns.flush() 可以用来清空 cache 。注意：cache 保存在调用者的服务中，并非针对整个 skynet 进程。所以，推荐写一个独立的 dns 查询服务统一处理 dns 查询。

上述代码还启用了两个协程，分别执行 websocket_main_loop 和 console_main_loop。

-- 网络循环
local function handle_resp(ws_id, res)for _, cmd_mod in pairs(cmds) do if cmd_mod.handle_res then cmd_mod.handle_res(ws_id, res)end end 
end local function websocket_main_loop()-- 连接服务器local ws_protocol = skynet.getenv("ws_watchdog_protocol")local ws_port = skynet.getenv("ws_watchdog_port")local server_host = skynet.getenv("server_host")local url = string.format("%s://%s:%s/client", ws_protocol, server_host, ws_port)ws_id = websocket.connect(url)while true do local res, close_reason = websocket.read(ws_id)local ok, err = xpcall(handle_resp, debug.traceback, ws_id, cjson.decode(res))websocket.ping(ws_id)end 
end

网络循环中，连接服务器成功后会对服务器下行的数据进行读取处理，调度到模块中的 handle_res 分别找到不同的模块处理相应的网络协议。

-- 执行注册的命令
local function run_command(cmd, ...)local cmd_mod = cmds[cmd]if cmd_mod then cmd_mod.run_command(ws_id, ...) end 
end -- 命令交互
-- ws login acc
local function console_main_loop()local stdin = socket.stdin()while true do local cmdline = socket.readline(stdin, "\n")if cmdline ~= "" then local split = split_cmdline(cmdline)local cmd = split[1]local ok, err = xpcall(run_command, debug.traceback, cmd, select(2, table.unpack(split)))end end 
end

命令的读取循环中，通过输入 ws login user_count 等指令，找到 ws 模块下的 login 回调方法，对应 c2s_login RPC 协议发送数据 websocket.write 给服务端。

完整代码：test/client.lua、test/cmds/ws.lua

逻辑梳理

通过上面几节描述，我们已经实现了自定义 RPC 协议，完成客户端与服务端的通信。

最后，我们再来梳理一下完整的服务端启动逻辑，客户端连入、退出等逻辑。

服务端启动：主服务 main.lua 启动看门狗服务，看门狗启动网关和代理服务，并且主服务发了一个 start 消息给看门狗，看门狗随即发了一个 open 消息给网关，网关则打开了监听端口，等待连接的接入。同时代理也注册好了 client 消息，等待客户端的数据交互。

客户端连接登录：客户端 websocket.connect(url) 成功连入服务器，网关 websocket.accept(fd, handler, protocol, addr) 感知到是一个连接消息，handler.connect(fd) 随即执行，设置好该连接相关属性后通知看门狗对连接进行处理，即设置一个连接超时定时器。然后，客户端输入登录命令，数据 websocket.write(ws_id, cjson.encode(req)) 上行到服务器，handler.message(fd, msg) 感知到来了一条客户端消息，对该连接判断是否绑定代理，未绑定代理则通知看门狗处理一下这条 socket 消息，仅限登录协议 c2s_login 消息能处理。随后，看门狗验证通过，通知代理授理登录消息。代理服务通知网关将这条连接绑定上代理。之后客户端上行的数据，网关通过 skynet.redirect(agent, c.client, "client", fd, msg) 直接转发给代理处理。

客户端退出：

客户端主动退出 CTRL + C：触发网关的 handler.error(fd)，调用 close_fd 清空网关维护的该连接资源，并且通知看门狗这条 socket 的 error 消息，看门狗调用 close_fd 清空看门狗维护的该连接资源，并由看门狗通知代理 disconnect 该连接和网关 kick 该连接。代理清空维护的该客户端的资源，网关则执行 websocket.close(fd) 实际的关闭了连接，但由于客户端主动断开，未经 websocket 协议正常关闭流程，不触发 handler.close。
客户端被动退出：连接超时，看门狗 timeout_auth 进行超时检测，调用 close_fd 同上述。但是在网关 websocket.close(fd) 执行后，服务端发送一个 websocket 关闭帧通知客户端并等待回应，所以会触发 handler.close(fd, code, reason)，然后执行网关 close_fd ，并给看门狗发了一个 socket 的 close 消息，看门狗再一次 close_fd。编码保证了多次释放资源不会造成问题。
客户端被动退出：心跳超时，代理 check_user_online 进行心跳检测，调用 close_fd，会执行 disconnect 逻辑释放代理维护的资源，然后通知网关 kick 连接，同上述。