基于Redis和openresty实现高并发缓存架构

概述

本文是对项目中 QPS 高并发相关问题的一种解决方案，利用 Nginx 与 Redis 的高并发、超低延迟响应，结合 Canal 进行实现。

openrestry官网

当程序需要提供较高的并发访问时，往往需要在程序中引入缓存技术，通常都是使用Redis 作为缓存，如若再更进一步提升性能，不仅要使用 redis 还要提高 并发,使用能支持超高并发的组件，并将请求响应大部分落在这些组件中。

原本访问缓存逻辑

User—> Nginx -> Tomcat -> Redis

User—> Nginx -> Redis

相关组件可自由下载
懒人直通: openresty/openresty:1.25.3.1-4-alpine-fat redis-7.0.15 docker离线镜像安装包

缓存架构设计

基本数据库 mysql ，整合 canal , 异步的同步数据至 redis 中，主要是利用 redis 与 nginx 的低延迟与高并发。
在这里插入图片描述

HTML页面做缓存，浏览器端可以缓存HTML页面和其他静态资源，防止用户频繁刷新对后端造成巨大压力
Lvs实现记录不同协议以及不同用户请求链路缓存
Nginx这里会做HTML页面缓存配置以及Nginx自身缓存配置(本次没有做Nginx缓存)
数据查找这里用Lua取代了其他语言查找，提高了处理的性能效率，并发处理能力将大大提升
集成Canal实现数据库数据增量实时同步Redis

实践

代码

路由

nginx 路由配置

server {listen 8000;set $target_server '';location /test{access_by_lua_file /usr/local/openresty/nginx/lua/router.lua;proxy_pass $target_server;}location /cache{default_type text/html;content_by_lua_file /usr/local/openresty/nginx/lua/cache_redis.lua;}
}

业务

创建文件 cache_redis.lua

local redis = require "redis_iresty"
local cjson = require("cjson")
local ngx_ERR = ngx.ERR
local ngx_exit = ngx.exit
local ngx_print = ngx.print
-- local ngx_re_match = ngx.re.match
local ngx_var = ngx.var-- 响应输出内容
-- body   http输出body内容
-- status http状态码
-- header http响应头,table格式
local function response(body,status,header)ngx.status = statusif header thenfor key, val in pairs(header) dongx.header[key] = valendend--ngx_print(body)ngx.say(cjson.encode(body))ngx_exit(ngx.status)
endlocal opts = {ip = "10.32.36.142",port = "6379",password = "123456",db_index = 0
}local red = redis:new(opts)local status = 200
local header = {}
local content = {}-- 返回的是一个table类型
local args = ngx.req.get_uri_args()
-- 获取名为"key"的参数
local key = args["key"]  header['content_type'] = 'application/json; charset=utf-8'
local value = red:get(key)
content['data'] = value
content['msg'] = '数据获取成功'
content['key'] = key
response(content,status,header)

封装redis

网上寻找的 redis 二次封装 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" thennew_tab = function (narr, nrec) return {} end
endlocal _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" thenfor k,v in pairs(res) doif v ~= ngx.null thenreturn falseendendreturn trueelseif res == ngx.null thenreturn trueelseif res == nil thenreturn trueendreturn false
endfunction _M.close_redis(self, redis)  if not redis then  return  end  --释放连接(连接池实现)local pool_max_idle_time = self.pool_max_idle_time --最大空闲时间 毫秒  local pool_size = self.pool_size --连接池大小  local ok, err = redis:set_keepalive(pool_max_idle_time, pool_size)  if not ok then  ngx.say("set keepalive error : ", err)  end  
end  -- change connect address as you need
function _M.connect_mod( self, redis )redis:set_timeout(self.timeout)local ok, err = redis:connect(self.ip, self.port)if not ok then  ngx.say("connect to redis error : ", err)  return self:close_redis(redis)  endif self.password then ----密码认证local count, err = redis:get_reused_times()if 0 == count then ----新建连接，需要认证密码ok, err = redis:auth(self.password)if not ok thenngx.say("failed to auth: ", err)returnendelseif err then  ----从连接池中获取连接，无需再次认证密码ngx.say("failed to get reused times: ", err)returnendendreturn ok,err;
endfunction _M.init_pipeline( self )self._reqs = {}
endfunction _M.commit_pipeline( self )local reqs = self._reqsif nil == reqs or 0 == #reqs thenreturn {}, "no pipeline"elseself._reqs = nilendlocal redis, err = redis_c:new()if not redis thenreturn nil, errendlocal ok, err = self:connect_mod(redis)if not ok thenreturn {}, errendredis:init_pipeline()for _, vals in ipairs(reqs) dolocal fun = redis[vals[1]]table.remove(vals , 1)fun(redis, unpack(vals))endlocal results, err = redis:commit_pipeline()if not results or err thenreturn {}, errendif is_redis_null(results) thenresults = {}ngx.log(ngx.WARN, "is null")end-- table.remove (results , 1)--self.set_keepalive_mod(redis)self:close_redis(redis)  for i,value in ipairs(results) doif is_redis_null(value) thenresults[i] = nilendendreturn results, err
endlocal function do_command(self, cmd, ... )if self._reqs thentable.insert(self._reqs, {cmd, ...})returnendlocal redis, err = redis_c:new()if not redis thenreturn nil, errendlocal ok, err = self:connect_mod(redis)if not ok or err thenreturn nil, errendredis:select(self.db_index)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, errendif is_redis_null(result) thenresult = nilend--self.set_keepalive_mod(redis)self:close_redis(redis)  return result, err
endfor i = 1, #commands dolocal cmd = commands[i]_M[cmd] =function (self, ...)return do_command(self, cmd, ...)end
endfunction _M.new(self, opts)opts = opts or {}local timeout = (opts.timeout and opts.timeout * 1000) or 1000local db_index= opts.db_index or 0local ip = opts.ip or '127.0.0.1'local port = opts.port or 6379local password = opts.passwordlocal pool_max_idle_time = opts.pool_max_idle_time or 60000local pool_size = opts.pool_size or 100return setmetatable({timeout = timeout,db_index = db_index,ip = ip,port = port,password = password,pool_max_idle_time = pool_max_idle_time,pool_size = pool_size,_reqs = nil }, mt)
endreturn _M