- 减少网络开销。可以将多个请求通过脚本的形式一次发送，减少网络时延。
- 原子操作。Redis会将整个脚本作为一个整体执行，中间不会被其他请求插入。因此在脚本运行过程中无需担心会出现竞态条件，无需使用事务。
- 复用。客户端发送的脚本会永久存在redis中，这样其他客户端可以复用这一脚本，而不需要使用代码完成相同的逻辑。
//返回redis中KEYS[1] 的值
retrun redis.call('GET',KEYS[1])
redis中的一些lua命令
- EVAL
- 命令格式：EVAL script numkeys key [key …] arg [arg …]
- script参数是一段 Lua 脚本程序
- numkeys指定后续参数有几个key，即：key [key …]中key的个数。如没有key，则为0
- key [key …] 从 EVAL 的第三个参数开始算起，表示在脚本中所用到的那些 Redis 键(key)。在Lua脚本中通过KEYS[1], KEYS[2]获取。
- arg [arg …] 附加参数。在Lua脚本中通过ARGV[1],ARGV[2]获取。
//相当于SETEX key1 60 10 EVAL "redis.call('SET',KEYS[1],ARGV[1]);redis.call('EXPIRE',KEYS[1],ARGV[2]);return 1;" 1 key1 10 60 - SCRIPT LOAD
- 命令格式 SCRIPT LOAD script
- SCRIPT LOAD 将脚本 script 添加到Redis服务器的脚本缓存中，并不立即执行这个脚本，而是会立即对输入的脚本进行求值。并返回给定脚本的 SHA1 校验和。如果给定的脚本已经在缓存里面了，那么不执行任何操作。
- EVALSHA
- 命令格式：EVALSHA sha1 numkeys key [key …] arg [arg …]
- 在脚本被加入到缓存之后，在任何客户端通过EVALSHA命令，可以使用脚本的 SHA1 校验和来调用这个脚本。脚本可以在缓存中保留无限长的时间，直到执行SCRIPT FLUSH为止。
- SCRIPT EXISTS
- 命令格式：SCRIPT EXISTS sha1 [sha1 …]
- 给定一个或多个脚本的 SHA1 校验和，返回一个包含 0 和 1 的列表，表示校验和所指定的脚本是否已经被保存在缓存当中
- SCRIPT FLUSH
- 命令格式：SCRIPT FLUSH
- 清除Redis服务端所有 Lua 脚本缓存
- SCRIPT KILL
- 命令格式：SCRIPT KILL
- 杀死当前正在运行的 Lua 脚本，当且仅当这个脚本没有执行过任何写操作时，这个命令才生效。 这个命令主要用于终止运行时间过长的脚本，比如一个因为 BUG 而发生无限 loop 的脚本，诸如此类。
- 假如当前正在运行的脚本已经执行过写操作，那么即使执行SCRIPT KILL，也无法将它杀死，因为这是违反 Lua 脚本的原子性执行原则的。在这种情况下，唯一可行的办法是使用SHUTDOWN NOSAVE命令，通过停止整个 Redis 进程来停止脚本的运行，并防止不完整(half-written)的信息被写入数据库中。
127.0.0.1:6379> SCRIPT LOAD "redis.call('SET',KEYS[1],ARGV[1]);redis.call('EXPIRE',KEYS[1],ARGV[2]);return 1;"
"6cc501292668ceef3dd487b3e4e889dc08d07587"
127.0.0.1:6379> EVALSHA "6cc501292668ceef3dd487b3e4e889dc08d07587" 1 key1 10 60
(integer) 1
127.0.0.1:6379> get key1
"10"
127.0.0.1:6379> ttl key1
(integer) 42
127.0.0.1:6379> SCRIPT EXISTS "6cc501292668ceef3dd487b3e4e889dc08d07587"
1) (integer) 1
127.0.0.1:6379> SCRIPT FLUSH
OK
127.0.0.1:6379> SCRIPT EXISTS "6cc501292668ceef3dd487b3e4e889dc08d07587"
1) (integer) 0
redis直接执行lua脚本文件
- script.lua
redis.call('SET',KEYS[1],ARGV[1]); redis.call('EXPIRE',KEYS[1],ARGV[2]); return 1; - 终端中执行命令： redis-cli -a 密码 --eval Lua脚本路径 key [key …] , arg [arg …]
redis-cli -a 123456 --eval ./script.lua key1 , 10 , 60
这里我们使用redigo，下面是几个简单的函数
Dial- func Dial(network, address string, options …DialOption) (Conn, error)
//dial创建单条连接
//example
c, err := redis.Dial("tcp", ":6379")
if err != nil {
// handle error
}
defer c.Close()
//建一个连接池
pool := &redis.Pool{
// 最大空闲链接
MaxIdle: 10,
// 最大激活链接
MaxActive: 10,
// 最大空闲链接等待时间
IdleTimeout: 10 * time.Second,
Dial: func() (redis.Conn, error) {
c, err := redis.Dial("tcp", "127.0.0.1:6379")
if err != nil {
fmt.Println("Connect to redis error", err)
}
return c, nil
},
}
c:=pool.Get()
Do or Send
//Do让redis客户端执行命令
c.Do('GET',key)
//Send写入到客户端输出缓冲区，再调用Flush执行缓冲区的命令
c.Send('SET',key,val)
c.Send('GET',key)
c.Flush()
NewScript 加载脚本
- func NewScript(keyCount int, src string) *Script
- func (s *Script) Do(c Conn, keysAndArgs …interface{}) (interface{}, error)
lua := redis.NewScript(2, string(script)) lua.Do(c,key1,key2)run a demo
好已经学废了来写个生产消费模型来跑一下
--参数（用户，商品）
--消费成功的用户一秒内不能再消费
--返回值 0-库存不足 1-成功 2-用户冷却中
if redis.call("EXISTS",KEYS[1]) == 1 then
return 2
else
if redis.call("EXISTS",KEYS[2])==1 and tonumber(redis.call("GET",KEYS[2]))>0 then
--商品数量-1
redis.call("DECR",KEYS[2])
--用户拥有商品数量+1
redis.call("HINCRBY",KEYS[1].."buy",KEYS[2],1)
--set一个1s的key表示冷却时间
redis.call("SET",KEYS[1],1)
redis.call("EXPIRE",KEYS[1],1)
return 1
else
return 0
end
end
main.go
func producer(c redis.Conn, product string) {
defer c.Close()
for {
c.Do("INCR", product)
fmt.Println(product + "+1")
time.Sleep(time.Second)
}
}
func consumer(c redis.Conn, user string, product string) {
defer c.Close()
script, err := ioutil.ReadFile("./consumer.lua")
if err != nil {
fmt.Println("Script read error", err)
return
}
lua := redis.NewScript(2, string(script))
for {
res, err := lua.Do(c, user, product)
if err != nil {
fmt.Println(err)
return
}
if r, _ := res.(int64); r == 1 {
fmt.Println(user + " buy " + product)
} else if r == 0 {
fmt.Println("Unluckly " + user + "," + product + " is sold out!")
} else {
fmt.Print(user + "/")
}
time.Sleep(time.Millisecond * time.Duration(rand.Intn(10)*100))
}
}
func main() {
pool := &redis.Pool{
// 最大空闲链接
MaxIdle: 10,
// 最大激活链接
MaxActive: 10,
// 最大空闲链接等待时间
IdleTimeout: 10 * time.Second,
Dial: func() (redis.Conn, error) {
c, err := redis.Dial("tcp", "127.0.0.1:6379")
if err != nil {
fmt.Println("Connect to redis error", err)
}
return c, nil
},
}
//每次开始先清空一下redis
c := pool.Get()
c.Do("FLUSHALL")
c.Close()
go producer(pool.Get(), "apple")
go producer(pool.Get(), "pineapple")
go consumer(pool.Get(), "001", "apple")
go consumer(pool.Get(), "002", "apple")
go consumer(pool.Get(), "002", "pineapple")
time.Sleep(time.Second * 10)
}
result
apple+1 pineapple+1 001 buy apple Unluckly 002,apple is sold out! 002 buy pineapple 001/002/002/002/pineapple+1 apple+1 001 buy apple 001/002 buy pineapple 002/002/001/002/002/001/apple+1 pineapple+1 001 buy apple 001/Unluckly 002,apple is sold out! 002 buy pineapple 001/pineapple+1 apple+1 002/002/001 buy apple 002 buy pineapple apple+1 pineapple+1 002/001/002/002/002 buy apple Unluckly 001,apple is sold out! Unluckly 001,apple is sold out! pineapple+1 apple+1 002/002/002/001 buy apple 002 buy pineapple pineapple+1 apple+1 002/002/001/002/002/002/002 buy pineapple 001 buy apple pineapple+1 apple+1 002/002/001/001/002 buy pineapple pineapple+1 apple+1 002/002/002/001 buy apple 002/002/apple+1 pineapple+1 002 buy pineapple 002/001/001 buy applecheck redis
没有超卖，看起来没什么问题
127.0.0.1:6379> get apple "1" 127.0.0.1:6379> get pineapple "2" 127.0.0.1:6379> hmget 001buy apple pineapple 1) "8" 2) (nil) 127.0.0.1:6379> hmget 002buy apple pineapple 1) "1" 2) "8"redis+lua限流 计数器模式
-- 获取调用脚本时传入的第一个key值（用作限流的 key）
local key = KEYS[1]
-- 获取调用脚本时传入的第一个参数值（限流大小）
local limit = tonumber(ARGV[1])
-- 获取当前流量大小
local curentLimit = tonumber(redis.call('get', key) or "0")
-- 是否超出限流
if curentLimit + 1 > limit then
-- 返回(拒绝)
return 0
else
-- 没有超出 value + 1
redis.call("INCRBY", key, 1)
-- 设置过期时间
redis.call("EXPIRE", key, 2)
-- 返回(放行)
return 1
end
计数器模式缺点：粒度不够细的情况下, 会出现在同一个窗口时间内出现双倍请求数，例如设置1000/s的限流，如果0.9秒时有900个请求，1.1秒时有800个，那么0.5s到1.5s这一秒的窗口时间就有1700个请求。
令牌桶模式每次访问从桶中取令牌,有令牌才允许通过,一段时间后放回,
根据桶中令牌数量和令牌恢复的时间来控制流量
具体令牌验证在Controller实现
--last_mill_second 最后加入时间(毫秒)
--curr_permits 当前可用的令牌
--max_burst 令牌桶最大值
--rate 每秒生成几个令牌
--app 应用
--ARGV[1] 当前时间
--ARGV[2] 需取走令牌数量
if KEYS[1] == nil then
return -1
end
if(redis.pcall("EXISTS",KEYS[1])==0)
then
--初始化
redis.pcall("HMSET",KEYS[1],
"last_mill_second",ARGV[1],
"curr_permits",ARGV[2],
"max_burst",ARGV[3],
"rate",ARGV[4],
"app",ARGV[5])
end
local ratelimit_info=redis.pcall("HMGET",KEYS[1],"last_mill_second","curr_permits","max_burst","rate","app")
local last_mill_second=ratelimit_info[1]
local curr_permits=tonumber(ratelimit_info[2])
local max_burst=tonumber(ratelimit_info[3])
local rate=tonumber(ratelimit_info[4])
local app=tostring(ratelimit_info[5])
local local_curr_permits=max_burst;
if(type(last_mill_second) ~='boolean' and last_mill_second ~=nil) then
--计算可以加入的最大令牌
local reverse_permits=math.floor((ARGV[1]-last_mill_second)/1000)*rate
if(reverse_permits>0) then
--如果可以加入，则把当前时间作为最后加入令牌时间
redis.pcall("HMSET",KEYS[1],"last_mill_second",ARGV[1])
end
--计算加入令牌后最大值
--防止节点转发出现时间早的后出现,保证传输顺序
reverse_permits=math.max(reverse_permits,0);
local expect_curr_permits=reverse_permits+curr_permits
--超出最大容量则令牌溢出 丢弃
local_curr_permits=math.min(expect_curr_permits,max_burst);
else
redis.pcall("HMSET",KEYS[1],"last_mill_second",ARGV[1])
end
local result=-1
if(local_curr_permits-ARGV[2]>0) then
result=1
--如果可以获取令牌，则去掉此时需要拿走的令牌
redis.pcall("HMSET",KEYS[1],"curr_permits",local_curr_permits-ARGV[2])
else
--如果不行，则把当前最新的令牌数写入
redis.pcall("HMSET",KEYS[1],"curr_permits",local_curr_permits)
end
return result
ip限流
--KEYS[1] 限流ip
local key = "rate.limit:" .. KEYS[1]
local limit = tonumber(ARGV[1])
local expire_time = ARGV[2]
local is_exists = redis.call("EXISTS", key)
if is_exists == 1 then
if redis.call("INCR", key) > limit then
return 0
else
return 1
end
else
redis.call("SET", key, 1)
redis.call("EXPIRE", key, expire_time)
return 1
end
