客服电话
APP下载
迪恩网络APP
随时随地掌握行业动态
官方微信
扫描二维码
关注迪恩网络微信公众号
|
引言 前段时间组内有个投票的产品，上线前考虑欠缺，导致被刷票严重。后来，通过研究，发现可以通过 redis lua 脚本实现限流，这里将 redis lua 脚本相关的知识分享出来，讲的不到位的地方还望斧正。 redis lua 脚本相关命令 这一小节的内容是基本命令，可粗略阅读后跳过，等使用的时候再回来查询 redis 自 2.6.0 加入了 lua 脚本相关的命令，
生产环境中，推荐使用 Redis 中 lua 脚本的书写和调试 redis lua 脚本是对其现有命令的扩充，单个命令不能完成、需要多个命令，但又要保证原子性的动作可以用脚本来实现。脚本中的逻辑一般比较简单，不要加入太复杂的东西，因为 redis 是单线程的，当脚本执行的时候，其他命令、脚本需要等待直到当前脚本执行完成。因此，对 lua 的语法也不需完全了解，了解基本的使用就足够了，这里对 lua 语法不做过多介绍，会穿插到脚本示例里面。 一个秒杀抢购示例 假设有一个秒杀活动，商品库存 100，每个用户 uid 只能抢购一次。设计抢购流程如下：
local goodsSurplus
local flag
-- 判断用户是否已抢过
local buyMembersKey = tostring(KEYS[1])
local memberUid = tonumber(ARGV[1])
local goodsSurplusKey = tostring(KEYS[2])
local hasBuy = redis.call("sIsMember", buyMembersKey, memberUid)
-- 已经抢购过，返回0
if hasBuy ~= 0 then
return 0
end
-- 准备抢购
goodsSurplus = redis.call("GET", goodsSurplusKey)
if goodsSurplus == false then
return 0
end
-- 没有剩余可抢购物品
goodsSurplus = tonumber(goodsSurplus)
if goodsSurplus <= 0 then
return 0
end
flag = redis.call("SADD", buyMembersKey, memberUid)
flag = redis.call("DECR", goodsSurplusKey)
return 1
即使不了解 lua，相信你也可以将上面的脚本看个一二，其中 另外 redis lua 脚本中用到 lua table 的地方还比较多，这里要注意，lua 脚本中的 table 下标是从 1 开始的，比如 对于主要使用 PHP 这种弱类型语言开发同学来说，一定要注意变量的类型，不同类型比较的时候可能会出现类似 在调试之前呢，我们先看看效果，将上面的代码保存到 lua 文件中 ➜ ~ redis-cli set goodsSurplus 5 OK ➜ ~ redis-cli --eval /path/to/buy.lua hadBuyUids goodsSurplus , 5824742984 (integer) 1 ➜ ~ redis-cli --eval /path/to/buy.lua hadBuyUids goodsSurplus , 5824742984 (integer) 0 ➜ ~ redis-cli --eval /path/to/buy.lua hadBuyUids goodsSurplus , 5824742983 (integer) 1 ➜ ~ redis-cli --eval /path/to/buy.lua hadBuyUids goodsSurplus , 5824742982 (integer) 1 ➜ ~ redis-cli --eval /path/to/buy.lua hadBuyUids goodsSurplus , 5824742981 (integer) 1 ➜ ~ redis-cli --eval /path/to/buy.lua hadBuyUids goodsSurplus , 5824742980 (integer) -1 ➜ ~ redis-cli --eval /path/to/buy.lua hadBuyUids goodsSurplus , 58247 (integer) -1 在命令行运行脚本的时候，脚本后面传入的是参数，通过 debug 调试 上一小节，我们写了很长一段 redis lua 脚本，怎么调试呢，有没有像 GDB 那样的调试工具呢，答案是肯定的。redis 从 v3.2.0 开始支持 lua debugger，可以加断点、print 变量信息、展示正在执行的代码......我们结合上一小节的脚本，来详细说说 redis 中 lua 脚本的调试。 如何进入调试模式 执行 调试命令详解 这一小节的内容是调试时候的详细命令，可以粗略阅读后跳过，等使用的时候再回来查询 帮助信息 [h]elp 调试模式下，输入 流程相关 [s]tep 、 [n]ext 、 [c]continue 执行当前行代码，并停留在下一行，如下所示
* Stopped at 4, stop reason = step over
-> 4 local buyMembersKey = tostring(KEYS[1])
lua debugger> n
* Stopped at 5, stop reason = step over
-> 5 local memberUid = tonumber(ARGV[1])
lua debugger> n
* Stopped at 6, stop reason = step over
-> 6 local goodsSurplusKey = tostring(KEYS[2])
lua debugger> s
* Stopped at 7, stop reason = step over
-> 7 local hasBuy = redis.call("sIsMember", buyMembersKey, memberUid)
展示相关 [l]list 、 [l]list [line] 、 [l]list [line] [ctx] 、 [w]hole 展示当前行附近的代码， 打印相关 [p]rint 、 [p]rint <var> 打印当前所有局部变量， lua debugger> print <value> goodsSurplus = nil <value> flag = nil <value> buyMembersKey = "hadBuyUids" <value> memberUid = 58247 lua debugger> print buyMembersKey <value> "hadBuyUids" 断点相关 [b]reak 、 [b]reak <line> 、 [b]reak -<line> 、 [b]reak 0 展示断点、像指定行添加断点、删除指定行的断点、删除所有断点 其他命令 [r]edis <cmd> 、 [m]axlen [len] 、 [a]bort 、 [e]eval <code> 、 [t]race
详细说下
local myTable = {}
local count = 0
while count < 1000 do
myTable[count] = count
count = count + 1
end
return 1
在最后一行打印断点，执行 详细说下 local function func1(num) num = num + 1 return num end local function func2(num) num = func1(num) num = num + 1 return num end func2(123) 执行 lua debugger> t In func1: ->#3 return num From func2: 7 num = func1(num) From top level: 12 func2(123) 请求限流 至此，算是对 redis lua 脚本有了基本的认识，基本语法、调试也做了了解，接下来就实现一个请求限流器。流程和代码如下：
--[[
传入参数：
业务标识
ip
限制时间
限制时间内的访问次数
]]--
local busIdentify = tostring(KEYS[1])
local ip = tostring(KEYS[2])
local expireSeconds = tonumber(ARGV[1])
local limitTimes = tonumber(ARGV[2])
local identify = busIdentify .. "_" .. ip
local times = redis.call("GET", identify)
--[[
获取已经记录的时间
获取到继续判断是否超过限制
超过限制返回0
否则加1，返回1
]]--
if times ~= false then
times = tonumber(times)
if times >= limitTimes then
return 0
else
redis.call("INCR", identify)
return 1
end
end
-- 不存在的话，设置为1并设置过期时间
local flag = redis.call("SETEX", identify, expireSeconds, 1)
return 1
将上面的 lua 脚本保存到 好了，至此，一个请求限流功能就完成了，连续执行三次之后上面的程序会返回 0，过 10 秒钟在执行，又可以返回 1，这样便达到了限流的目的。 有同学可能会说了，这个请求限流功能还有值得优化的地方，如果连续的两个计数周期，第一个周期的最后请求 3 次，接着马上到第二个周期了，又可以请求了，这个地方如何优化呢，我们接着往下看。 请求限流优化 上面的计数器法简单粗暴，但是存在临界点的问题。为了解决这个问题，引入类似滑动窗口的概念，让统计次数的周期是连续的，可以很好的解决临界点的问题，滑动窗口原理如下图所示： 建立一个 redis list 结构，其长度等价于访问次数，每次请求时，判断 list 结构长度是否超过限制次数，未超过的话，直接加到队首返回成功，否则，判断队尾一条数据是否已经超过限制时间，未超过直接返回失败，超过删除队尾元素，将此次请求时间插入队首，返回成功。
local busIdentify = tostring(KEYS[1])
local ip = tostring(KEYS[2])
local expireSeconds = tonumber(ARGV[1])
local limitTimes = tonumber(ARGV[2])
-- 传入额外参数，请求时间戳
local timestamp = tonumber(ARGV[3])
local lastTimestamp
local identify = busIdentify .. "_" .. ip
local times = redis.call("LLEN", identify)
if times < limitTimes then
redis.call("RPUSH", identify, timestamp)
return 1
end
lastTimestamp = redis.call("LRANGE", identify, 0, 0)
lastTimestamp = tonumber(lastTimestamp[1])
if lastTimestamp + expireSeconds >= timestamp then
return 0
end
redis.call("LPOP", identify)
redis.call("RPUSH", identify, timestamp)
return 1
上面的 lua 脚本保存到 最开始，我想着把时间戳计算 另外，redis 从版本 5 开始，默认支持script effects replication，不需要在第一行调用开启函数了。如果是耗时计算，这样当然很好，同步、恢复的时候只需要计算一次后边就不用计算了，但是如果是一个循环生成的数据，可能在同步的时候会浪费更多的带宽，没有脚本来的更直接，但这种情况应该比较少。 至此，脚本优化完成了，但我又想到一个问题，我们的环境是单机环境，如果是分布式环境的话，脚本怎么执行、何处理呢，接下来一节，我们来讨论下这个问题。 集群环境中 lua 处理 redis 集群中，会将键分配的不同的槽位上，然后分配到对应的机器上，当操作的键为一个的时候，自然没问题，但如果操作的键为多个的时候，集群如何知道这个操作落到那个机器呢？比如简单的 首先用 docker 启动一个 redis 集群， 我们从任意一个节点进入集群，比如 执行
同样，还是上面的 lua 脚本，我们加上集群端口号，执行 针对这个问题，redis官方为我们提供了 同理，我们对传入脚本的键名做 hash tag 处理就可以了，这里要注意不仅传入键名要有相同的 hash tag，里面实际操作的 key 也要有相同的 hash tag，不然会报错
redis-cli -c -p 7000 --eval /tmp/limit_fun.lua limit_vgroup{yes} 192.168.1.19{yes} , 10 3 1548660999
如果我们在脚本里面加上 另外，这里有个 hash tag 规则： 键中包含 所以，键 使用 golang 连接使用 redis 这里我们使用 golang 实例展示下，通过
package main
import (
"github.com/go-redis/redis"
"fmt"
)
func createScript() *redis.Script {
script := redis.NewScript(`
local busIdentify = tostring(KEYS[1])
local ip = tostring(KEYS[2])
local expireSeconds = tonumber(ARGV[1])
local limitTimes = tonumber(ARGV[2])
-- 传入额外参数，请求时间戳
local timestamp = tonumber(ARGV[3])
local lastTimestamp
local identify = busIdentify .. "_" .. ip
local times = redis.call("LLEN", identify)
if times < limitTimes then
redis.call("RPUSH", identify, timestamp)
return 1
end
lastTimestamp = redis.call("LRANGE", identify, 0, 0)
lastTimestamp = tonumber(lastTimestamp[1])
if lastTimestamp + expireSeconds >= timestamp then
return 0
end
redis.call("LPOP", identify)
redis.call("RPUSH", identify, timestamp)
return 1
`)
return script
}
func scriptCacheToCluster(c *redis.ClusterClient) string {
script := createScript()
var ret string
c.ForEachMaster(func(m *redis.Client) error {
if result, err := script.Load(m).Result(); err != nil {
panic("缓存脚本到主节点失败")
} else {
ret = result
}
return nil
})
return ret
}
func main() {
redisdb := redis.NewClusterClient(&redis.ClusterOptions{
Addrs: []string{
":7000",
":7001",
":7002",
":7003",
":7004",
":7005",
},
})
// 将脚本缓存到所有节点，执行一次拿到结果即可
sha := scriptCacheToCluster(redisdb)
// 执行缓存脚本
ret := redisdb.EvalSha(sha, []string{
"limit_vgroup{yes}",
"192.168.1.19{yes}",
}, 10, 3,1548660999)
if result, err := ret.Result(); err != nil {
fmt.Println("发生异常，返回值：", err.Error())
} else {
fmt.Println("返回值：", result)
}
// 示例错误情况，sha 值不存在
ret1 := redisdb.EvalSha(sha + "error", []string{
"limit_vgroup{yes}",
"192.168.1.19{yes}",
}, 10, 3,1548660999)
if result, err := ret1.Result(); err != nil {
fmt.Println("发生异常，返回值：", err.Error())
} else {
fmt.Println("返回值：", result)
}
}
执行上面的代码，返回值如下：
好了，目前为止，相信你对 redis lua 脚本已经有了很好的了解，可以实现一些自己想要的功能了，感谢大家的阅读。希望对大家的学习有所帮助，也希望大家多多支持极客世界。 |
请发表评论