Redis 接口的二次封装（发布订阅）

Redis 接口的二次封装（发布订阅）

其实这一小节完全可以放到上一个小节，只是这里用了完全不同的玩法，所以我还是决定单拿出来分享一下这方面的小细节。

上一小节有关订阅部分的代码，请看：

function _M.subscribe( self, channel )
    local redis, err = redis_c:new()
    if not redis then
        return nil, err
    end
    local ok, err = self:connect_mod(redis)
    if not ok or err then
        return nil, err
    end
    local res, err = redis:subscribe(channel)
    if not res then
        return nil, err
    end
    res, err = redis:read_reply()
    if not res then
        return nil, err
    end
    redis:unsubscribe(channel)
    self.set_keepalive_mod(redis)
    return res, err
end

其实这里的实现是有问题的，各位看官，你能发现这段代码的问题么？给个提示，在高并发订阅场景下，极有可能存在漏掉部分订阅信息。原因在于每次订阅到内容后，都会把 Redis 对象进行释放，处理完订阅信息后再次去连接 Redis，在这个时间差里面，很可能有消息已经漏掉了。

通过下面的代码可以解决这个问题：

function _M.subscribe( self, channel )
    local redis, err = redis_c:new()
    if not redis then
        return nil, err
    end
    local ok, err = self:connect_mod(redis)
    if not ok or err then
        return nil, err
    end
    local res, err = redis:subscribe(channel)
    if not res then
        return nil, err
    end
    -- 封装成一个函数，开始
    local function do_read_func ( do_read )
        if do_read == nil or do_read == true then
            res, err = redis:read_reply()
            if not res then
                return nil, err
            end
            return res
        end
        redis:unsubscribe(channel)
        self.set_keepalive_mod(redis)
        return
    end
    -- 结束
    return do_read_func  -- 返回上面封装的函数
end

调用示例代码：

local red  = redis:new({timeout=1000})
local func = red:subscribe( "channel" )
if not func then
  return nil
end
while true do
    local res, err = func()
    if err then
        func(false)
    end
    ... ...
end
return cbfunc

另一个潜在的问题是，调用了 unsubscribe 之后，Redis 对象里面有可能还遗留没被读取的数据。在这种情况下，无法直接通过 set_keepalive_mod 复用连接。什么时候会发生这样的情况呢？

当 Redis 对象处于 subscribe 状态时，Redis 会给它推送订阅的消息，然后我们通过 read_reply 把消息读出来。调用 unsubscribe 的时候，只是退订了对应的频道，并不会把当前接收到的数据清空。如果要想复用该连接，我们就需要保证清空当前读取到的数据，保证它是干净的。就像这样：

local res, err = red:unsubscribe("ch")
if not res then
    ngx.log(ngx.ERR, err)
    return
else
    -- redis 推送的消息格式，可能是
    -- {"message", ...} 或
    -- {"unsubscribe", $channel_name, $remain_channel_num}
    -- 如果返回的是前者，说明我们还在读取 Redis 推送过的数据
    if res[1] ~= "unsubscribe" then
        repeat
            -- 需要抽空已经接收到的消息
            res, err = red:read_reply()
            if not res then
                ngx.log(ngx.ERR, err)
                return
            end
        until res[1] == "unsubscribe"
    end
    -- 现在再复用连接，就足够安全了
    self.set_keepalive_mod(redis)
end

redis 接口的二次封装（发布订阅）

Redis 接口的二次封装（发布订阅）