最近實現(xiàn)一個二維碼掃描登錄的功能，當用戶用移動設備掃描PC端頁面的二維碼之后，移動設備通過常規(guī)HTTP短連接向服務器獲取認證數(shù)據(jù)，認證通過后，服務器向PC瀏覽器主動推送帳號相關信息以完成PC端頁面的登錄。

服務器主動向瀏覽器推送數(shù)據(jù)，基本上就是ajax輪詢、iframe stream、websocket等等，可以參見 《Comet (web技術)》

推送服務器有很多種，當然用強大穩(wěn)定又順手的nginx了。 nginx相關的推送插件模塊有nginx-push-stream-module、nginx_http_push_module，但是很遺憾，可配置不可編程。又到我們的主角 OpenResty （OpenResty (aka. ngx_openresty) is a full-fledged web application server by bundling the standard Nginx core, lots of 3rd-party Nginx modules, as well as most of their external dependencies）出馬的時候了。

1、PC瀏覽器向php發(fā)起ajax請求，獲取一個與當前session相關的唯一的二維碼URL，和一個唯一的sub訂閱URL。

2、PC瀏覽器顯示二維碼，并對sub訂閱URL發(fā)起ajax長連接或者websocket連接，這個請求將直接由nginx來hold住，超時時間由配置參數(shù)push_free_timeout決定。

3、手機端掃描并解析二維碼，向php發(fā)起認證。

4、php收到移動設備的請求后，解出sessionid，向nginx的pub接口發(fā)布數(shù)據(jù)，該數(shù)據(jù)將被直接投遞到對應的sub接口，并回傳到瀏覽器。

5、如果sub接口在push_free_timeout指定的時間內一直沒有收到數(shù)據(jù)，將主動斷開與瀏覽器端的連接。此時，瀏覽器可以根據(jù)業(yè)務場景決定是否重新發(fā)起連接。

出于性能考慮，使用ngx.shared共享內存存儲消息，只能共享于一個ngx實例內，對于10k級別的聊天室并發(fā)連接應該是夠用了。
使用redis作為外部存儲，也是可以的，如果100k的并發(fā)，需要注意ngx對nosql發(fā)起連接時耗盡socket，當然這個是可以解決的。
更大規(guī)模的并發(fā)，值得自研推送服務器。

在生產(chǎn)環(huán)境某個LB節(jié)點上試運行過，openresty跑著waf\fastcgi proxy\http proxy\comet。常態(tài)1k并發(fā)連接數(shù)，load 0.01，40k并發(fā)時，load只有0.15。

下面是相關測試代碼，如果有空完善了再托管到github上，考慮寫一個聊天室的完整demo。

resty.push基礎模塊（需要使用到ngx.shared共享內存來存儲消息，在nginx.conf的http段配置lua_shared_dict push 10m;）

--[[
-- /usr/local/openresty/lualib/resty/push.lua
-- push.lua ，resty.push 基于nginx_lua的push推送方案
-- 支持多對多頻道 
-- 支持long-pooling, stream, websocket
--
-- Author: chuyinfeng.com <Liujiaxiong@kingsoft.com> 
-- 2014.03.12
--]]

local _M = {_VERSION = '0.01'}

local function debug(msg)
  --ngx.say(msg)
  --ngx.flush(true)
end

-- 配置信息
_M.config = {
  -- 推送間隔，1s
  ['push_interval'] = 1,
  -- 消息隊列最大長度
  ['msglist_len'] = 100,
  -- 消息生存周期
  ['msg_lefttime'] = 3,
  -- 頻道空閑超時
  ['channel_timeout'] = 30,
  -- 推送空閑超時，在改時間段內無消息則關閉當前推送連接
  ['push_free_timeout'] = 10,
  -- 共享內存名
  ['store_name'] = 'push',
  -- 頻道號
  ['channels'] = {1, 2},
}


-- 頻道數(shù)量
_M.channels_len = 0
-- 當前讀位置
_M.idx_read = 0
-- 共享內存
_M.store = nil


-- cjson 模塊
local cjson = require "cjson"

--[[ 
-- 設置
--]]
_M.opt = function(self, k, v)
  local t = type(k)
  if t == 'table' then
    for key, val in pairs(k) do
      self.config[key] = val
    end
  end

  if t == 'string' then
    self.config[k] = v
  end

  self.channels_len = table.maxn(self.config['channels'])
  self.store = ngx.shared[self.config['store_name']]

end


--[[
-- 向頻道寫入消息
--
-- @param ngx.shared.dict, 共享內存
-- @param string channel_id,可用ngx.crc32_long生成
-- @param int channel_timeout, 頻道空閑超時時間
-- @param string msg,消息內容 必須為字符串
-- @param int msg_lefttime， 消息生存周期
-- @param int msglist_len, 消息隊列長度
-- @return boolean
--]]
local function _write(store, channel_id, channel_timeout, msg, msg_lefttime, msglist_len)
  local idx, ok, err

  -- 消息當前讀取位置計數(shù)器+1
  idx, err = store:incr(channel_id, 1)

  -- 如果異常，則新建頻道
  if err then
    ok, err = store:set(channel_id, 1, channel_timeout)
    if err then return 0 end
    idx = 1
  else
    store:replace(channel_id, idx, channel_timeout)
  end


  -- 寫入消息
  debug("write " .. channel_id .. idx .. " , lefttime: " .. msg_lefttime.. " , msg: " .. msg)
  ok, err = store:set('m' .. channel_id .. idx, msg,  msg_lefttime)
  if err then return 0 end

  -- 清除隊列之前的舊消息
  if idx > msglist_len then
    store:delete('m' .. channel_id .. (idx - msglist_len))
  end

  return idx
