redis在使用lua脚本以及实现redis分布式锁
Posted 牧竹子
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了redis在使用lua脚本以及实现redis分布式锁相关的知识，希望对你有一定的参考价值。
目录
背景介绍
redis内置lua执行命令
环境准备
redis运行lua脚本
访问次数限制
lua脚本实现redis分布式锁
java调用脚本操作redis
Lua+Redis 断点调试环境搭建
背景介绍
Redis在2.6推出了脚本功能，允许开发者使用Lua语言编写脚本传到Redis中执行。使用脚本的好处如下:
1.减少网络开销：本来5次网络请求的操作，可以用一个请求完成，原先5次请求的逻辑放在redis服务器上完成。使用脚本，减少了网络往返时延。
2.原子操作：Redis会将整个脚本作为一个整体执行，中间不会被其他命令插入(java等客户端则会执行多次命令完成一个业务，违反了原子性操作)。
3.复用：客户端发送的脚本会永久存储在Redis中，意味着其他客户端可以复用这一脚本而不需要使用代码完成同样的逻辑。
redis在服务器端内置lua解释器（版本2.6以上）
redis-cli提供了EVAL与EVALSHA命令执行Lua脚本:
redis内置lua执行命令
EVAL命令语法
EVAL script numkeys key [key …] arg [arg …]
EVAL —lua程序的运行环境上下文
script —lua脚本
numkeys —参数的个数(key的个数)
key —redis键 访问下标从1开始,例如:KEYS[1]
arg —redis键的附加参数
EVALSHA 命令语法
EVALSHA SHA1 numkeys key [key …] arg [arg …]
EVALSHA命令允许通过脚本的SHA1来执行(节省带宽)
Redis在执行EVAL/SCRIPT LOAD后会计算脚本SHA1缓存, EVALSHA根据SHA1取出缓存脚本执行.
Redis中管理Lua脚本
script load script 将Lua脚本加载到Redis内存中(如果redis重启则会丢失)
script exists sh1 [sha1 …] 判断sha1脚本是否在内存中
script flush 清空Redis内存中所有的Lua脚本
script kill 杀死正在执行的Lua脚本。（如果此时Lua脚本正在执行写操作，那么script kill将不会生效）
Redis提供了一个lua-time-limit参数，默认5秒，它是Lua脚本的超时时间,如果Lua脚本超时，其他执行正常命令的客户端会收到“Busy Redis is busy running a script”错误，但是不会停止脚本运行，此时可以使用script kill 杀死正在执行的Lua脚本。
lua函数
主要有两个函数来执行redis命令
redis.call() – 出错时返回具体错误信息,并且终止脚本执行
redis.pcall() –出错时返回lua table的包装错误,但不引发错误
使用流程如下：
1.编写脚本
2.脚本提交到REDIS并获取SHA
3.使用SHA调用redis脚本
环境准备
win10中的bash做的实验, 可以下载
luaforwindows
https://github.com/rjpcomputing/luaforwindows/releases
redis-windows
https://github.com/ServiceStack/redis-windows
redis运行lua脚本
EVAL 直接运行脚本
127.0.0.1:6379> EVAL "return {KEYS[1],KEYS[2],ARGV[1],ARGV[2]}" 2 key1 key2 first second
1) "key1"
2) "key2"
3) "first"
4) "second"EVALSHA使用
需要SCRIPT LOAD和EVALSHA配合使用
1.SCRIPT LOAD加载到内存，返回SHA签名
2.EVALSHA使用已经存在的签名
这样只用加载一次，便可重复使用已经加载的签名脚本，可以多次使用，避免长脚本输入
127.0.0.1:6379> SCRIPT LOAD "return {KEYS[1],KEYS[2],ARGV[1],ARGV[2]}"
"a42059b356c875f0717db19a51f6aaca9ae659ea"
127.0.0.1:6379> EVALSHA "a42059b356c875f0717db19a51f6aaca9ae659ea" 2 key1 key2 first second
1) "key1"
2) "key2"
3) "first"
4) "second"在redis下使用脚本文件执行
在路径下创建脚本文件，这里直接在redis/bin下创建，方便使用，其他目录可以使用全路径
set.lua
--[[ set.lua, redis的set命令使用
redis: set key val
--]]
local key = KEYS[1]
local val = ARGV[1]
return redis.call('set', key, val)get.lua
--[[ get.lua, redis的get命令使用
redis: get key
--]]
local key = KEYS[1]
local val = redis.call("GET", key);
return val;保存两个文件到redis/bin目录下,执行如下命令
设置k-v值
redis-cli --eval set.lua foo , bar通过redis-cli查看值
127.0.0.1:6379> get foo
"bar"获取k值
redis-cli --eval get.lua foo
"bar"可以看到， 可以用lua脚本操作redis数据。
注意： redis-cli --eval set.lua foo , bar， foo和bar之间的逗号左右都有空格分隔,否则会当做一个字符串
通常做法是
1.脚本文件保存到一个路径下或者数据库中，/mnt/redis/lua/set.lua
2.SCRIPT LOAD 加载脚本文件内容，返回SHA签名保存到应对的值K-V值，(set,SHA)
3.获取对应脚本名称的SHA签名，如果存在则执行，否则进行第二部操作
访问次数限制
ratelimiting.lua
local times = redis.call('incr',KEYS[1])
if times == 1 then
redis.call('expire',KEYS[1], ARGV[1])
end
if times > tonumber(ARGV[2]) then
return 0
end
return 1运行脚本
redis-cli --eval ratelimiting.lua rate.limitingl:127.0.0.1 , 10 3rate.limitingl:127.0.0.1是前缀+ip组成的KEY,用KEYS[1]获取，
”,”后面的10和3是参数，在脚本中能够使用ARGV[1]和ARGV[2]获得
命令的作用是将访问频率限制为每10秒最多3次，所以在终端中不断的运行此命令会发现当访问频率在10秒内小于或等于3次时返回1，否则返回0。
lua脚本实现redis分布式锁
setnx 如果key不存在则添加值并返回1，如果已经存在key则返回0
加锁
使用业务setnx(key,业务流水号)当加锁成功返回1时设置过期时间，避免业务异常没有解锁时防止死锁
当同一业务再次申请锁时，如果随机值相同 则认为是重试,则直接设置过期时长；如果随机值不同则直接返回0，获取锁失败
解锁
业务完成直接del(key)完成
以上方案是很多客户端实现的方式，建立和释放锁，并保证绝对的安全，是这个锁的设计比较棘手的地方。有两个潜在的陷阱：
1.应用程序通过网络和redis交互，这意味着从应用程序发出命令到redis结果返回之间会有延迟。这段时间内，redis可能正在运行其他的命令，而redis内数据的状态可能不是你的程序所期待的。如果保证程序中获取锁的线程和其他线程不发生冲突？
2.如果程序在获取锁后突然crash，而无法释放它？这个锁会一直存在而导致程序进入“死锁”
对于第一个问题，除了pile批量一次执行，目前只有lua脚本是在同一个线程中一次执行完的。
第二个问题，如果在获取锁之后，设置expire之前发生了异常，那么这个key-v永远都不会过期，即便是lua脚本也是一样会发生这样的情况（通常是设置过期时间这个参数设置的不是数字类型，虽然这种情况不太可能发生），但仍然比客户端多条命令执行来的更加简短
lock.lua
-- Set a lock
-- 如果获取锁成功，则返回 1
local key = KEYS[1]
local content = ARGV[1]
local ttl = tonumber(ARGV[2])
local lockSet = redis.call('setnx', key, content)
if lockSet == 1 then
redis.call('PEXPIRE', key, ttl)
else
-- 如果value相同，则认为是同一个线程的请求，则认为重入锁
local value = redis.call('get', key)
if(value == content) then
lockSet = 1;
redis.call('PEXPIRE', key, ttl)
end
end
return lockSetunlock.lua
-- unlock key
local key = KEYS[1]
local content = ARGV[1]
local value = redis.call('get', key)
if value == content then
return redis.call('del', key)
else
return 0
end测试加锁和解锁
redis-cli --eval lock.lua lo3 , 2 60000
redis-cli --eval unlock.lua lo3 , 2在java代码中我们可以使用AOP获取当前业务的key,业务主键实现加锁，如果一旦业务异常 那么在超时后自动解锁
java调用脚本操作redis
直接使用脚本执行
private boolean accessLimit(String ip, int limit, int timeout, Jedis connection) throws IOException {
List<String> keys = Collections.singletonList(ip);
List<String> argv = Arrays.asList(String.valueOf(limit), String.valueOf(timeout));
return 1 == (long) connection.eval(loadScriptString("script.lua"), keys, argv);
}
// 加载Lua代码
private String loadScriptString(String fileName) throws IOException {
Reader reader = new InputStreamReader(Client.class.getClassLoader().getResourceAsStream(fileName));
return CharStreams.toString(reader);
}脚本工具
/**
* @author jifang
* @since 16/8/25 下午3:35.
*/
public class ScriptCaller {
private static final ConcurrentMap<String, String> SHA_CACHE = new ConcurrentHashMap<>();
private String script;
private ScriptCaller(String script) {
this.script = script;
}
public static ScriptCaller getInstance(String script) {
return new ScriptCaller(script);
}
public Object call(Jedis connection, List<String> keys, List<String> argv, boolean forceEval) {
if (!forceEval) {
String sha = SHA_CACHE.get(this.script);
if (Strings.isNullOrEmpty(sha)) {
// load 脚本得到 sha1 缓存
sha = connection.scriptLoad(this.script);
SHA_CACHE.put(this.script, sha);
}
return connection.evalsha(sha, keys, argv);
}
return connection.eval(script, keys, argv);
}
}调用端：
1.提交脚本到redis获取SHA的值
2.利用SHA的值执行脚本
public class Client {
private ScriptCaller acquireCaller = ScriptCaller.getInstance(
"local key = KEYS[1]\n" +
"local identifier = ARGV[1]\n" +
"local lockTimeOut = ARGV[2]\n" +
"\n" +
"if redis.call(\"SETNX\", key, identifier) == 1 then\n" +
" redis.call(\"EXPIRE\", key, lockTimeOut)\n" +
" return 1\n" +
"elseif redis.call(\"TTL\", key) == -1 then\n" +
" redis.call(\"EXPIRE\", key, lockTimeOut)\n" +
"end\n" +
"return 0"
);
private ScriptCaller releaseCaller = ScriptCaller.getInstance(
"local key = KEYS[1]\n" +
"local identifier = ARGV[1]\n" +
"\n" +
"if redis.call(\"GET\", key) == identifier then\n" +
" redis.call(\"DEL\", key)\n" +
" return 1\n" +
"end\n" +
"return 0"
);
@Test
public void client() {
Jedis jedis = new Jedis("127.0.0.1", 9736);
String identifier = acquireLockWithTimeOut(jedis, "ret1", 200 * 1000, 300);
System.out.println(releaseLock(jedis, "ret1", identifier));
}
String acquireLockWithTimeOut(Jedis connection, String lockName, long acquireTimeOut, int lockTimeOut) {
String identifier = UUID.randomUUID().toString();
List<String> keys = Collections.singletonList("lock:" + lockName);
List<String> argv = Arrays.asList(identifier,
String.valueOf(lockTimeOut));
long acquireTimeEnd = System.currentTimeMillis() + acquireTimeOut;
boolean acquired = false;
while (!acquired && (System.currentTimeMillis() < acquireTimeEnd)) {
if (1 == (long) acquireCaller.call(connection, keys, argv, false)) {
acquired = true;
} else {
try {
Thread.sleep(10);
} catch (InterruptedException ignored) {
}
}
}
return acquired ? identifier : null;
}
boolean releaseLock(Jedis connection, String lockName, String identifier) {
List<String> keys = Collections.singletonList("lock:" + lockName);
List<String> argv = Collections.singletonList(identifier);
return 1 == (long) releaseCaller.call(connection, keys, argv, true);
}
}Lua+Redis 断点调试环境搭建
redis3.2之后内置了debug引擎，可以通过–ldb选项进入debug模式
luarocks install remdebug
luarocks install prtr-dump
luarocks install redis-lua
4、打开ZeroBraneStudio，建lua脚本，名字随意，比如my.lua，添加如下内容：
local redis = require 'redis'
local host = "127.0.0.1"
local port = 6379
client = redis.connect(host, port)
redis.call = function(cmd, ...)
return assert(loadstring('return client:'.. string.lower(cmd) ..'(...)'))(...)
end以上是关于redis在使用lua脚本以及实现redis分布式锁的主要内容，如果未能解决你的问题，请参考以下文章