Redis 接口的二次封装

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

  1. local redis = require("resty.redis")
  2. local red = redis:new()
  4. red:set_timeout(1000) -- 1 sec
  6. local ok, err = red:connect("127.0.0.1", 6379)
  7. if not ok then
  8.     ngx.say("failed to connect: ", err)
  9.     return
  10. end
  12. ok, err = red:set("dog", "an animal")
  13. if not ok then
  14.     ngx.say("failed to set dog: ", err)
  15.     return
  16. end
  18. ngx.say("set result: ", ok)
  20. -- put it into the connection pool of size 100,
  21. -- with 10 seconds max idle time
  22. local ok, err = red:set_keepalive(10000, 100)
  23. if not ok then
  24.     ngx.say("failed to set keepalive: ", err)
  25.     return
  26. end

这是一个标准的 Redis 接口调用，如果你的代码中 Redis 被调用频率不高，那么这段代码不会有任何问题。

但如果你的项目重度依赖 Redis，工程中有大量的程序在重复 创建连接—>数据操作—>关闭连接（或放到连接池） 这条完整的链路。甚至在调用完毕还要考虑不同的 return 情况做不同处理，你很快就会发现代码有大量的重复。

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

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

  1. local red     = redis:new()
  2. local ok, err = red:set("dog", "an animal")
  3. if not ok then
  4.     ngx.say("failed to set dog: ", err)
  5.     return
  6. end
  8. ngx.say("set result: ", ok)
  10. local res, err = red:get("dog")
  11. if not res then
  12.     ngx.say("failed to get dog: ", err)
  13.     return
  14. end
  16. if res == ngx.null then
  17.     ngx.say("dog not found.")
  18.     return
  19. end
  21. ngx.say("dog: ", res)

