Skynet服务端框架搭建4-分布式登录流程

迷南。 2022-08-28 00:42 367阅读 0赞

Skynet服务端框架搭建4-分布式登录流程

本篇在上一节的基础上，处理玩家登陆逻辑，以便熟悉整套框架

连通网关和玩家

首先修改上节的gateway模块，要把gateway和agent关联起来，定义conns和players两个表，以及conn和gateplayer两个类

service/gateway/init.lua

  1. conns = {
  2.     } --[socket_id] = conn
  3. players = {
  4.     } --[playerid] = gateplayer
  6. --连接类
  7. function conn()
  8.     local m = {
  10.         fd = nil,
  11.         playerid = nil,
  12.     }
  13.     return m
  14. end
  16. --玩家类
  17. function gateplayer()
  18.     local m = {
  20.         playerid = nil,
  21.         agent = nil,
  22.         conn = nil,
  23.     }
  24.     return m
  25. end

接受客户端的连接

首先要开启Socket监听，前面配置了runconfig文件，这里也读取该配置，找到gateway的监听端口port，然后使用skynet.socket模块的listen和start方法开启监听。当有客户端连接时，start方法的回调函数connect会被调用

service/gateway/init.lua

  1. local socket = require "skynet.socket"
  2. local runconfig = require "runconfig"
  4. function s.init()
  5.     local node = skynet.getenv("node")
  6.     local nodecfg = runconfig[node]
  7.     local port = nodecfg.gateway[s.id].port
  9.     local listenfd = socket.listen("0.0.0.0", port)
  10.     skynet.error("Listen socket :", "0.0.0.0", port)
  11.     socket.start(listenfd , connect)
  12. end
  14. --有新连接时
  15. local connect = function(fd, addr)
  16.     print("connect from " .. addr .. " " .. fd)
  17.     local c = conn()
  18.     conns[fd] = c
  19.     c.fd = fd
  20.     skynet.fork(recv_loop, fd)
  21. end

recv_loop负责接收客户端消息，参数fd由skynet.fork传入，表示客户端的标识。

service/gateway/init.lua

  1. --每一条连接接收数据处理
  2. --协议格式 cmd,arg1,arg2,...#
  3. local recv_loop = function(fd)
  4.     socket.start(fd)
  5.     skynet.error("socket connected " ..fd)
  6.     local readbuff = ""
  7.     while true do
  8.         local recvstr = socket.read(fd)
  9.         if recvstr then
  10.             readbuff = readbuff..recvstr
  11.             readbuff = process_buff(fd, readbuff)
  12.         else
  13.             skynet.error("socket close " ..fd)
  14.             disconnect(fd)
  15.             socket.close(fd)
  16.             return
  17.         end
  18.     end
  19. end

处理客户端协议

process_buff中实现消息切分的工作，如果readbuff的内容是”login,101,134\r\nwork\r\nwo”，则会处理成”login,101,123”和”work”两条消息返回给recv_loop

service/gateway/init.lua

  1. local process_buff = function(fd, readbuff)
  2.     while true do
  3.         local msgstr, rest = string.match( readbuff, "(.-)\r\n(.*)")
  4.         if msgstr then
  5.             readbuff = rest
  6.             process_msg(fd, msgstr)
  7.         else
  8.             return readbuff
  9.         end
  10.     end
  11. end

fd为客户端连接标识，readbuff为接收数据的缓冲区，这里的string.match就是正则规则，”(.-)\r\n(.*)”意思是把取出的第一条消息和剩余的部分

编码和解码

其实这个很像之前我写的Erlang游戏服务端mini项目实战4_消息协议层搭建1(封装协议号)，其实就是客户端和服务端之间的封包和解包

service/gateway/init.lua

  1. local str_pack = function(cmd, msg)
  2.     return table.concat( msg, ",").."\r\n"
  3. end
  5. local str_unpack = function(msgstr)
  6.     local msg = {
  7.     }
  9.     while true do
  10.         local arg, rest = string.match( msgstr, "(.-),(.*)")
  11.         if arg then
  12.             msgstr = rest
  13.             table.insert(msg, arg)
  14.         else
  15.             table.insert(msg, msgstr)
  16.             break
  17.         end
  18.     end
  20.     return msg[1], msg
  21. end

