开篇

Redis 中使用 lua 脚本编写的任务，可以将保证其执行任务的原子性，这里我整理了 3 个常用的lua脚本。

• 简单的限流
• 累计行为限制
• 延迟任务

脚本案例

• 案例 1：简单的限流

作用：N秒内最多执行多少次人数。

--获取KEY
local k = KEYS[1]
--次数
local mt = tonumber(ARGV[1] or '3')
-- 过期时间，单位是 秒
local expire = tonumber(ARGV[2] or '60')

--加1操作
local ct = redis.call('incr', k)
-- 第一次才进行过期时间设置的操作
if ct == 1 then
    redis.call('expire', k, expire)
end

if ct > mt then
    return 0
else
    return 1
end

大概会传入 3 个参数：

• 当前任务的 key，由业务决定
• 当前任务的最大次数 max-times，上述代码的mt
• 当前任务的执行判定周期，也就是expire表示的就是N秒内

简单说下 lua的一些格式

• local 声明局部变量
• 对于数组通过下标的形式来获取对应的值，下标从1开始
• 调用 redis是通过redis.call('cmd',...)来调用的
• if exp then do1 else do2 end 就是基础的 if 语句

• 案例 2：不同行为累计不超过 N次

作用：用户点击文章可以加1 分，点击重复文章无效，每天最多加 N分。

假如用传统的做法有很多，但是如果利用redis，可以利用set集合来保存其阅读过的文章 id，判定当集合的容量是否超过N

--获取KEY
local k = KEYS[1]
--获取ARGV内的参数并打印
local para = ARGV[1]
--限制值
local limit = tonumber(ARGV[2] or '5')
--获取长度
local len = tonumber(redis.call('scard', k))

-- 1 表示已经添加满了
if len >= limit then
    return 1
else
    -- 没有达到最大长度，那么添加元素
    local flag = redis.call('sadd', k, para)
    -- 重复添加返回的是 0
    if tonumber(flag) == 0 then
        return 0
    else
        -- 2 表示成功添加
        return 2
    end
end

这个案例用到的语法同案例 1，不再详述。

• 案例 3：延迟任务

前置知识 1

就是利用 redis 来进行延迟任务，我们先简单聊聊zset的特性。

• zset 带有一个score和value，其核心功能就是自动排序。结合业务来看value就可以是业务相关的json字符串，score可以是任务执行的时间。
• zset有一个api为zrangebyscore，核心功能就是根据传入的score范围来获取这个范围内的集合。需要注意的是这里不仅仅返回了 1 个元素，而是多个，主要也是因为考虑到定时任务多的时候，一个一个返回浪费性能。
• zset有一个api为zrem，表示的就是根据 key删除这个值

前置知识2

单纯的使用zset可以方便的添加任务，但是假如要删除的时候，就不方便了，因为 value是一个 json字符串，语义和保存上都不够优美。所以我们可以引入redis的数据结构hash来完善该延迟任务。

• zset中保存的 value是待执行任务的key
• hash的 key是上述的key，value才是业务的json

这样一来删除就非常方便了，只需要根据ket，删除hash再删除zset即可（lua 脚本保证了原子性，所以如何删除都可）

前置知识 3

有了上述 3 个 redis 相关的 API，我们再了解下lua的几个知识，就可以进行延迟任务的lua脚本编写了。

• lua 中可以通过#table 来获取数组或者table的长度
• lua中的不等于是~=，而不是!=
• lua中的不为空判定可以用~=false进行判定
• redis返回了多个值后，加入要在 lua中使用这些值，需要通过unpack函数处理

相关的数据结构转换详情，可以参看下表

http://redisdoc.com/script/eval.html

• 脚本

-- 传入的参数主要是当前时间戳 和 limit
-- 当前时间戳
local zSetKey = KEYS[1]
local hashKey = KEYS[2]
-- 当前的limit
local maxScore = ARGV[1]
local limit = ARGV[2]
-- body
local zList = redis.call('zrangebyscore', zSetKey, '0', maxScore, 'limit', '0', limit)
--命令代码
if zList ~= false and #zList ~= 0 then
    redis.call('zrem', zSetKey, unpack(zList))
    -- 移除 hash 表中的 key 的内容，方便移除定时任务
    local hashValues = redis.call('hmget', hashKey, unpack(zList))
    -- 找到了对应的值
    if hashValues ~= false and #hashValues ~= 0 then
        redis.call('hdel', hashKey, unpack(zList))
        return hashValues
    else
        return nil
    end
else
    return nil
end

因为自身对lua语法的不熟悉以及调试的不熟悉，上述脚本折腾了一天。

我们根据上述的前置知识来分析下代码，这是我从上述代码抽离出来的...

local zList = redis.call('zrangebyscore', zSetKey, '0', maxScore, 'limit', '0', limit)
...
redis.call('zrem', zSetKey, unpack(zList))

1. 通过函数zrangebyscore获取到了从0到maxScore的值，但是仅仅获取前limit个，获取的结果是zList
2. 然后加入要在 lua 脚本中，删除获取到的zList，传参的时候，就必须要通过unpack函数转化后才行

调试的时候为了方便，我才去了一些小办法：

-- 拼接命令并保存到 redis 中
local cmd = 'zrangebyscore ' .. zSetKey .. ' 0 ' .. maxScore .. ' limit ' .. '0 ' .. limit
redis.call('set', 'cmd', cmd)

还有就是如果返回的是hash这种复杂的数据结构，可以采取转换为json字符串的方法

--  cjson.encode
redis.call('set', 'result', cjson.encode(hashValues))
return cjson.encode(hashValues)

结语

对于某些原子性的redis操作通过lua脚本来完成是相当不错的选择。

但是也需要特别注意了，因为redis的单线程特性，假如lua脚本太复杂，执行频率又高，这样会大大降低Redis的QPS。

感谢大家阅读，欢迎提出问题共同讨论！