利用redis来实现定时触发器的一些研究和疑问，希望春哥能解答一下。

1loto9s · 2015-09-14T15:37:35+00:00

十分感谢明白了 Date: Mon, 14 Sep 2015 10:46:35 +0800Subject: Re: [openresty] access_by_lua_file 执&...

利用redis来实现定时触发器的一些研究和疑问，希望春哥能解答一下。

1loto9s

我利用了redis的key过期事件来完成定时器的逻辑

比如：3小时后写一个邮件什么之类的需求

原理：redis2.8之后添加的 Redis Keyspace Notifications http://redis.io/topics/notifications

我知道有一些别的服务可以实现比如beanstalkd等。

只不过突然发现redis原来也可以实现，所以想刨根问底一下。

文章下面是一个微调一下就能运行的伪代码。

原理是每1分钟执行一下 accept 方法，判断redis订阅是否还健在。

如果健在就把lock key 的过期时间设置成90秒。

如果不在就通过 ngx.timer.at 或 ngx.thread.spawn不阻塞当前请求的子进程去接收消息。

receive方法处理消息，至于为什么有两个超时设置的原因是。

我发现如果只设置一个3S超时，while true 的 read_reply时会每隔设置的超时时间向nginx error log里写一条超时日志。

错误大概是：*14 lua tcp socket read timed out, client: 10.0.2.2, server:

这个原因我通过google groups讨论组搜索的结果是春哥对 redis 的 read_reply 方法有超时重置的机制。

所以我设置了一个比较大的时间来避免这个事情。

我的问题有3个：

1、ngx.timer.at和ngx.thread.spawn的区别具体是什么？

2、我这个业务逻辑哪种更合适？

3、这么写有什么问题么？

环境的相关说明：

ngx_openresty-1.7.7.2

CentOS-6.6-x86_64

redis 2.8.19

------------伪代码-------------------

-- 添加触发器
addTrigger(roleId, triggerId, interval)
-- 设置一个interval过期的key，并设置一个interval+10的参数过期key



local globals = ngx.shared.GLOBALS
local redis = require "resty.redis"
local redisInf = loadMod('redisInf')

local lockKey = 'trigger.lock'
local lockInterval = 90

function accept()
    if globals:safe_add(lockKey, true, lockInterval) then
        ngx.thread.spawn(function()
            receive()
        end)

        util:debug('aaaaaaa', 'accept')
    else
        util:debug('aaaaaaa', 'replace')

        globals:replace(lockKey, true, lockInterval)
    end
end

function receive()
    local client = redis:new()
    client:set_timeout(3000)
    client:connect(redisInf.HOST, redisInf.PORT)

    local result = client:psubscribe('__key*__:*')

    util:debug('xxxxxxxxx', result)

    if not result then
        return
    end

    client:set_timeout(1080000)

    while true do
        local res, err = client:read_reply()

        if not res and err ~= 'timeout' then
            util.file:append('/tmp/redis.log', 'err:' .. tostring(err))
            break
        end

        if res then  -- 返回数据样本 ["pmessage","__key*__:*","__keyevent@7__:expired","roleId.1848.tollgateId.410406"]
            local matchs = ngx.re.match(res[3], '^__keyevent@([0-9]+?)__:expired$')

            if matchs ~= ngx.null and matchs ~= nil then
                local dbIndex = matchs[1]

                matchs = ngx.re.match(res[4], '^roleId.([0-9]+?).triggerId.([0-9]+)$')

                if matchs ~= ngx.null and matchs ~= nil then
                    --- 这里重新调用loadMod的原因是在while true 里 懒加载（热更新 shared + version file）
                    local util = loadMod('shark.util')

                    util.file:append('/tmp/redis.log', 'roleId: ' .. matchs[1] .. ', triggerId: ' .. matchs[2] .. '\n')
                    
                    --- 调用不同的 trigger 触发器，如果是ngx.timer.at的话就不能用懒加载模式了
                    --- 只能用子请求，ngx.location.capture 来处理了
                end
            end
        end
    end

    client:close()
    globals:delete(lockKey)
end

membphis

