商贸网站建设,莱芜都市网房产,邯郸做网站询安联网络,南京网站建设推南京网站建设设计文章目录 前言网关服务RPC 协议看门狗服务代理服务客户端逻辑梳理 前言
上两章学习了如何搭建一个项目#xff0c;简单实现了几个基础模块。本章节会实现基本的客户端与服务端的通信#xff0c;包括网关#xff08;gate#xff09;、看门狗#xff08;watchdog#xff0… 文章目录 前言网关服务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({okfalse}))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 wslocal port assert(conf.port)local address conf.address or 0.0.0.0local 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 连接 gategate 通知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 cjsonlocal 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 dont 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 md5local _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, dont 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 configthread 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 连接同上述。
