openresty 學(xué)習(xí)筆記三:連接redis和進(jìn)行相關(guān)操作

openresty 因其非阻塞的調(diào)用，令服務(wù)器擁有高性能高并發(fā)，當(dāng)涉及到數(shù)據(jù)庫(kù)操作時(shí)，更應(yīng)該選擇有高速讀寫(xiě)速度的redis進(jìn)行數(shù)據(jù)處理。避免其應(yīng)為讀寫(xiě)數(shù)據(jù)而造成瓶頸。

openresty 默認(rèn)就帶了redis的庫(kù)，這里先梳理下其自帶redis連接庫(kù)的操作流程，再根據(jù)存在問(wèn)題進(jìn)行二次封裝。

自帶redis連接庫(kù)的操作流程

首先是連接redis

local redis = require "resty.redis"
local redisCache = redis.new()
local ok ,err = redisCache.connect(redisCache,127.0.0.1,6379)
redisCache:set_timeout(1000)

if not ok then
    ngx.log(ngx.ERR, "failed to connect: ", err)
    return
end

　　

簡(jiǎn)單的reids操作

local code , err = redisCache:get("code")
local res , err = redisCache:incr("code")
local ok , err = redisCache:set("code",1)

　　

然后操作后要進(jìn)行close

ok , err = redisCache:close()

　　

如果項(xiàng)目中有大量redis操作，使用這個(gè)標(biāo)準(zhǔn)庫(kù)就變成會(huì)有大量的代碼在重復(fù)創(chuàng)建連接-->數(shù)據(jù)操作-->關(guān)閉連接（或放到連接池），甚至還要考慮不同的 return 情況做不同處理。

redis 庫(kù)的二次封裝

針對(duì)以上問(wèn)題，也通過(guò)學(xué)習(xí)openresty的最佳實(shí)踐，直接用上其中已經(jīng)做好的二次封裝。這個(gè)封裝是要實(shí)現(xiàn)一下目的

new、connect 函數(shù)合體，使用時(shí)只負(fù)責(zé)申請(qǐng)，盡量少關(guān)心什么時(shí)候具體連接、釋放；
默認(rèn) redis 數(shù)據(jù)庫(kù)連接地址，但是允許自定義；
每次 redis 使用完畢，自動(dòng)釋放 redis 連接到連接池供其他請(qǐng)求復(fù)用；
要支持 redis 的重要優(yōu)化手段 pipeline；

我只理解了前三點(diǎn)，最后的pipeline也還不是很懂，先貼上代碼：

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(self.db_host, self.db_port)
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

    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 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
    local db_host= opts.host or '127.0.0.1'
    local db_port= opts.port or 6379

    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,
            db_host = db_host,
            db_port = db_port,
            _reqs = nil }, mt)
end

return _M

　　

這個(gè)二次封裝除了解決上面幾個(gè)問(wèn)題，其實(shí)還重寫(xiě)了兩個(gè)方法，不過(guò)訂閱和發(fā)布暫時(shí)沒(méi)用到，所以還沒(méi)去細(xì)看這一部分。

使用示例

_config.redisConfig = {
    timeout = 1000,
    host = '127.0.0.1',
    port = 6379,
}

local redis = require (ngx.var.SERVER_DIR .. ".common.myRedis")
local redisCache = redis:new(config.redisConfig)

local accessDetails , err = redisCache:get("code")

　　

甚至可以直接

local redisCache = redis:new()

總結(jié)

以上是生活随笔為你收集整理的openresty 学习笔记三:连接redis和进行相关操作的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。