应用过 Redis 事务的应该分明，Redis 事务实现是通过打包多条命令，独自的隔离操作，事务中的所有命令都会按程序地执行。事务在执行的过程中，不会被其余客户端发送来的命令申请所打断。事务中的命令要么全副被执行，要么全副都不执行（原子操作）。但其中有命令因业务起因执行失败并不会阻断后续命令的执行，且也无奈回滚曾经执行过的命令。如果想要实现和 MySQL 一样的事务处理能够应用 Lua 脚本来实现，Lua 脚本中可实现简略的逻辑判断，执行停止等操作。
1 初始 Lua 脚本
Lua 是一个玲珑的脚本语言，Redis 脚本应用 Lua 解释器来执行脚本。Reids 2.6 版本通过内嵌反对 Lua 环境。执行脚本的常用命令为 EVAL。编写 Lua 脚本就和编写 shell 脚本一样的简略。Lua 语言具体教程参见
示例：
--[[
version:1.0
检测 key 是否存在，如果存在并设置过期工夫
入参列表：参数个数量：1
KEYS[1]:goodsKey 商品 Key
返回列表 code:
+0：不存在
+1：存在
--]]
local usableKey = KEYS[1]
--[判断 usableKey 在 Redis 中是否存在 存在将过期工夫缩短 1 分钟 并返回是否存在后果 --]
local usableExists = redis.call('EXISTS', usableKey)
if (1 == usableExists) then
redis.call('PEXPIRE', usableKey, 60000)
end
return {usableExists}- 示例代码中 redis.call(), 是 Redis 内置办法，用与执行 redis 命令
- if () then end 是 Lua 语言根本分支语法
- KEYS 为 Redis 环境执行 Lua 脚本时 Redis Key 参数，如果应用变量入参应用 ARGV 接管
- “—”代表单行正文“—[[多行正文 —]]”
2 实际利用
2.1 需要剖析
经典案例需要：库存量扣减并检测库存量是否短缺。
根底需要剖析：商品以后库存量 >= 扣减数量时，执行扣减。商品以后库存量 < 扣减数量时，返回库存有余
实现计划剖析：
1）MySQL 事务实现：
利用 DB 行级锁，锁定要扣减商品库存量数据，再判断库存量是否短缺，短缺执行扣减，否则返回库存有余。
执行库存扣减，再判断扣减后后果是否小于 0，小于 0 阐明库存有余，事务回滚，否则提交事务。
2）计划优缺点剖析：
长处：MySQL 人造反对事务，实现难度低。
毛病：不思考热点商品场景，当业务量达到一定量级时会达到 MySQL 性能瓶颈，单库无奈反对业务时扩大问题成为难点，分表、分库等计划对性能开发、业务运维、数据运维都须要有针对于分表、分库计划所配套的零碎或计划。对于零碎革新实现难度较高。
Redis Lua 脚本事务实现：将库存扣减判断库存量最小原子操作逻辑编写为 Lua 脚本。
从 DB 中初始化商品库存数量，利用 Redis WATCH 命令。
判断商品库存量是否短缺，短缺执行扣减，否则返回库存有余。
执行库存扣减，再判断扣减后后果是否小于 0，小于 0 阐明库存有余，反向操作减少缩小库存量，返回操作后果
计划优缺点剖析：
长处：Redis 命令执行单线程个性，毋庸思考并发锁竟争所带来的实现复杂度。Redis 人造反对 Lua 脚本，Lua 语言学习难度低，实现与 MySQL 计划难度相当。Redis 同一时间单位反对的并发量比 MySQL 大，执行耗时更小。对于业务量的增长能够扩容 Redis 集群分片。
毛病：暂无
2.2 Redis Lua 脚本事务计划实现
初始化商品库存量：
// 利用 Watch 命令乐观乐个性，缩小锁竞争所损耗的性能
public boolean init(InitStockCallback initStockCallback, InitOperationData initOperationData) {
//SessionCallback 会话级 Rdis 事务回调接口 针对于 operations 所有操作将在同一个 Redis tcp 连贯上实现
List<Object> result = stringRedisTemplate.execute(new SessionCallback<List<Object>>() {public List<Object> execute(RedisOperations operations) {Assert.notNull(operations, "operations must not be null");
//Watch 命令用于监督一个(或多个) key，如果在事务执行之前这个(或这些) key 被其余命令所改变，那么事务将被打断
// 当出前并发初始化同一个商品库存量时，只有一个能胜利
operations.watch(initOperationData.getWatchKeys());
int initQuantity;
try {
// 查问 DB 商品库存量
initQuantity = initStockCallback.getInitQuantity(initOperationData);
} catch (Exception e) {
// 异样后开释 watch
operations.unwatch();
throw e;
}
// 开启 Reids 事务
operations.multi();
//setNx 设置商品库存量
operations.opsForValue().setIfAbsent(initOperationData.getGoodsKey(), String.valueOf(initQuantity));
// 设置商品库存量 key 过期工夫
operations.expire(initOperationData.getGoodsKey(), Duration.ofMinutes(60000L));
/// 执行事事务
return operations.exec();}
});
// 判断事务执行后果
if (!CollectionUtils.isEmpty(result) && result.get(0) instanceof Boolean) {return (Boolean) result.get(0);
}
return false;
}库存扣减逻辑
--[[
version:1.0
减可用库存
入参列表：参数个数量：KEYS[1]:usableKey 商品可用量 Key
KEYS[3]:usableSubtractKey 减量记录 key
KEYS[4]:operateKey 操作防重 Key
KEYS[5]:hSetRecord 记录操作单号信息
ARGV[1]:quantity 操作数量
ARGV[2]:version 操作版本号
ARGV[5]:serialNumber 单据流水编码
ARGV[6]:record 是否记录过程量
返回列表：+1：操作胜利
0: 操作失败
-1: KEY 不存在
-2：反复操作
-3: 库存有余
-4：过期操作
-5：缺量库存有余
-6：可用负库存
--]]
local usableKey = KEYS[1];
local usableSubtractKey = KEYS[3]
local operateKey = KEYS[4]
local hSetRecord = KEYS[5]
local quantity = tonumber(ARGV[1])
local version = ARGV[2]
local serialNumber = ARGV[5]
--[判断商品库存 key 是否存在 不存在返回 -1 --]
local usableExists = redis.call('EXISTS', usableKey);
if (0 == usableExists) then
return {-1, version, 0, 0};
end
--[设置防重 key 设置失败阐明操作反复返回 -2 --]
local isNotRepeat = redis.call('SETNX', operateKey, version);
if (0 == isNotRepeat) then
redis.call('SET', operateKey, version);
return {-2, version, quantity, 0};
end
--[商品库存量扣减后小 0 阐明库存有余 回滚扣减数量 并革除防重 key 立刻过期 返回 -3 --]
local usableResult = redis.call('DECRBY', usableKey, quantity);
if (usableResult < 0) then
redis.call('INCRBY', usableKey, quantity);
redis.call('PEXPIRE', operateKey, 0);
return {-3, version, 0, usableResult};
end
--[记录扣减量并设置防重 key 30 天后过期 返回 1--]
-- [须要记录过程量与过程单据信息 --]
local usableSubtractResult = redis.call('INCRBY', usableSubtractKey, quantity);
redis.call('HSET', hSetRecord, serialNumber, quantity)
redis.call('PEXPIRE', hSetRecord, 3600000)
redis.call('PEXPIRE', operateKey, 2592000000)
redis.call('PEXPIRE', usableKey, 3600000)
return {1, version, quantity, 0, usableResult ,usableSubtractResult}初始化 Lua 脚本到 Redis 服务器
// 读取 Lua 脚本文件
private String readLua(File file) {StringBuilder sbf = new StringBuilder();
try (BufferedReader reader = new BufferedReader(new FileReader(file))) {
String temp;
while (Objects.nonNull(temp = reader.readLine())) {sbf.append(temp);
sbf.append('\n');
}
return sbf.toString();} catch (FileNotFoundException e) {LOGGER.error("[{}]文件不存在", file.getPath());
} catch (IOException e) {LOGGER.error("[{}]文件读取异样", file.getPath());
}
return null;
}
// 初始化 Lua 脚本到 Redis 服务器 胜利后会返回脚本对应的 sha1 码，零碎缓存脚本 sha1 码，// 通过 sha1 码能够在 Redis 服务器执行对应的脚本
public String scriptLoad(File file) {String script = readLua(file)
return stringRedisTemplate.execute((RedisCallback<String>) connection -> connection.scriptLoad(script.getBytes()));
}脚本执行
public OperationResult evalSha(String redisScriptSha1,OperationData operationData) {List<String> keys = operationData.getKeys();
String[] args = operationData.getArgs();
// 执行 Lua 脚本 keys 为 Lua 脚本中应用到的 KEYS args 为 Lua 脚本中应用到的 ARGV 参数
// 如果是在 Redis 集群模式下，同一个脚本中的多个 key, 要满足多个 key 在同一个分片
// 服务器开启 hash tag 性能，多个 key 应用 {} 将雷同局部包裹
// 例：usableKey:{EMG123} operateKey:operate:{EMG123}
Object result = stringRedisTemplate.execute(redisScriptSha1, keys, args);
// 解析执行后果
return parseResult(operationData, result);
}3 总结
Redis 在小数据操作并发可达到 10W, 针对与业务中对资源强校验且高并发场景下应用 Redis 配合 Lua 脚本实现简略逻辑解决抗并发量是个不错的抉择。
注：Lua 脚本逻辑尽量简略，Lua 脚本实用于耗时短且原子操作。耗时长影响 Redis 服务器性能，非原子操作或逻辑简单会减少于脚本调试与维度难度。现实状态是将业务用 Lua 脚本包装成一个如 Redis 命令一样的操作。
作者：王纯