redis接口的二次封装

旧城等待， 2022-05-25 08:59 238阅读 0赞

> 先看一下官方的调用示例代码：

local redis = require "resty.redis"
    local red = redis:new()
    
    red:set_timeout(1000) -- 1 sec
    
    local ok, err = red:connect("127.0.0.1", 6379)
    if not ok then
        ngx.say("failed to connect: ", err)
        return
    end
    
    ok, err = red:set("dog", "an animal")
    if not ok then
        ngx.say("failed to set dog: ", err)
        return
    end
    
    ngx.say("set result: ", ok)
    
    -- put it into the connection pool of size 100,
    -- with 10 seconds max idle time
    local ok, err = red:set_keepalive(10000, 100)
    if not ok then
        ngx.say("failed to set keepalive: ", err)
        return
    end

这是一个标准的redis接口调用，如果你的代码中redis被调用频率不高，那么这段代码不会有任何问题。但如果你的项目重度依赖redis，工程中有大量的代码在重复这样一个动作，创建连接-->数据操作-->关闭连接（或放到连接池）这个完整的链路调用完毕，甚至还要考虑不同的return情况做不同处理，就很快发现代码中有大量的重复。

`Lua`是不支持面向对象的。很多人用尽各种招术利用元表来模拟。可是，`Lua`的发明者似乎不想看到这样的情形，因为他们把取长度的`__len`方法以及析构函数`__gc`留给了 C API。纯 Lua 只能望洋兴叹。

> 我们期望的代码应该是这样的：

local red = redis:new()
    local ok, err = red:set("dog", "an animal")
    if not ok then
        ngx.say("failed to set dog: ", err)
        return
    end
    
    ngx.say("set result: ", ok)
    
    local res, err = red:get("dog")
    if not res then
        ngx.say("failed to get dog: ", err)
        return
    end
    
    if res == ngx.null then
        ngx.say("dog not found.")
        return
    end
    
    ngx.say("dog: ", res)

并且他自身具备以下几个特征：

*  new、connect函数合体，使用时只负责申请，尽量少关心什么时候具体连接、释放
 *  默认redis数据库连接地址，但是允许自定义
 *  每次redis使用完毕，自动释放redis连接到连接池供其他请求复用
 *  要支持redis的重要优化手段 pipeline

> 不卖关子，只要干货，我们最后是这样干的，可以这里看到[gist代码][gist]