消息分发

这里分为两种情况，如果玩家没登录则进入分配agent让客户端进入login协议，如果已经登录了链接到对应的agent服务

service/gateway/init.lua

  1. local process_msg = function(fd, msgstr)
  2.     local cmd, msg = str_unpack(msgstr)
  3.     skynet.error("recv "..fd.." ["..cmd.."] {"..table.concat( msg, ",").."}")
  5.     local conn = conns[fd]
  6.     local playerid = conn.playerid
  7.     --尚未完成登录流程
  8.     if not playerid then
  9.         local node = skynet.getenv("node")
  10.         local nodecfg = runconfig[node]
  11.         local loginid = math.random(1, #nodecfg.login)
  12.         local login = "login"..loginid
  13.         skynet.send(login, "lua", "client", fd, cmd, msg)
  14.     --完成登录流程
  15.     else
  16.         local gplayer = players[playerid]
  17.         local agent = gplayer.agent
  18.         skynet.send(agent, "lua", "client", cmd, msg)
  19.     end
  20. end

发消息接口

service/gateway/init.lua

  1. s.resp.send_by_fd = function(source, fd, msg)
  2.     if not conns[fd] then
  3.         return
  4.     end
  6.     local buff = str_pack(msg[1], msg)
  7.     skynet.error("send "..fd.." ["..msg[1].."] {"..table.concat( msg, ",").."}")
  8.     socket.write(fd, buff)
  9. end
  11. s.resp.send = function(source, playerid, msg)
  12.     local gplayer = players[playerid]
  13.     if gplayer == nil then
  14.         return
  15.     end
  16.     local c = gplayer.conn
  17.     if c == nil then
  18.         return
  19.     end
  21.     s.resp.send_by_fd(nil, c.fd, msg)
  22. end

确认登录&登出接口

service/gateway/init.lua

  1. s.resp.sure_agent = function(source, fd, playerid, agent)
  3.     local conn = conns[fd]
  4.     if not conn then --登陆过程中已经下线
  5.         skynet.call("agentmgr", "lua", "reqkick", playerid, "未完成登陆即下线")
  6.         return false
  7.     end
  9.     conn.playerid = playerid
  11.     local gplayer = gateplayer()
  12.     gplayer.playerid = playerid
  13.     gplayer.agent = agent
  14.     gplayer.conn = conn
  15.     players[playerid] = gplayer
  17.     return true
  18. end
  20. local disconnect = function(fd)
  21.     local c = conns[fd]
  22.     if not c then
  23.         return
  24.     end
  26.     local playerid = c.playerid
  27.     --还没完成登录
  28.     if not playerid then
  29.         return
  30.     --已在游戏中
  31.     else
  32.         players[playerid] = nil
  33.         local reason = "断线"
  34.         skynet.call("agentmgr", "lua", "reqkick", playerid, reason)
  35.     end
  36. end

最后还有踢出接口

  1. s.resp.kick = function(source, playerid)
  2.     local gplayer = players[playerid]
  3.     if not gplayer then
  4.         return
  5.     end
  7.     local c = gplayer.conn
  8.     players[playerid] = nil
  10.     if not c then
  11.         return
  12.     end
  13.     conns[c.fd] = nil
  15.     disconnect(c.fd)
  16.     socket.close(c.fd)
  17. end

接下来编写第二个服务——登录

登录协议

刚才客户端发送的登录请求是 login,123,456 三个参数分别是协议名、玩家id、密码，我们服务端只需要返回 login,0/1,登陆成功/登录失败 三个参数分别是协议名，结果参数和msg

login/init.lua

  1. local skynet = require "skynet"
  2. local s = require "service"
  4. s.client = {
  5.     }
  6. s.resp.client = function(source, fd, cmd, msg)
  7.     if s.client[cmd] then
  8.         local ret_msg = s.client[cmd]( fd, msg, source)
  9.         skynet.send(source, "lua", "send_by_fd", fd, ret_msg)
  10.     else
  11.         skynet.error("s.resp.client fail", cmd)
  12.     end
  13. end
  15. s.client.login = function(fd, msg, source)
  16.     local playerid = tonumber(msg[2])
  17.     local pw = tonumber(msg[3])
  18.     local gate = source
  19.     node = skynet.getenv("node")
  20.     --校验用户名密码
  21.     if pw ~= 123 then
  22.         return {
  23.     "login", 1, "密码错误"}
  24.     end
  25.     --发给agentmgr
  26.     local isok, agent = skynet.call("agentmgr", "lua", "reqlogin", playerid, node, gate)
  27.     if not isok then
  28.         return {
  29.     "login", 1, "请求mgr失败"}
  30.     end
  31.     --回应gate
  32.     local isok = skynet.call(gate, "lua", "sure_agent", fd, playerid, agent)
  33.     if not isok then
  34.         return {
  35.     "login", 1, "gate注册失败"}
  36.     end
  37.     skynet.error("login succ "..playerid)
  38.     return {
  39.     "login", 0, "登陆成功"}
  40. end
  42. s.start(...)

真正业务上的时候可能还得是从数据库中存取这个啦

agentmgr

再看看agentmgr模块，他是管理agent的服务，是控制玩家登陆流程的主要模块，模块内维护一个列表players，保存着所有玩家的在线状态。

agentmgr/init.lua

  1. local skynet = require "skynet"
  2. local s = require "service"
  4. --状态
  5. STATUS = {
  7.     LOGIN = 2,
  8.     GAME = 3,
  9.     LOGOUT = 4,
  10. }
  12. --玩家列表
  13. local players = {
  14.     }
  16. --玩家类
  17. function mgrplayer()
  18.     local m = {
  20.         playerid = nil,
  21.         node = nil,
  22.         agent = nil,
  23.         status = nil,
  24.         gate = nil,
  25.     }
  26.     return m
  27. end
  29. s.resp.reqlogin = function(source, playerid, node, gate)
  30.     local mplayer = players[playerid]
  31.     --登陆过程禁止顶替
  32.     if mplayer and mplayer.status == STATUS.LOGOUT then
  33.         skynet.error("reqlogin fail, at status LOGOUT " ..playerid )
  34.         return false
  35.     end
  36.     if mplayer and mplayer.status == STATUS.LOGIN then
  37.         skynet.error("reqlogin fail, at status LOGIN " ..playerid)
  38.         return false
  39.     end
  40.     --在线，顶替
  41.     if mplayer then
  42.         local pnode = mplayer.node
  43.         local pagent = mplayer.agent
  44.         local pgate = mplayer.gate
  45.         mplayer.status = STATUS.LOGOUT,
  46.         s.call(pnode, pagent, "kick")
  47.         s.send(pnode, pagent, "exit")
  48.         s.send(pnode, pgate, "send", playerid, {
  49.     "kick","顶替下线"})
  50.         s.call(pnode, pgate, "kick", playerid)
  51.     end
  52.     --上线
  53.     local player = mgrplayer()
  54.     player.playerid = playerid
  55.     player.node = node
  56.     player.gate = gate
  57.     player.agent = nil
  58.     player.status = STATUS.LOGIN
  59.     players[playerid] = player
  60.     local agent = s.call(node, "nodemgr", "newservice", "agent", "agent", playerid)
  61.     player.agent = agent
  62.     player.status = STATUS.GAME
  63.     return true, agent
  64. end
  66. s.resp.reqkick = function(source, playerid, reason)
  67.     local mplayer = players[playerid]
  68.     if not mplayer then
  69.         return false
  70.     end
  72.     if mplayer.status ~= STATUS.GAME then
  73.         return false
  74.     end
  76.     local pnode = mplayer.node
  77.     local pagent = mplayer.agent
  78.     local pgate = mplayer.gate
  79.     mplayer.status = STATUS.LOGOUT
  81.     s.call(pnode, pagent, "kick")
  82.     s.send(pnode, pagent, "exit")
  83.     s.send(pnode, pgate, "kick", playerid)
  84.     players[playerid] = nil
  86.     return true
  87. end
  89. --情况 永不下线
  92. s.start(...)

代码中包括请求登陆和请求登出的接口

nodemgr

nodemgr/init.lua

  1. local skynet = require "skynet"
  2. local s = require "service"
  4. s.resp.newservice = function(source, name, ...)
  5.     local srv = skynet.newservice(name, ...)
  6.     return srv
  7. end
  9. s.start(...)

nodemgr管理节点服务，每个节点开一个新的服务，这里只是简单调用了newservice方法

agent

最后一个模块了，

agent/init.lua

  1. local skynet = require "skynet"
  2. local s = require "service"
  4. s.client = {
  5.     }
  6. s.gate = nil
  8. require "scene"
  11. s.resp.client = function(source, cmd, msg)
  12.     s.gate = source
  13.     if s.client[cmd] then
  14.         local ret_msg = s.client[cmd]( msg, source)
  15.         if ret_msg then
  16.             skynet.send(source, "lua", "send", s.id, ret_msg)
  17.         end
  18.     else
  19.         skynet.error("s.resp.client fail", cmd)
  20.     end
  21. end
  23. s.client.work = function(msg)
  24.     s.data.coin = s.data.coin + 1
  25.     return {
  26.     "work", s.data.coin}
  27. end
  29. s.resp.kick = function(source)
  30.     s.leave_scene()
  31.     --在此处保存角色数据
  32.     skynet.sleep(200)
  33. end
  35. s.resp.exit = function(source)
  36.     skynet.exit()
  37. end
  39. s.resp.send = function(source, msg)
  40.     skynet.send(s.gate, "lua", "send", s.id, msg)
  41. end
  43. s.init = function( )
  44.     --playerid = s.id
  45.     --在此处加载角色数据
  46.     skynet.sleep(200)
  47.     s.data = {
  49.         coin = 100,
  50.         hp = 200,
  51.     }
  52. end
  54. s.start(...)

这里主要实现的是消息分发、数据加载和保存退出

测试登录流程

修改main.lua里的内容，调用刚才写的各个模块

main.lua

  1. local skynet = require "skynet"
  2. local skynet_manager = require "skynet.manager"
  3. local runconfig = require "runconfig"
  4. local cluster = require "skynet.cluster"
  8. skynet.start(function()
  9.     --初始化
  10.     local mynode = skynet.getenv("node")
  11.     local nodecfg = runconfig[mynode]
  12.     --节点管理
  13.     local nodemgr = skynet.newservice("nodemgr","nodemgr", 0)
  14.     skynet.name("nodemgr", nodemgr)
  15.     --集群
  16.     cluster.reload(runconfig.cluster)
  17.     cluster.open(mynode)
  18.     --gate
  19.     for i, v in pairs(nodecfg.gateway or {
  20.     }) do
  21.         local srv = skynet.newservice("gateway","gateway", i)
  22.         skynet.name("gateway"..i, srv)
  23.     end
  24.     --login
  25.     for i, v in pairs(nodecfg.login or {
  26.     })  do
  27.     local srv = skynet.newservice("login","login", i)
  28.         skynet.name("login"..i, srv)
  29.     end
  30.     --agentmgr
  31.     local anode = runconfig.agentmgr.node
  32.     if mynode == anode then
  33.         local srv = skynet.newservice("agentmgr", "agentmgr", 0)
  34.         skynet.name("agentmgr", srv)
  35.     else
  36.         local proxy = cluster.proxy(anode, "agentmgr")
  37.         skynet.name("agentmgr", proxy)
  38.     end
  39.     --scene (sid->sceneid)
  40.     for _, sid in pairs(runconfig.scene[mynode] or {
  41.     }) do
  42.         local srv = skynet.newservice("scene", "scene", sid)
  43.         skynet.name("scene"..sid, srv)
  44.     end
  45.     --退出自身
  46.     skynet.exit()
  47. end)

测试

GIF 2021-10-29 20-27-57

发表评论

表情：
评论列表 （有 0 条评论，367人围观）

还没有评论，来说两句吧...

相关阅读