MQ

MQ庫

MQ庫是CF框架針對httpd的Websocket路由實(shí)現(xiàn)的一種訂閱發(fā)布庫.

眾所周知: 客戶端在使用Websocket協(xié)議與服務(wù)端進(jìn)行通信的情況下, 服務(wù)端需要主動推送必須處于一種'定時(shí)器/訂閱'的狀態(tài)才能主動觸發(fā)推送消息到客戶端.

MQ借用Redis/MQTT的subscribe與publish實(shí)現(xiàn)此種功能.

MQ:new(opt)

opt.hostMQ Broker的主機(jī)名或IP地址.

opt.portMQ Broker的主機(jī)端口號.

opt.typeMQ Broker的類型(redis/mqtt).

opt.authMQ Broker(redis類型專有)用于redis broken驗(yàn)證.

opt.username/passwordMQ Broker(mqtt類型專有)用于username/password驗(yàn)證.

此方法返回一個(gè)MQ實(shí)例.

MQ:on(pattern, function handle(msg) end)

patternstring類型, 用來指定訂閱的位置;

handlefunction類型, 訂閱消息到回調(diào)處理函數(shù);

msg table類型, 消息根據(jù)不同類型broker各有差異;

這里根據(jù)script/ws.lua文件精簡之后給出一段代碼示例:

    local MQ = require "MQ"
    local class = require "class"


    local websocket = class("websocket")


    function websocket:ctor(opt)
        self.ws = opt.ws             -- websocket對象
        self.mq = MQ:new({host = 'localhost', port = 6379, type = 'redis'})
    end


    function websocket:on_open()
        self.mq:on('/test/*', function (msg) -- 消息隊(duì)列
            if not msg then
                self.ws:send('{"code":500,"message":"無法連接到mq broker(redis)"}')
                return self.ws:close()
            end
            self.ws:send('{"code":200}')
        end)
    end

此方法返回true表示訂閱成功, 但情況下可忽略. msg為nil的情況下說明mq broker已經(jīng)斷開連接.

MQ:emit(pattern, data)

patternstring類型, 用來指定data所需要發(fā)布到的訂閱位置;

datastring類型, 需要發(fā)布的消息;

  --- 代碼示例
  local ok = MQ:emit('/test/admin', '{"code":200}')

此方法返回true表示發(fā)布成功, 但是不要過于依賴返回值. 因?yàn)镸Q內(nèi)置斷線重連機(jī)制, 因?yàn)榫W(wǎng)絡(luò)異常等原因可能會很久才會返回.

MQ:close()

此方法用來正確關(guān)閉MQ的發(fā)布與訂閱.

如果使用者將MQ用在websocket, 需要關(guān)閉時(shí)應(yīng)該在websocket的on_close方法關(guān)閉(推薦)

如果用在其它地方, 請仔細(xì)調(diào)試過后再試用. 作者不保證在websocket以外的地方能正常(正確)工作.

此方法沒有返回值.

最后

有小伙伴對websocket長連接有興趣的可以下載cf發(fā)布在docker hub上的容器體驗(yàn)一下, 使用chrome的websocket客戶端插件(smart websocket client)就可以對接查看效果.

MQ庫是因?yàn)闉閷拥谌絽f(xié)議(gRPC等)平臺而自己創(chuàng)建的一種消息通訊方式. 很多使用者因?yàn)樽孕卸ㄖ苹瘏f(xié)議, 所以CF作者用MQ示例來作為一種最直觀的對接方式.