end

--[[
-- 從頻道讀取消息
--
-- @param int channel_id, 必須為整形，可用ngx.crc32_long生成
-- @param int offset，歷史偏移量，最小為0
-- @return int len, 剩余消息數(shù)量
-- @return string msg, 消息  
--]]
local _read = function (store, channel_id, msglist_len, idx_read)
  local idx_msg, err, msg

   

  -- 獲取最新消息的位置
  idx_msg, _ = store:get(channel_id)
  idx_msg = idx_msg or 0

  if idx_msg == 0 then
    idx_read = 0
  end
  
  if idx_msg - idx_read > msglist_len then
    idx_read = idx_msg - msglist_len
  end
  
  if idx_read < idx_msg then
    idx_read = idx_read + 1
    msg, _ = store:get('m' .. channel_id .. idx_read) 
  end

  -- 返回讀的位置和消息的最大位置，以及消息
  return idx_read, idx_msg, msg
end

--[[
-- 推送消息
-- @param callback wrapper, 消息包裝回調函數(shù)
--]]
_M.push = function(self, wrapper)
  local flag_work = true
  local flag_read = true
  local idx_read, idx_msg, msg, err
  local time_last_msg = ngx.time()
  
  while flag_work do
     for i = 1, self.channels_len do
      -- 循環(huán)讀取當前頻道，直到EOF
      flag_read = true
      while flag_read do
        debug("read from idx_read: " .. self.idx_read)
        self.idx_read, idx_msg, msg = _read(self.store, self.config['channels'][i], self.config['msglist_len'], self.idx_read)
        if msg ~= nil then
          debug("got msg and wrapper  msg: " .. msg)
          time_last_msg = ngx.time()
          wrapper(msg)
        end
        debug("idx_read: " .. self.idx_read .. ", idx_msg: " .. idx_msg)
        
        if self.idx_read == idx_msg then flag_read = false end
      end
      
    end

    debug("push_free: " .. ngx.time() - time_last_msg)
    if ngx.time() - time_last_msg  >= self.config['push_free_timeout'] then
      debug("push_timeout: " .. " last: " .. time_last_msg .. " , now: " .. ngx.time())
      flag_work = false
    end
    

    debug("sleep: " .. self.config['push_interval'])
    ngx.sleep(self.config['push_interval'])
  end

end

--[[
-- 發(fā)送消息到指定頻道
--]]
_M.send = function(self, msg)
  local idx = 0
  for i = 1, self.channels_len do
    idx = _write(self.store, self.config['channels'][i], self.config['channel_timeout'], msg, self.config['msg_lefttime'], self.config['msglist_len'])
  end
  return true
end

--[[
-- jsonp格式化
--]]
_M.jsonp = function(self, data, cb)
    if cb then
        return cb .. "(" .. cjson.encode(data)  .. ");"
    else
        return cjson.encode(data)
    end
end


--[[
-- 公開成員
--]]

_M.new = function(self)
  return setmetatable({}, { __index = _M })
end

return _M

public發(fā)布

local push = require "resty.push"

local function exit(is_ws)
  if is_ws == nil then ngx.eof() end
  ngx.exit(444)
end
local ok, err = ngx.on_abort(exit)
if err then return end

local pub = push:new()

pub:opt({
  ['channels'] = 123,
  ['push_interval'] = 0.1,
  ['push_free_timeout'] = 27,
})

ngx.req.read_body()
local body, err = ngx.req.get_post_args()
if err then exit() end
if pub:send(body['data'] or '') then
  ngx.header['Content-Type'] = 'text/javascript;charset=UTF-8'
  ngx.status = ngx.HTTP_OK
  ngx.say(pub:jsonp({['status'] = 1}, args['callback']))
end

subscribe 訂閱接口

local push = require "resty.push"

local function exit(is_ws)
  if is_ws == nil then ngx.eof() end
  ngx.exit(444)
end
local ok, err = ngx.on_abort(exit)
if err then return end

local args = ngx.req.get_uri_args()

local sub = push:new()

sub:opt({
  ['channels'] = 123,
  ['push_interval'] = 0.1,
  ['push_free_timeout'] = 27,
})

local wrapper = function(msg)
  ngx.header['Content-Type'] = 'text/javascript;charset=UTF-8'
  ngx.status = ngx.HTTP_OK
  ngx.say(sub:jsonp(msg, args['callback']))
  exit()
end

sub:push(wrapper)
wrapper(sub:jsonp({['status'] = 1, ['tips'] = 'timeout'}))