期望它自身具备以下几个特征：

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

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

  1. -- file name: resty/redis_iresty.lua
  2. local redis_c = require("resty.redis")
  5. local ok, new_tab = pcall(require, "table.new")
  6. if not ok or type(new_tab) ~= "function" then
  7.     new_tab = function (narr, nrec) return {} end
  8. end
  11. local _M = new_tab(0, 155)
  12. _M._VERSION = '0.01'
  15. local commands = {
  16.     "append",            "auth",              "bgrewriteaof",
  17.     "bgsave",            "bitcount",          "bitop",
  18.     "blpop",             "brpop",
  19.     "brpoplpush",        "client",            "config",
  20.     "dbsize",
  21.     "debug",             "decr",              "decrby",
  22.     "del",               "discard",           "dump",
  23.     "echo",
  24.     "eval",              "exec",              "exists",
  25.     "expire",            "expireat",          "flushall",
  26.     "flushdb",           "get",               "getbit",
  27.     "getrange",          "getset",            "hdel",
  28.     "hexists",           "hget",              "hgetall",
  29.     "hincrby",           "hincrbyfloat",      "hkeys",
  30.     "hlen",
  31.     "hmget",              "hmset",      "hscan",
  32.     "hset",
  33.     "hsetnx",            "hvals",             "incr",
  34.     "incrby",            "incrbyfloat",       "info",
  35.     "keys",
  36.     "lastsave",          "lindex",            "linsert",
  37.     "llen",              "lpop",              "lpush",
  38.     "lpushx",            "lrange",            "lrem",
  39.     "lset",              "ltrim",             "mget",
  40.     "migrate",
  41.     "monitor",           "move",              "mset",
  42.     "msetnx",            "multi",             "object",
  43.     "persist",           "pexpire",           "pexpireat",
  44.     "ping",              "psetex",            "psubscribe",
  45.     "pttl",
  46.     "publish",      --[[ "punsubscribe", ]]   "pubsub",
  47.     "quit",
  48.     "randomkey",         "rename",            "renamenx",
  49.     "restore",
  50.     "rpop",              "rpoplpush",         "rpush",
  51.     "rpushx",            "sadd",              "save",
  52.     "scan",              "scard",             "script",
  53.     "sdiff",             "sdiffstore",
  54.     "select",            "set",               "setbit",
  55.     "setex",             "setnx",             "setrange",
  56.     "shutdown",          "sinter",            "sinterstore",
  57.     "sismember",         "slaveof",           "slowlog",
  58.     "smembers",          "smove",             "sort",
  59.     "spop",              "srandmember",       "srem",
  60.     "sscan",
  61.     "strlen",       --[[ "subscribe",  ]]     "sunion",
  62.     "sunionstore",       "sync",              "time",
  63.     "ttl",
  64.     "type",         --[[ "unsubscribe", ]]    "unwatch",
  65.     "watch",             "zadd",              "zcard",
  66.     "zcount",            "zincrby",           "zinterstore",
  67.     "zrange",            "zrangebyscore",     "zrank",
  68.     "zrem",              "zremrangebyrank",   "zremrangebyscore",
  69.     "zrevrange",         "zrevrangebyscore",  "zrevrank",
  70.     "zscan",
  71.     "zscore",            "zunionstore",       "evalsha"
  72. }
  75. local mt = { __index = _M }
  78. local function is_redis_null( res )
  79.     if type(res) == "table" then
  80.         for k,v in pairs(res) do
  81.             if v ~= ngx.null then
  82.                 return false
  83.             end
  84.         end
  85.         return true
  86.     elseif res == ngx.null then
  87.         return true
  88.     elseif res == nil then
  89.         return true
  90.     end
  92.     return false
  93. end
  96. -- change connect address as you need
  97. function _M.connect_mod( self, redis )
  98.     redis:set_timeout(self.timeout)
  99.     return redis:connect("127.0.0.1", 6379)
  100. end
  103. function _M.set_keepalive_mod( redis )
  104.     -- put it into the connection pool of size 100, with 60 seconds max idle time
  105.     return redis:set_keepalive(60000, 1000)
  106. end
  109. function _M.init_pipeline( self )
  110.     self._reqs = {}
  111. end
  114. function _M.commit_pipeline( self )
  115.     local reqs = self._reqs
  117.     if nil == reqs or 0 == #reqs then
  118.         return {}, "no pipeline"
  119.     else
  120.         self._reqs = nil
  121.     end
  123.     local redis, err = redis_c:new()
  124.     if not redis then
  125.         return nil, err
  126.     end
  128.     local ok, err = self:connect_mod(redis)
  129.     if not ok then
  130.         return {}, err
  131.     end
  133.     redis:init_pipeline()
  134.     for _, vals in ipairs(reqs) do
  135.         local fun = redis[vals[1]]
  136.         table.remove(vals , 1)
  138.         fun(redis, unpack(vals))
  139.     end
  141.     local results, err = redis:commit_pipeline()
  142.     if not results or err then
  143.         return {}, err
  144.     end
  146.     if is_redis_null(results) then
  147.         results = {}
  148.         ngx.log(ngx.WARN, "is null")
  149.     end
  150.     -- table.remove (results , 1)
  152.     self.set_keepalive_mod(redis)
  154.     for i,value in ipairs(results) do
  155.         if is_redis_null(value) then
  156.             results[i] = nil
  157.         end
  158.     end
  160.     return results, err
  161. end
  164. function _M.subscribe( self, channel )
  165.     local redis, err = redis_c:new()
  166.     if not redis then
  167.         return nil, err
  168.     end
  170.     local ok, err = self:connect_mod(redis)
  171.     if not ok or err then
  172.         return nil, err
  173.     end
  175.     local res, err = redis:subscribe(channel)
  176.     if not res then
  177.         return nil, err
  178.     end
  180.     res, err = redis:read_reply()
  181.     if not res then
  182.         return nil, err
  183.     end
  185.     redis:unsubscribe(channel)
  186.     self.set_keepalive_mod(redis)
  188.     return res, err
  189. end
  192. local function do_command(self, cmd, ... )
  193.     if self._reqs then
  194.         table.insert(self._reqs, {cmd, ...})
  195.         return
  196.     end
  198.     local redis, err = redis_c:new()
  199.     if not redis then
  200.         return nil, err
  201.     end
  203.     local ok, err = self:connect_mod(redis)
  204.     if not ok or err then
  205.         return nil, err
  206.     end
  208.     local fun = redis[cmd]
  209.     local result, err = fun(redis, ...)
  210.     if not result or err then
  211.         -- ngx.log(ngx.ERR, "pipeline result:", result, " err:", err)
  212.         return nil, err
  213.     end
  215.     if is_redis_null(result) then
  216.         result = nil
  217.     end
  219.     self.set_keepalive_mod(redis)
  221.     return result, err
  222. end
  225. for i = 1, #commands do
  226.     local cmd = commands[i]
  227.     _M[cmd] =
  228.             function (self, ...)
  229.                 return do_command(self, cmd, ...)
  230.             end
  231. end
  234. function _M.new(self, opts)
  235.     opts = opts or {}
  236.     local timeout = (opts.timeout and opts.timeout * 1000) or 1000
  237.     local db_index= opts.db_index or 0
  239.     return setmetatable({
  240.             timeout  = timeout,
  241.             db_index = db_index,
  242.             _reqs = nil }, mt)
  243. end
  246. 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 搞垮了。