OpenResty 对接Redis，并读写数据_openresty redis

OpenResty

使用场景，静态资源写入的redis中，以及数据读取

结构设计

挂载Lua脚本

http {
    include       mime.types;
		... ...
    lua_package_path "/apps/nginx/conf/lualib/?.lua;;";
    include upstream/*.conf;
    include hosts/*/upstream.conf;
    include hosts/*/pre-server.conf;
    include hosts/*/server.conf;
}
1
2
3
4
5
6
7
8
9

B端对数据写入

Server.conf

server {
    listen       443 ssl;
    server_name   galaxy.shangwenhe.com;
    index index.html index.htm;
    # 注入环境变量区分当前进行环境
    include common/subfilter/environment.conf;
    include error.conf;
    include hosts/galaxy.shangwenhe.com/locations/*.path;
    location / {
        default_type application/json;
        return 200 '{"domain": "galaxy.shangwenhe.com"}';
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13

Adam.path

hosts/galaxy.shangwenhe.com/locations/adam.path

location ^~ /adam/ {
    default_type application/json;
    content_by_lua_file /apps/nginx/conf/lualib/adam.lua; # 使用Lua脚本接管/adam/接口
}
1
2
3
4

C端对数据读取

Server.conf

server {
    listen       443 ssl;
    server_name   mgt.shangwenhe.com;
    index index.html index.htm;
    # 注入环境变量区分当前进行环境
    include common/subfilter/environment.conf;
    include error.conf;
    include hosts/mgt.shangwenhe.com/locations/*.path;
    location / {
        default_type application/json;
        return 200 '{"domain": "mgt.shangwenhe.com"}';
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13

Mgt.Adam.path

hosts/mgt.shangwenhe.com/locations/mgt.adam.path;

location ^~ /adam/ {
    client_body_buffer_size 5m;
    client_max_body_size 5m;
    if ($request_method = DELETE) {
        return 403;
    }
    default_type application/json;
    # 默认读取 body
    lua_need_request_body on;
    access_by_lua_file  /apps/nginx/conf/lualib/adam.mgt.lua;  # 使用Lua脚本接管/adam/接口
    # 需要以 / 结束
    proxy_pass http://galaxy_adam_es/; # 对接入ES
    proxy_connect_timeout 600;
    proxy_read_timeout 600;
    proxy_send_timeout 600;
    proxy_buffer_size 128k;
    proxy_buffers 16 256k;
    proxy_set_header   Host 'mgt.shangwenhe.com';
    proxy_http_version 1.1;
    proxy_set_header Connection "";
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

Lua脚本

Redis数据写入

local params = {
    sentinels = {
        { host = "10.00.000.89", port = 26379 },
        { host = "10.00.000.7", port = 26379 },
    },
    master_name = "galaxy-adam-redis",
    role = "master",
    connect_timeout = 300
}

-- 链接redis
local rc = require("redis.connector").new()
local redis, err = rc:connect_via_sentinel(params)
if not redis then
    ngx.say(string.format('{"status": "failed to connect", "error":"%s"}', err))
    return
end

-- get redis key
local document_uri = ngx.var.document_uri
local key = string.gsub(string.sub(string.lower(document_uri), string.len('/adam/') + 1 ), '/', ':')

-- 外理PUT,POST请求
if "PUT" == ngx.var.request_method or "POST" == ngx.var.request_method then
    local body = ngx.req.get_body_data()
    if nil == body then
        ngx.log(ngx.ERR, 'body:', body)
    end
    local setOk, setErr = redis:set(key, body)
    if not setOk then
        ngx.say(string.format('{"key": "%s", "status": "none", "error":"%s"}', key, setErr))
        return
    end
    -- 设置过期时间为3个月 60*60*24*30*3
    redis:expire(key, 7776000)
end

-- 删除请求 同时清除redis中的内容
if "DELETE" == ngx.var.request_method then
    local res = redis:get(key)
    if res then
        -- 设置过期时间为3个月 60*60*24*30*3
        redis:expire(key, 0)
        return
    end 
end

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

Redis数据读取

-- 非 get 请求拒绝外理
if "GET" ~= ngx.var.request_method then
    -- ngx.status = 403
    ngx.say('{"status": "403", "msg":"403 Forbidden"}')
    return
end

local rc = require("redis.connector").new()
-- 飞书通知
local notice = require("notice.feishu")
local params = {
    sentinels = { 
        { host = "10.00.000.89", port = 26379 },
        { host = "10.00.000.7", port = 26379 },
    },
    master_name = "galaxy-adam-redis",
  	-- 为了可以设置过期时间，这里选择了使用master
    role = "master",
    connect_timeout = 300
}

local redis, err = rc:connect_via_sentinel(params)
if not redis then
    ngx.say("failed to connect: ", err)
    return
end

local document_uri = ngx.var.document_uri
local key = string.gsub(string.sub(string.lower(document_uri), string.len('/galaxy/adam/') + 1 ), '/', ':')
local res, err = redis:get(key)
if not res then
    ngx.say(string.format('{"key": "%s", "status": "none"}', key))
    return
end
if res == ngx.null then
    local errMsg = string.format('{"key": "%s", "uri":"%s","status": "404"}', key, document_uri)
    ngx.say(errMsg)
  	-- 飞书通知
    notice.send(errMsg)
    return
end
-- 设置过期时间为3个月 60*60*24*30*3
redis:expire(key, 7776000)
ngx.say(res)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

Redis 方法

Redis哨兵

sentinel.lua 咨询哨兵，并请求流量转发目标

local ipairs, type = ipairs, type

local ngx_null = ngx.null

local tbl_insert = table.insert
local ok, tbl_new = pcall(require, "table.new")
if not ok then
    tbl_new = function (narr, nrec) return {} end -- luacheck: ignore 212
end


local _M = {
    _VERSION = '0.09'
}


function _M.get_master(sentinel, master_name)
    local res, err = sentinel:sentinel(
        "get-master-addr-by-name",
        master_name
    )
    if res and res ~= ngx_null and res[1] and res[2] then
        return { host = res[1], port = res[2] }
    else
        return nil, err
    end
end


function _M.get_slaves(sentinel, master_name)
    local res, err = sentinel:sentinel("slaves", master_name)

    if res and type(res) == "table" then
        local hosts = tbl_new(#res, 0)
        for _,slave in ipairs(res) do
            local num_recs = #slave
            local host = tbl_new(0, num_recs + 1)
            for i = 1, num_recs, 2 do
                host[slave[i]] = slave[i + 1]
            end

            local master_link_status_ok = host["master-link-status"] == "ok"
            local is_down = host["flags"] and (string.find(host["flags"],"s_down")
                or string.find(host["flags"],"disconnected"))
            if master_link_status_ok and not is_down then
                host.host = host.ip -- for parity with other functions
                tbl_insert(hosts, host)
            end
        end
        if hosts[1] ~= nil then
            return hosts
        else
            return nil, "no slaves available"
        end
    else
        return nil, err
    end
end
return _M

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

redis链接器

connector.lua

local ipairs, pcall, error, tostring, type, next, setmetatable, getmetatable =
    ipairs, pcall, error, tostring, type, next, setmetatable, getmetatable

local ngx_log = ngx.log
local ngx_ERR = ngx.ERR
local ngx_re_match = ngx.re.match

local str_find = string.find
local tbl_remove = table.remove
local tbl_sort = table.sort
local ok, tbl_new = pcall(require, "table.new")
if not ok then
    tbl_new = function (narr, nrec) return {} end -- luacheck: ignore 212
end

local redis = require("resty.redis")
redis.add_commands("sentinel")

local get_master = require("redis.sentinel").get_master
local get_slaves = require("redis.sentinel").get_slaves


-- A metatable which prevents undefined fields from being created / accessed
local fixed_field_metatable = {
    __index =
        function(t, k) -- luacheck: ignore 212
            error("field " .. tostring(k) .. " does not exist", 3)
        end,
    __newindex =
        function(t, k, v) -- luacheck: ignore 212
            error("attempt to create new field " .. tostring(k), 3)
        end,
}


-- Returns a new table, recursively copied from the one given, retaining
-- metatable assignment.
--
-- @param   table   table to be copied
-- @return  table
local function tbl_copy(orig)
    local orig_type = type(orig)
    local copy
    if orig_type == "table" then
        copy = {}
        for orig_key, orig_value in next, orig, nil do
            copy[tbl_copy(orig_key)] = tbl_copy(orig_value)
        end
        setmetatable(copy, tbl_copy(getmetatable(orig)))
    else -- number, string, boolean, etc
        copy = orig
    end
    return copy
end


-- Returns a new table, recursively copied from the combination of the given
-- table `t1`, with any missing fields copied from `defaults`.
--
-- If `defaults` is of type "fixed field" and `t1` contains a field name not
-- present in the defults, an error will be thrown.
--
-- @param   table   t1
-- @param   table   defaults
-- @return  table   a new table, recursively copied and merged
local function tbl_copy_merge_defaults(t1, defaults)
    if t1 == nil then t1 = {} end
    if defaults == nil then defaults = {} end
    if type(t1) == "table" and type(defaults) == "table" then
        local copy = {}
        for t1_key, t1_value in next, t1, nil do
            copy[tbl_copy(t1_key)] = tbl_copy_merge_defaults(
                t1_value, tbl_copy(defaults[t1_key])
            )
        end
        for defaults_key, defaults_value in next, defaults, nil do
            if t1[defaults_key] == nil then
                copy[tbl_copy(defaults_key)] = tbl_copy(defaults_value)
            end
        end
        return copy
    else
        return t1 -- not a table
    end
end


local DEFAULTS = setmetatable({
    connect_timeout = 100,
    read_timeout = 1000,
    connection_options = {}, -- pool, etc
    keepalive_timeout = 60000,
    keepalive_poolsize = 30,

    host = "127.0.0.1",
    port = 6379,
    path = "", -- /tmp/redis.sock
    password = "",
    sentinel_password = "",
    db = 0,
    url = "", -- DSN url

    master_name = "mymaster",
    role = "master",  -- master | slave
    sentinels = {},

    -- Redis proxies typically don't support full Redis capabilities
    connection_is_proxied = false,

    disabled_commands = {},

}, fixed_field_metatable)


-- This is the set of commands unsupported by Twemproxy
local default_disabled_commands = {
    "migrate", "move", "object", "randomkey", "rename", "renamenx", "scan",
    "bitop", "msetnx", "blpop", "brpop", "brpoplpush", "psubscribe", "publish",
    "punsubscribe", "subscribe", "unsubscribe", "discard", "exec", "multi",
    "unwatch", "watch", "script", "auth", "echo", "select", "bgrewriteaof",
    "bgsave", "client", "config", "dbsize", "debug", "flushall", "flushdb",
    "info", "lastsave", "monitor", "save", "shutdown", "slaveof", "slowlog",
    "sync", "time"
}


local _M = {
    _VERSION = '0.09',
}

local mt = { __index = _M }


local function parse_dsn(params)
    local url = params.url
    if url and url ~= "" then
        local url_pattern = [[^(?:(redis|sentinel)://)(?:([^@]*)@)?([^:/]+)(?::(\d+|[msa]+))/?(.*)$]]

        local m, err = ngx_re_match(url, url_pattern, "oj")
        if not m then
            return nil, "could not parse DSN: " .. tostring(err)
        end

        -- TODO: Support a 'protocol' for proxied Redis?
        local fields
        if m[1] == "redis" then
            fields = { "password", "host", "port", "db" }
        elseif m[1] == "sentinel" then
            fields = { "password", "master_name", "role", "db" }
        end

        -- password may not be present
        if #m < 5 then tbl_remove(fields, 1) end

        local roles = { m = "master", s = "slave" }

        local parsed_params = {}

        for i,v in ipairs(fields) do
            if v == "db" or v == "port" then
                parsed_params[v] = tonumber(m[i + 1])
            else
                parsed_params[v] = m[i + 1]
            end

            if v == "role" then
                parsed_params[v] = roles[parsed_params[v]]
            end
        end

        return tbl_copy_merge_defaults(params, parsed_params)
    end

    return params
end
_M.parse_dsn = parse_dsn


function _M.new(config)
    -- Fill out gaps in config with any dsn params
    if config and config.url then
        local err
        config, err = parse_dsn(config)
        if err then ngx_log(ngx_ERR, err) end
    end

    local ok, config = pcall(tbl_copy_merge_defaults, config, DEFAULTS)
    if not ok then
        return nil, config  -- err
    else
        -- In proxied Redis mode disable default commands
        if config.connection_is_proxied == true and
            not next(config.disabled_commands) then

            config.disabled_commands = default_disabled_commands
        end

        return setmetatable({
            config = setmetatable(config, fixed_field_metatable)
        }, mt)
    end
end


function _M.connect(self, params)
    if params and params.url then
        local err
        params, err = parse_dsn(params)
        if err then ngx_log(ngx_ERR, err) end
    end

    params = tbl_copy_merge_defaults(params, self.config)

    if #params.sentinels > 0 then
        return self:connect_via_sentinel(params)
    else
        return self:connect_to_host(params)
    end
end


local function sort_by_localhost(a, b)
    if a.host == "127.0.0.1" and b.host ~= "127.0.0.1" then
        return true
    else
        return false
    end
end


function _M.connect_via_sentinel(self, params)
    local sentinels = params.sentinels
    local master_name = params.master_name
    local role = params.role
    local db = params.db
    local password = params.password
    local sentinel_password = params.sentinel_password
    if sentinel_password then
        for _,host in ipairs(sentinels) do
            host.password = sentinel_password
        end
    end

    local sentnl, err, previous_errors = self:try_hosts(sentinels)
    if not sentnl then
        return nil, err, previous_errors
    end

    if role == "master" then
        local master, err = get_master(sentnl, master_name)
        if not master then
            return nil, err
        end

        sentnl:set_keepalive()

        master.db = db
        master.password = password

        local redis, err = self:connect_to_host(master)
        if not redis then
            return nil, err
        end

        return redis
    else
        -- We want a slave
        local slaves, err = get_slaves(sentnl, master_name)
        if not slaves then
            return nil, err
        end

        sentnl:set_keepalive()

        -- Put any slaves on 127.0.0.1 at the front
        tbl_sort(slaves, sort_by_localhost)

        if db or password then
            for _, slave in ipairs(slaves) do
                slave.db = db
                slave.password = password
            end
        end

        local slave, err, previous_errors = self:try_hosts(slaves)
        if not slave then
            return nil, err, previous_errors
        end

        return slave
    end
end


-- In case of errors, returns "nil, err, previous_errors" where err is
-- the last error received, and previous_errors is a table of the previous errors.
function _M.try_hosts(self, hosts)
    local errors = tbl_new(#hosts, 0)
    for i, host in ipairs(hosts) do
        
        local r, err = self:connect_to_host(host)
        if r and not err then
            return r, nil, errors
        else
            errors[i] = err
        end
    end

    return nil, "no hosts available", errors
end


function _M.connect_to_host(self, host)
    local r = redis.new()

    -- config options in 'host' should override the global defaults
    -- host contains keys that aren't in config
    -- this can break tbl_copy_merge_defaults, hence the mannual loop here
    local config = tbl_copy(self.config)
    for k, _ in pairs(config) do
        if host[k] then
            config[k] = host[k]
        end
    end

    r:set_timeout(config.connect_timeout)

    -- Stub out methods for disabled commands
    if next(config.disabled_commands) then
        for _, cmd in ipairs(config.disabled_commands) do
            r[cmd] = function()
                return nil, ("Command "..cmd.." is disabled")
            end
        end
    end

    local ok, err
    local path = host.path
    local opts = config.connection_options
    if path and path ~= "" then
        if opts then
            ok, err = r:connect(path, config.connection_options)
        else
            ok, err = r:connect(path)
        end
    else
        if opts then
            ok, err = r:connect(host.host, host.port, config.connection_options)
        else
            ok, err = r:connect(host.host, host.port)
        end
    end

    if not ok then
        return nil, err
    else
        r:set_timeout(config.read_timeout)

        local password = host.password
        if password and password ~= "" then
            local res, err = r:auth(password)
            if err then
                return res, err
            end
        end

        -- No support for DBs in proxied Redis.
        if config.connection_is_proxied ~= true and host.db ~= nil then
            local res, err = r:select(host.db)

            -- SELECT will fail if we are connected to sentinel:
            -- detect it and ignore error message it that's the case
            if err and str_find(err, "ERR unknown command") then
                local role = r:role()
                if role and role[1] == "sentinel" then
                    err = nil
                end
            end
            if err then
                return res, err
            end
        end
        return r, nil
    end
end


function _M.set_keepalive(self, redis)
    -- Restore connection to "NORMAL" before putting into keepalive pool,
    -- ignoring any errors.
    -- Proxied Redis does not support transactions.
    if self.config.connection_is_proxied ~= true then
        redis:discard()
    end

    local config = self.config
    return redis:set_keepalive(
        config.keepalive_timeout, config.keepalive_poolsize
    )
end


-- Deprecated: use config table in new() or connect() instead.
function _M.set_connect_timeout(self, timeout)
    self.config.connect_timeout = timeout
end


-- Deprecated: use config table in new() or connect() instead.
function _M.set_read_timeout(self, timeout)
    self.config.read_timeout = timeout
end


-- Deprecated: use config table in new() or connect() instead.
function _M.set_connection_options(self, options)
    self.config.connection_options = options
end


return setmetatable(_M, fixed_field_metatable)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422