用redis实现openresty的缓存同步

"一切单机缓存都是魔鬼,与其被消灭，不如与其共舞"

来由

之前接到我们uAuth的一个bug,具体原因为,当一个用户改密后，原token理应失效，但是线上时常会有原token访问的正常的情况。可是在测试环境上，确无论如何也复现不出来。

后来仔细分析了源码,是由于token的存储用了openresty的缓存，当token失效后，只在线上的n台服务器中的一台做了失效处理，而其他的n-1台的缓存仍然有效。

思路

缓存不一致 ———— 这确实是好多场景容易碰到的问题，那么怎么办？解决方式有二：

干掉openresty的缓存,将存储设计由openresty缓存/redis/mysql 改为redis/mysql 两层结构。
设计openresty的缓存同步机制，从根儿上解决这个问题。

方式1，确实是简单直接有效的方式，but:

使用openresty，不让我用缓存，那和一条咸鱼有什么区别？

于是，我选择了第2种方式，那么问题来了，如何设计这个同步机制？经典的同步肯定是发布/订阅来搞定，第一时间自然想到了kafka，可是查了一圈发现openresty中官方的resty.kafka只支持生产,并且大多数场景都是用来日志记录。到时redis,有一个经典的subscribe例子,那么好，就这么干。

撸码

封装发布/订阅操作

既然同步，那咱就整到位。第一步,封装一个redis_message.lua

 1 local redis_c = require "resty.redis"
 2 local cjson = require 'cjson.safe'
 3 local M = {}
 4 local mt = {__index = M}
 5 function M:new(cfg)
 6     ocal ins = {
 7         timeout = cfg.timeout or 60000,
 8         pool = cfg.pool or {maxIdleTime = 120000,size = 200},
 9         database = cfg.database or 0,
10         host = cfg .host,
11         port = cfg. port,
12         password = cfg .password or ""
13     }
14     setmetatable(ins,mt)
15     return ins
16 end
17 local function get_con(cfg)
18     local red = redis_c:new()
19     red:set_timeout(cfg.timeout)
20     local ok,err = red:connect(cfg.host,cfg.port)
21     if not ok then
22         return nil
23     end
24     local count ,err = red:get_reused_times()
25     if 0 == count then
26         ok ,err = red:auth(cfg.password)
27     elseif err then
28         return nil
29     end
30 
31     red:select(cfg.database)
32     return red
33 end
34  
35 local function keep_alive(red,cfg)
36     local ok,err = red:set_keepalive(cfg.pool.maxIdleTime,cfg.pool.size)
37     if not ok then
38         red:close()
39     end
40     return true
41 end
42 
43 function M:subscribe(key,func)
44     local co = coroutine.create(function()
45         local red = get_con(self)
46         local ok,err = red:subscribe(key)
47         if not ok then
48             return err
49         end
50         local flag = true
51         while flag do
52             local res,err = red:read_reply()
53             if err then
54                 ;
55             else
56                 if res[1] == "message" then
57                     local obj = cjson.decode(res[3])
58                     flag = func(obj.msg)
59                 end
60             end
61             red:set_keepalive(100,100)
62         end
63     end)
64     coroutine.resume(co)
65 
66 end
67 function M:publish(key,msg)
68     local red = get_con(self)
69     local obj = {}
70     obj.type = type(msg)
71     obj.msg = msg
72     local ok,err = red:publish(key,cjson.encode(obj))
73     if not ok then
74         return false
75     else
76         return true
77     end
78     keep_alive(red,self)
79 end
80 
81 return M

View Code