（1）Redis的事務
1.1 Redis事務的定義：
Redis事務是一個單獨的隔離操作：事務中的所有命令都會序列化、按順序地執行。事務在執行的過程中，不會被其他客戶端發送來的命令請求所打斷。 Redis事務的主要作用就是串聯多個命令防止別的命令插隊。
1.2 Multi、Exec、discard命令
組隊階段：從輸入multi命令開始，后面輸入的任務命令都會依次放入到隊列中，但不會執行；
執行階段：及就是從輸入exec開始，Redis會將之前的命令隊列中的命令依次執行；
取消事務：只能在組隊的過程中可以通過discard命令來放棄組隊。
1.3 實操如下：
場景一：組隊成功，提交也成功
場景二：組隊階段報錯，提交失敗
場景三：組隊成功，提交有成功有失敗情況
1.4 Redis事務三特性
單獨的隔離操作：事務中的所有命令都會序列化、按順序地執行。事務在執行的過程中，不會被其他客戶端發送來的命令請求所打斷。
沒有隔離級別的概念：隊列中的命令沒有提交之前都不會實際被執行，因為事務提交前任何指令都不會被實際執行
不保證原子性 ：事務中如果有一條命令執行失敗，其后的命令仍然會被執行，沒有回滾
（2）Redis的事務鎖機制
2.1 Redis的鎖機制
在實際業務中，有一些場景例如：秒殺、搶車票等等，同一時間多個請求進來，那可能就會存在超賣現象，針對這種情況我們可以使用事務和redis的鎖機制來解決這種問題。
樂觀鎖(Optimistic Lock)：顧名思義，就是很樂觀，每次去拿數據的時候都認為別人不會修改，所以不會上鎖，但是在更新的時候會判斷一下在此期間別人有沒有去更新這個數據，可以使用版本號等機制。樂觀鎖適用於多讀的應用類型，這樣可以提高吞吐量。Redis就是利用這種check-and-set機制實現事務的。
悲觀鎖(Pessimistic Lock)：顧名思義，就是很悲觀，每次去拿數據的時候都認為別人會修改，所以每次在拿數據的時候都會上鎖，這樣別人想拿這個數據就會block直到它拿到鎖。傳統的關系型數據庫里邊就用到了很多這種鎖機制，比如行鎖，表鎖等，讀鎖，寫鎖等，都是在做操作之前先上鎖。
2.2 watch和unwatch的命令
watch key [key ...]-----在執行multi之前，先執行watch key1 key2，可以監視一個(或多個) key ，如果在事務執行之前這個(或這些) key 被其他命令所改動，那么事務將被打斷。
unwatch 取消 WATCH 命令對所有 key 的監視。 如果在執行 WATCH 命令之后，EXEC 命令或DISCARD 命令先被執行了的話，那么就不需要再執行UNWATCH 了。
http://doc.redisfans.com/transaction/exec.html
（3）秒殺案例
- 使用Redis解決計數器和人員記錄的事務操作
- 模擬：單個請求到並發秒殺（使用工具JMeter模擬測試）
- 超賣問題：利用事務和樂觀鎖淘汰用戶，解決超賣問題
- 模擬：加大庫存，會存在秒殺結束卻還有庫存
- 使用LUA腳本解決庫存剩余問題
1.使用Redis解決計數器和人員記錄的事務操作
寫個秒殺測試類如下：
/** * 秒殺案例，一個用戶只能秒殺成功一次 */ @RestController @RequestMapping("/testRedisSeckill") public class TestRedisSeckillController { @Autowired private RedisTemplate redisTemplate; @GetMapping("/doSeckill") public boolean doSeckill() throws IOException { String usrId = new Random().nextInt(50000) + ""; return doSeckillFun(usrId, "20210731"); } }
/** * 秒殺過程1（高並發下會超賣） * * @param usrId 用戶id * @param atcId 活動id * @return * @throws IOException */ private boolean doSeckillFun(String usrId, String atcId) throws IOException { //1.參數校驗 if (usrId == null || atcId == null) return false; //2.設置Redis值（庫存key== atcId:stock, 秒殺成功用戶key== atcId:userId） String stockKey = atcId + ":stock"; String userIdKey = atcId + ":userId"; //3.獲取庫存，如果庫存是空,秒殺還沒開始 Object stock = redisTemplate.opsForValue().get(stockKey);//獲取庫存 if (stock == null) { System.out.println("別着急，秒殺還沒開始呢！！"); return false; } //4.判斷用戶是否重復秒殺操作(Set類型操作) if (redisTemplate.opsForSet().isMember(userIdKey, usrId)) { System.out.println("你已經秒殺成功了，不能重復秒殺"); return false; } //5.判斷庫存數量，小於1，秒殺結束 int stock1 = (int) stock; if (stock1 < 1) { System.out.println("秒殺結束了。。。"); return false; } //6.秒殺過程（庫存減1，把秒殺成功用戶添加到用戶清單） redisTemplate.opsForSet().add(userIdKey, usrId); redisTemplate.opsForValue().decrement(stockKey);//庫存-1 System.out.println("恭喜你！秒殺成功了！"); return true; }
1.1 模擬場景1：單個請求
先設置庫存10個
Jmeter模擬單個請求
查看redis剩余庫存和用戶清單：
1.2 模擬高並發500個請求
看着沒什么毛病對吧，那如果我把並發加大到500，會出現什么情況呢？
在執行之前先清空redis的數據，點擊jmeter執行
控制台輸出：
查看redis數據
發現庫存 -190，出現超賣了，所以我們的場景1的代碼在高並發的情況下會出現超賣的問題，那么針對這個問題我們需要使用樂觀鎖來解決
2.使用樂觀鎖來解決
代碼如下：
/** * 秒殺過程2(樂觀鎖解決超賣問題) * * @param usrId 用戶id * @param atcId 活動id * @return * @throws IOException */ private boolean doSeckillFun(String usrId, String atcId) throws IOException { //1.參數校驗 if (usrId == null || atcId == null) return false; //2.設置Redis值（庫存key== atcId:stock, 秒殺成功用戶key== atcId:userId） String stockKey = atcId + ":stock"; String userIdKey = atcId + ":userId"; //通過 SessionCallback，保證所有的操作都在同一個 Session 中完成 //更常見的寫法仍是采用 RedisTemplate 的默認配置，即不開啟事務支持。 // 但是，我們可以通過使用 SessionCallback，該接口保證其內部所有操作都是在同一個Session中 SessionCallback<Object> callback = new SessionCallback<Object>() { @Override public Object execute(RedisOperations operations) throws DataAccessException { //3. 打開事務支持 //redisTemplate.setEnableTransactionSupport(true); //4.增加樂觀鎖進行對庫存的監視 operations.watch(stockKey); //5.獲取庫存，如果庫存是空,秒殺還沒開始 Object stock = operations.opsForValue().get(stockKey);//獲取庫存 if (stock == null) { System.out.println("別着急，秒殺還沒開始呢！！"); return false; } //6.判斷用戶是否重復秒殺操作(Set類型操作) if (operations.opsForSet().isMember(userIdKey, usrId)) { System.out.println("你已經秒殺成功了，不能重復秒殺"); return false; } //7.判斷庫存數量，小於1，秒殺結束 int stock1 = (int) stock; if (stock1 < 1) { System.out.println("秒殺結束了。。。"); return false; } //8. 增加事務 operations.multi(); //9.秒殺過程 operations.opsForValue().decrement(stockKey);//庫存-1 operations.opsForSet().add(userIdKey, usrId);//把秒殺成功用戶添加到用戶清單 //10.執行事務 List<Object> list = operations.exec(); //11.判斷事務提交是否失敗 if (list == null || list.size() == 0) { System.out.println("秒殺失敗"); return false; } System.out.println("恭喜你！秒殺成功了！"); return true; } }; return (boolean) redisTemplate.execute(callback); }
2.1 設置10個庫存，繼續模擬500個並發請求，結果如下：
終於解決超賣問題了，嘻嘻
2.2 那我把庫存加大到300個，繼續模擬500個並發請求，會出現什么情況呢？
執行Jmeter模擬500個並發，結果如下：
雖然沒有超賣問題了，但是有500個請求卻還剩余102個庫存，那么就有下邊的lua解決庫存遺留問題。
3. LUA腳本在Redis中的優勢 ：將復雜的或者多步的redis操作，寫為一個腳本，一次提交給redis執行，減少反復連接redis的次數。提升性能。 LUA腳本是類似redis事務，有一定的原子性，不會被其他命令插隊，可以完成一些redis事務性的操作。 但是注意redis的lua腳本功能，只有在Redis 2.6 以上的版本才可以使用。 利用lua腳本淘汰用戶，解決超賣問題。 redis 2.6版本以后，通過lua腳本解決爭搶問題，實際上是redis 利用其單線程的特性，用任務隊列的方式解決多任務並發問題。
使用Lua腳本解決庫存遺留的問題，代碼如下：
/** * 秒殺過程3(LUA解決庫存剩余問題) * * @param usrId 用戶id * @param atcId 活動id * @return * @throws IOException */ private boolean doSeckillFun(String usrId, String atcId) throws IOException { String luaScript = "local userId=KEYS[1];\r\n" + "local stockKey=KEYS[2];\r\n" + "local userIdKey=KEYS[3];\r\n" + "local userExists=redis.call(\"sismember\",userIdKey,userId); \r\n" + "if tonumber(userExists)==1 \r\n" + "then \r\n" + " return 2;\r\n" + "end \r\n" + "local num= redis.call(\"get\" ,stockKey);\r\n" + "if tonumber(num)<=0 then return 0;\r\n" + "else \r\n " + " redis.call(\"decr\",stockKey);\r\n" + "redis.call(\"sadd\",userIdKey,userId);\r\n" + "end \r\n" + "return 1;"; // 指定 lua 腳本，並且指定返回值類型 // (為什么返回值不用 Integer 接收而是用 Long。這里是因為 spring-boot-starter-data-redis 提供的返回類型里面不支持 Integer。) DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>(luaScript, Long.class); List<String> keys = new ArrayList<>(); keys.add(usrId); keys.add(atcId + ":stock"); keys.add(atcId + ":userId"); // 參數一：redisScript，參數二：key列表，參數三：arg（可多個） Long result = (Long) redisTemplate.execute(redisScript, keys); if (0 == result) { System.out.println("秒殺結束了。。。"); } else if (1 == result) { System.out.println("恭喜你！秒殺成功了！"); return true; } else if (2 == result) { System.out.println("你已經秒殺成功了，不能重復秒殺"); } else { System.out.println("秒殺異常啦~"); } return false; }
lua腳本：
local userId=KEYS[1]; local stockKey=KEYS[2]; local userIdKey=KEYS[3]; local userExists=redis.call("sismember",userIdKey,userId); if tonumber(userExists)==1 then return 2; end local num= redis.call("get" ,stockKey); if tonumber(num)<=0 then return 0; else redis.call("decr",stockKey); redis.call("sadd",userIdKey,userId); end return 1;
這樣一個完美的秒殺案例就完成了。嘻嘻嘻~~~~