返回介绍

redis 接口的二次封装(发布订阅)

发布于 2019-08-25 05:47:51 字数 2794 浏览 1084 评论 0 收藏 0

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

上一小节有关订阅部分的代码,请看:

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

如果你对这篇内容有疑问,欢迎到本站社区发帖提问 参与讨论,获取更多帮助,或者扫码二维码加入 Web 技术交流群。

扫码二维码加入Web技术交流群

发布评论

需要 登录 才能够评论, 你可以免费 注册 一个本站的账号。
列表为空,暂无数据
    我们使用 Cookies 和其他技术来定制您的体验包括您的登录状态等。通过阅读我们的 隐私政策 了解更多相关信息。 单击 接受 或继续使用网站,即表示您同意使用 Cookies 和您的相关数据。
    原文