https://www.aliyun.com/jiaocheng/647017.html
https://github.com/chenxiaofa/p
前言作為一個(gè)游戲從業(yè)者不可能不使用推方案,以前一直使用 nginx-push-stream-module這個(gè)模塊的 Forever Iframe模式來(lái)實(shí)現(xiàn)推方案。
最近決定研究下 lua-resty-websocket來(lái)實(shí)現(xiàn)一個(gè)更加高效好用推方案
不推薦使用的場(chǎng)景由于 OpenResty目前還不能做到跨 worker通信,所以想到實(shí)現(xiàn)指定推送需要中轉(zhuǎn)一次,效率上可能不如其他語(yǔ)言如 golang等
過(guò)于復(fù)雜的業(yè)務(wù)邏輯 頻繁的指定推送(單對(duì)單、組、Tag等) 廣播過(guò)多 一起工作的好基友們想要推的優(yōu)雅以下幾個(gè)基友的幫忙是不可或缺的
ngx.semaphore它可以讓你在想要發(fā)消息的地方優(yōu)雅的進(jìn)入到發(fā)送階段,也可以讓你來(lái)優(yōu)雅的控制一個(gè)鏈接超時(shí)的關(guān)閉。
ngx.shared由于目前我們無(wú)法做到跨 worker的通信,所以必須借助共享內(nèi)存來(lái)中轉(zhuǎn)不屬于當(dāng)前 worker的消息。
lua-resty-websocket由于貪圖方便還是直接使用了現(xiàn)成的庫(kù),喜歡折騰的小伙伴請(qǐng)移步 stream-lua-nginx-module
大概的思路由于不能跨 worker通信所以我給每個(gè) worker申請(qǐng)了一個(gè) shared共享內(nèi)存來(lái)保存消息。
理論上 shared的數(shù)量等于 worker的數(shù)量最佳。
然后每個(gè) worker啟動(dòng)一個(gè) timer來(lái)判斷當(dāng)前 worker的 message id和 shared中的 message id是否有變化。
這里為什么不用 shared的有序列表來(lái)做,容我先賣個(gè)關(guān)子。
當(dāng)發(fā)生變化時(shí),判斷消息的目標(biāo)是否在自己的 session hash中,如果在則發(fā)之。
開(kāi)始準(zhǔn)備工作 修改配置文件首先修改 nginx.conf配置,增加以下設(shè)置
lua_shared_dict message_1 10m;
lua_shared_dict message_2 10m;
lua_shared_dict message_n 10m;
init_worker_by_lua_file scripts/init_worker_by_lua.lua;
init_worker_by_lua
local ngx = ngx
local ngx_log = ngx.log
local ngx_ERR = ngx.ERR
local ngx_timer_at = ngx.timer.at
local require = require
local socketMgr = require("socketMgr")
local delay = 1
local loopMessage
loopMessage = function(premature)
if premature then
ngx_log(ngx_ERR, "timer was shut: ", err)
return
end
socketMgr:loopMessages()
local ok, err = ngx_timer_at(delay, loopMessage)
if not ok then
ngx_log(ngx_ERR, "failed to create the timer: ", err)
return
end
end
loopMessage()
loopMessages判斷 local message id和 shared message id是否不等。
隨后每次 local message id+ 1 從 shared拉取數(shù)據(jù),進(jìn)行消息推送邏輯。
建立連接不做過(guò)多說(shuō)明,自行查看 lua-resty-websocket的 wiki
當(dāng)連接監(jiān)聽(tīng)好之后,要進(jìn)行一系列的管理。如:
session id和 user id的雙向映射 session id和 group name的雙向映射后面再詳細(xì)說(shuō)明
生成 session id我是用 ( worker id+ 1) * 100000 + worker's local incr id來(lái)生成唯一 session id比較簡(jiǎn)陋,但是夠用。
這么做的原因是,通過(guò)對(duì) session id進(jìn)行取余可以很方便的得知 worker id,可以方便的給 shared寫(xiě)消息。
local ngx_worker_id = ngx.worker.id()local _incr_id = 0local _gen_session_id = function()_incr_id = _incr_id + 1return (ngx_worker_id + 1) * 100000 + _incr_idend 設(shè)置消息映射這個(gè)可以用于收到當(dāng)前 worker所屬的 shared message判斷是否在當(dāng)前進(jìn)程。
_messages[session_id] = {}_semaphores[session_id] = semaphore.new(0) 接收消息&;發(fā)送消息代碼和 官方例子類同不做過(guò)多說(shuō)明,只說(shuō)我改了什么。
在接收消息中管理了一個(gè)變量即 close_flag用于管理 send message輕線程的退出。
以下是一段偽代碼,含義的話請(qǐng)聯(lián)系上下文。
local session_id = sessionMgr:gen_session_id()
local send_semaphore = sessionMgr:get_semaphore(session_id)
local close_flag = false
local function _push_thread_function()
while close_flag == false do
local ok, err = send_semaphore:wait(300)
if ok then
local messages = socketMgr:getMessages(session_id)
while messages and #messages > 0 do
local message = messages[1]
table_remove(messages, 1)
--- your send message function handler
end
end
if close_flag then
socketMgr:destory(session_id)
break
end
end
end
local push_thread = ngx_thread_spawn(_push_thread_function)
while true do
local data, typ, err = wbsocket:recv_frame()
while err == "again" do
local cut_data cut_data, _, err = wbsocket:recv_frame()
data = data .. cut_data
end
if not data then
close_flag = true
send_semaphore:post(1)
break
elseif typ == 'close' then
close_flag = true
send_semaphore:post(1)
break
elseif typ == 'ping' then
local bytes, err = wbsocket:send_pong(data)
if not bytes then
close_flag = true
send_semaphore:post(1)
break
end
elseif typ == 'pong' then
elseif typ == 'text' then
-- your receive function handler
elseif typ == 'continuation' then
elseif typ == 'binary' then
end
end
ngx_thread_wait(push_thread)
wbsocket:send_close() 消息推送現(xiàn)在說(shuō)說(shuō)為什么不用 shared的有序列表來(lái)存儲(chǔ)消息,我是使用了 shared的 set方法中的 flag屬性來(lái)存放 session id。
這樣在獲得一個(gè)消息的時(shí)候,能很方便的知道消息是發(fā)給哪個(gè) session id的。
繼續(xù)一段偽代碼。
local ngx_shared = ngx.shared
local _shared_dicts = {ngx_shared.message_1,ngx_shared.message_2,ngx_shared.message_n,}
local current_shared = _shared_dicts[ngx_worker_id + 1]
local current_message_id = 1
--- 如果在當(dāng)前進(jìn)程
if _messages[session_id] then
table.insert(_messages[session_id], "message data")
_semaphores[session_id]:post(1)
--- 會(huì)進(jìn)入到,上述 _push_thread_function 方法中,進(jìn)行發(fā)送邏輯
else
local shared_id = session_id % 100000
local message_shared = _shared_dicts[shared_id]
local message_id = message_shared:incr("id", 1, 0)
message_shared:set("message." .. message_id, "message data", 60, session_id)
end
其他https://github.com/chenxiaofa/p
借鑒了這位同學(xué)的設(shè)計(jì)思路,實(shí)現(xiàn)了額外邏輯。如:
加入、退出、銷毀組 各 worker之間的 cmd內(nèi)部命令執(zhí)行 熱更新的特殊處理 等