使用過Redis事務的應該清楚，Redis事務實現是通過打包多條命令，單獨的隔離操作，事務中的所有命令都會按順序地執行。事務在執行的過程中，不會被其他客戶端發送來的命令請求所打斷。事務中的命令要麼全部被執行，要麼全部都不執行（原子操作）。但其中有命令因業務原因執行失敗並不會阻斷後續命令的執行，且也無... ...
使用過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命令一樣的操作。
作者：王純