> 1. ngx.timer.at和ngx.thread.spawn的区别具体是什么

关于这个问题，我可以简单回答一下。

执行空间的区别：

ngx.timer.at 执行的空间与当前会话没有任何关系，完全“野生”。

ngx.thread.spawn 当前回话内执行，占用当前会话资源。

可控性：

ngx.timer.at 完全不可控

ngx.thread.spawn 父线程可以完成子线程控制，如杀死、获取状态等。

代价：

ngx.timer.at 不占用当前会话任何资源，效率比较高

ngx.thread.spawn 对当前会话有影响，不能开启过多轻线程，效率相对较低

On Monday, September 14, 2015 at 3:37:35 PM UTC+8, Xiaohui Sun wrote:

我利用了redis的key过期事件来完成定时器的逻辑
比如：3小时后写一个邮件什么之类的需求
原理：redis2.8之后添加的 Redis Keyspace Notifications http://redis.io/topics/notifications

我知道有一些别的服务可以实现比如beanstalkd等。

只不过突然发现redis原来也可以实现，所以想刨根问底一下。

文章下面是一个微调一下就能运行的伪代码。

原理是每1分钟执行一下 accept 方法，判断redis订阅是否还健在。
如果健在就把lock key 的过期时间设置成90秒。
如果不在就通过 ngx.timer.at 或 ngx.thread.spawn不阻塞当前请求的子进程去接收消息。

receive方法处理消息，至于为什么有两个超时设置的原因是。
我发现如果只设置一个3S超时，while true 的 read_reply时会每隔设置的超时时间向nginx error log里写一条超时日志。
错误大概是：*14 lua tcp socket read timed out, client: 10.0.2.2, server:
这个原因我通过google groups讨论组搜索的结果是春哥对 redis 的 read_reply 方法有超时重置的机制。
所以我设置了一个比较大的时间来避免这个事情。

我的问题有3个：
1、ngx.timer.at和ngx.thread.spawn的区别具体是什么？
2、我这个业务逻辑哪种更合适？
3、这么写有什么问题么？

环境的相关说明：
ngx_openresty-1.7.7.2
CentOS-6.6-x86_64
redis 2.8.19

------------伪代码-------------------

-- 添加触发器 addTrigger(roleId, triggerId, interval) -- 设置一个interval过期的key，并设置一个interval+10的参数过期key local globals = ngx.shared.GLOBALS local redis = require "resty.redis" local redisInf = loadMod('redisInf') local lockKey = 'trigger.lock' local lockInterval = 90 function accept() if globals:safe_add(lockKey, true, lockInterval) then ngx.thread.spawn(function() receive() end) util:debug('aaaaaaa', 'accept') else util:debug('aaaaaaa', 'replace') globals:replace(lockKey, true, lockInterval) end end function receive() local client = redis:new() client:set_timeout(3000) client:connect(redisInf.HOST, redisInf.PORT) local result = client:psubscribe('__key*__:*') util:debug('xxxxxxxxx', result) if not result then return end client:set_timeout(1080000) while true do local res, err = client:read_reply() if not res and err ~= 'timeout' then util.file:append('/tmp/redis.log', 'err:' .. tostring(err)) break end if res then -- 返回数据样本 ["pmessage","__key*__:*","__keyevent@7__:expired","roleId.1848.tollgateId.410406"] local matchs = ngx.re.match(res[3], '^__keyevent@([0-9]+?)__:expired$') if matchs ~= ngx.null and matchs ~= nil then local dbIndex = matchs[1] matchs = ngx.re.match(res[4], '^roleId.([0-9]+?).triggerId.([0-9]+)$') if matchs ~= ngx.null and matchs ~= nil then --- 这里重新调用loadMod的原因是在while true 里懒加载（热更新 shared + version file） local util = loadMod('shark.util') util.file:append('/tmp/redis.log', 'roleId: ' .. matchs[1] .. ', triggerId: ' .. matchs[2] .. '\n') --- 调用不同的 trigger 触发器，如果是ngx.timer.at的话就不能用懒加载模式了 --- 只能用子请求，ngx.location.capture 来处理了 end end end end client:close() globals:delete(lockKey) end