-- file name: resty/redis_iresty.lua
    local redis_c = require "resty.redis"
    
    local ok, new_tab = pcall(require, "table.new")
    if not ok or type(new_tab) ~= "function" then
        new_tab = function (narr, nrec) return {} end
    end
    
    local _M = new_tab(0, 155)
    _M._VERSION = '0.01'
    
    local commands = {
        "append",            "auth",              "bgrewriteaof",
        "bgsave",            "bitcount",          "bitop",
        "blpop",             "brpop",
        "brpoplpush",        "client",            "config",
        "dbsize",
        "debug",             "decr",              "decrby",
        "del",               "discard",           "dump",
        "echo",
        "eval",              "exec",              "exists",
        "expire",            "expireat",          "flushall",
        "flushdb",           "get",               "getbit",
        "getrange",          "getset",            "hdel",
        "hexists",           "hget",              "hgetall",
        "hincrby",           "hincrbyfloat",      "hkeys",
        "hlen",
        "hmget",              "hmset",      "hscan",
        "hset",
        "hsetnx",            "hvals",             "incr",
        "incrby",            "incrbyfloat",       "info",
        "keys",
        "lastsave",          "lindex",            "linsert",
        "llen",              "lpop",              "lpush",
        "lpushx",            "lrange",            "lrem",
        "lset",              "ltrim",             "mget",
        "migrate",
        "monitor",           "move",              "mset",
        "msetnx",            "multi",             "object",
        "persist",           "pexpire",           "pexpireat",
        "ping",              "psetex",            "psubscribe",
        "pttl",
        "publish",      --[[ "punsubscribe", ]]   "pubsub",
        "quit",
        "randomkey",         "rename",            "renamenx",
        "restore",
        "rpop",              "rpoplpush",         "rpush",
        "rpushx",            "sadd",              "save",
        "scan",              "scard",             "script",
        "sdiff",             "sdiffstore",
        "select",            "set",               "setbit",
        "setex",             "setnx",             "setrange",
        "shutdown",          "sinter",            "sinterstore",
        "sismember",         "slaveof",           "slowlog",
        "smembers",          "smove",             "sort",
        "spop",              "srandmember",       "srem",
        "sscan",
        "strlen",       --[[ "subscribe",  ]]     "sunion",
        "sunionstore",       "sync",              "time",
        "ttl",
        "type",         --[[ "unsubscribe", ]]    "unwatch",
        "watch",             "zadd",              "zcard",
        "zcount",            "zincrby",           "zinterstore",
        "zrange",            "zrangebyscore",     "zrank",
        "zrem",              "zremrangebyrank",   "zremrangebyscore",
        "zrevrange",         "zrevrangebyscore",  "zrevrank",
        "zscan",
        "zscore",            "zunionstore",       "evalsha"
    }
    
    local mt = { __index = _M }
    
    local function is_redis_null( res )
        if type(res) == "table" then
            for k,v in pairs(res) do
                if v ~= ngx.null then
                    return false
                end
            end
            return true
        elseif res == ngx.null then
            return true
        elseif res == nil then
            return true
        end
    
        return false
    end
    
    -- change connect address as you need
    function _M.connect_mod( self, redis )
        redis:set_timeout(self.timeout)
        return redis:connect("127.0.0.1", 6379)
    end
    
    function _M.set_keepalive_mod( redis )
        -- put it into the connection pool of size 100, with 60 seconds max idle time
        return redis:set_keepalive(60000, 1000)
    end
    
    function _M.init_pipeline( self )
        self._reqs = {}
    end
    
    function _M.commit_pipeline( self )
        local reqs = self._reqs
    
        if nil == reqs or 0 == #reqs then
            return {}, "no pipeline"
        else
            self._reqs = nil
        end
    
        local redis, err = redis_c:new()
        if not redis then
            return nil, err
        end
    
        local ok, err = self:connect_mod(redis)
        if not ok then
            return {}, err
        end
    
        redis:init_pipeline()
        for _, vals in ipairs(reqs) do
            local fun = redis[vals[1]]
            table.remove(vals , 1)
    
            fun(redis, unpack(vals))
        end
    
        local results, err = redis:commit_pipeline()
        if not results or err then
            return {}, err
        end
    
        if is_redis_null(results) then
            results = {}
            ngx.log(ngx.WARN, "is null")
        end
        -- table.remove (results , 1)
    
        self.set_keepalive_mod(redis)
    
        for i,value in ipairs(results) do
            if is_redis_null(value) then
                results[i] = nil
            end
        end
    
        return results, err
    end
    
    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
    
    local function do_command(self, cmd, ... )
        if self._reqs then
            table.insert(self._reqs, {cmd, ...})
            return
        end
    
        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 fun = redis[cmd]
        local result, err = fun(redis, ...)
        if not result or err then
            -- ngx.log(ngx.ERR, "pipeline result:", result, " err:", err)
            return nil, err
        end
    
        if is_redis_null(result) then
            result = nil
        end
    
        self.set_keepalive_mod(redis)
    
        return result, err
    end
    
    function _M.new(self, opts)
        opts = opts or {}
        local timeout = (opts.timeout and opts.timeout * 1000) or 1000
        local db_index= opts.db_index or 0
    
        for i = 1, #commands do
            local cmd = commands[i]
            _M[cmd] =
                    function (self, ...)
                        return do_command(self, cmd, ...)
                    end
        end
    
        return setmetatable({
                timeout = timeout,
                db_index = db_index,
                _reqs = nil }, mt)
    end
    
    return _M

> 调用示例代码：

local redis = require "resty.redis_iresty"
    local red = redis:new()
    
    local ok, err = red:set("dog", "an animal")
    if not ok then
        ngx.say("failed to set dog: ", err)
        return
    end
    
    ngx.say("set result: ", ok)

在最终的示例代码中看到，所有的连接创建、销毁连接、连接池部分，都被完美隐藏了，我们只需要业务就可以了。妈妈再也不用担心我把redis搞垮了。

Todo list：目前`resty.redis`并没有对redis 3.0的集群API做支持，既然统一了redis的入口、出口，那么对这个`redis_iresty`版本做适当调整完善，就可以支持redis 3.0的集群协议。由于我们目前还没引入redis集群，这里也希望有使用的同学贡献自己的补丁或文章。

[gist]: https://gist.github.com/moonbingbing/9915c66346e8fddcefb5