??各位小伙伴們大家好，歡迎來到這個小扎扎的Redis 6專欄，在這個系列專欄中我對B站黑馬的Redis教程進行一個總結，鑒于 看到就是學到、學到就是賺到 精神，這波依然是血賺 ┗|｀O′|┛

💡Redis知識點速覽

---

• 🍖 分布式鎖
• • 🥩 業務邏輯分析
  • 🥩 Redis命令
  • 🥩 代碼實現
• 🍖 分布式鎖誤刪問題
• • 🥩 問題原因分析
  • 🥩 代碼實現
• 🍖 Lua腳本

??講過上一節的分析可知，服務器集群項目中的鎖是無法精準的鎖住線程資源的，于是我們就是需要使用分布式鎖，分布式鎖該如何使用又有什么注意點呢？就讓我們進入接下來的學習

??首先，使用idea模擬搭建一個tomcat服務器集群，并使用Nginx對集群中的服務器實現負載均衡配置完負載均衡之后，發送兩次請求就會在idea的運行窗口中發現，兩次請求的運行是分別在兩個服務器中完成，這就是集群的輪詢機制

🍖 分布式鎖

🥩 業務邏輯分析

??在單JVM虛擬機多線程執行的情況下，可以使用JVM內部的鎖機制來控制多進程的并發執行，借此可以保證一個用戶只能下一個優惠券訂單。但是在分布式的情況下，每一個JVM虛擬機都有一個鎖監視器，不同JVM里的不同線程之間的訪問的并不是同一個鎖監視器，所以說此時再使用synchronized鎖就無法滿足一個用戶限買一單的業務情況了，于是就需要使用分布式鎖

??分布式鎖就是滿足分布式系統或集群模式下多進程可見并且互斥的鎖。一般實現分布式鎖的技術主要就是MySQL、Redis和ZooKeeper，但是綜合對比來看的話，Redis作分布式鎖的性能更高一些，Redis是在JVM虛擬機之外的一種應用可以滿足多線程都可見，互斥可以使用setnx這種的互斥命令來實現，但是使用Redis會存在安全性問題，如果Redis崩潰的話會導致鎖無法釋放而出現死鎖現象，解決這一問題的方案就是使用TTL過期時間，就算崩潰也可以實現到期自動釋放。

🥩 Redis命令

??使用Redis實現分布式鎖的步驟主要就是使用setnx體現互斥鎖，然后expire過期時間防止宕機死鎖，但是如果服務在setnx之后expire之前宕機的話，依舊會造成死鎖現象。于是我們可以使用以下命令在互斥的同時設置超時時間，這樣的話即是在設置鎖之后宕機，依舊可以憑借超時時間釋放鎖

SET lock thread NX EX ttl超時時間

🥩 代碼實現

??將獲取鎖和釋放鎖業務抽取出來，使用接口和實現類來完成

/*** @author : mereign* @date : 2022/6/10 - 12:01* @description : 分布式鎖*/
public interface ILock {/*** 嘗試獲取鎖* @param timeoutSec 鎖的超時時間* @return 是否成功獲取鎖*/boolean tryLock(long timeoutSec);/*** 釋放鎖*/void unLock();
}

/*** @author : mereign* @date : 2022/6/10 - 13:31* @description : 分布式鎖的實現類*/
public class SimpleRedisLock implements ILock {private String name;/***先獲取StringRedisTemplate對象，才能使用代碼操作Redis*/private StringRedisTemplate stringRedisTemplate;public SimpleRedisLock(String name, StringRedisTemplate stringRedisTemplate) {this.name = name;this.stringRedisTemplate = stringRedisTemplate;}@Overridepublic boolean tryLock(long timeoutSec) {// 獲取當前操作線程的標識long threadId = Thread.currentThread().getId();// 獲取鎖Boolean res = stringRedisTemplate.opsForValue().setIfAbsent(RedisConstants.KEY_PREFIX + name, threadId + "", timeoutSec, TimeUnit.SECONDS);// res是Boolean的包裝類，返回結果的時候涉及到拆箱問題，有可能存在結果為null的情況，此時就需要返回結果與true的比較，避免了空指針風險return Boolean.TRUE.equals(res);}@Overridepublic void unLock() {// 釋放鎖stringRedisTemplate.delete(RedisConstants.KEY_PREFIX + name);}
}

??定義了分布式鎖的獲取和釋放，接下來就是在一人一單業務代碼中將鎖機制升級成多線程鎖了，主要修改的代碼為就是5~14行，由單體的synchronized鎖改為使用自定義的Redis鎖，并根據不同線程獲取鎖的不同結果定義了不同的業務

public Result secKillVoucher(Long voucherId) {// 單用戶id(攔截器中做登錄驗證的用戶id)Long userId = UserHolder.getUser().getId();// 創建鎖對象SimpleRedisLock lock = new SimpleRedisLock("order:" + userId, stringRedisTemplate);// 獲取鎖boolean isLock = lock.tryLock(1200);// 判斷是否獲取鎖成功if (!isLock) {// 獲取鎖失敗，返回錯誤或者重試return Result.fail("不允許重復下單！" );}// 獲取鎖成功，繼續下單的業務邏輯try {// 查詢優惠券SeckillVoucher seckillVoucher = seckillVoucherService.getById(voucherId);// 獲取時間 判斷秒殺活動是否開始或者結束if (seckillVoucher.getBeginTime().isAfter(LocalDateTime.now())) {return Result.fail("活動暫未開始");} else if (seckillVoucher.getEndTime().isBefore(LocalDateTime.now())) {return Result.fail("活動已經結束");}// 判斷庫存是否充足if (seckillVoucher.getStock() < 1) {return Result.fail("庫存不足，活動結束");}// user_id和voucher_id聯合查詢訂單數int count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();// 訂單數為1 就說明已經下過單了if (count > 0) {return Result.fail("您已經購買過該商品了");}// 扣減庫存boolean update = seckillVoucherService.update().setSql("stock = stock - 1").eq("voucher_id", voucherId).gt("stock", 0).update();if (!update) {return Result.fail("庫存不足！");}// 創建訂單 并返回idVoucherOrder order = new VoucherOrder();// 訂單id(redis全局唯一id) 下單用戶id(攔截器中做登錄驗證的用戶id) 優惠券id(直接傳過來的id)long orderId = generator.nextId("order");order.setId(orderId);order.setUserId(userId);order.setVoucherId(voucherId);save(order);return Result.ok(orderId);} finally {// 釋放鎖lock.unLock();}
}

🍖 分布式鎖誤刪問題

🥩 問題原因分析

??這個問題出現在Redis鎖設置的超時時間上，由于設置了超時時間，所以可能出現一下情況：即當線程1獲取到鎖之后執行下單業務，但是由于業務堵塞鎖已經超出TTL時間自動釋放；此時線程2趁機獲取Redis鎖成功執行下單業務，線程2的下單業務執行到一半時線程1完成下單使用del命令釋放鎖；此時線程1釋放的是線程2的鎖，于是現在鎖又處于閑置狀態，于是線程3來獲取Redis鎖成功執行下單業務；此時，一共有同一個用戶的兩個線程在同時操作

??為了解決以上出現的問題，需要在每次釋放鎖之前都通過鎖的線程標識(Redis鎖對應的值)判斷一下是不是自己的鎖，如果是就使用del命令釋放鎖，否則就不做操作。但是有一點值得注意，之前鎖的線程標識使用的是線程的name，這樣的話很容易就造成不同JVM虛擬機里的線程name沖突影響判斷，于是可以使用UUID隨機生成一組數字加上線程name作為線程的標識，這樣更能確保唯一性

🥩 代碼實現

??綜上所述，一共有兩處需要改進的地方，一個是使用UUID加線程name作為線程標識(主要修改的是獲取鎖方法加上UUID的獲取)，一個是在使用del釋放鎖之前判斷一下是否是自己的鎖

public static final String ID_PREFIX = UUID.randomUUID(true) + "-";public boolean tryLock(long timeoutSec) {// 獲取當前操作線程的標識String threadId = RedisConstants.ID_PREFIX + Thread.currentThread().getId();// 獲取鎖Boolean res = stringRedisTemplate.opsForValue().setIfAbsent(RedisConstants.KEY_PREFIX + name, threadId, timeoutSec, TimeUnit.SECONDS);// res是Boolean的包裝類，返回結果的時候涉及到拆箱問題，有可能存在結果為null的情況，此時就需要返回結果與true的比較，避免了空指針風險return Boolean.TRUE.equals(res);
}

public void unLock() {// 獲取當前操作線程的標識String threadId = RedisConstants.ID_PREFIX + Thread.currentThread().getId();// 通過鎖名 獲取redis中存儲的鎖對應的標識String rid = stringRedisTemplate.opsForValue().get(RedisConstants.KEY_PREFIX + name);if (threadId.equals(rid)) {// 釋放鎖stringRedisTemplate.delete(RedisConstants.KEY_PREFIX + name);}
}

🍖 Lua腳本

??Redis提供了Lua腳本功能，在一個腳本中編寫多條Redis命令，確保多條命令執行時的原子性。Lua是一種編程語言，它的基本語法大家可以參考網站：https://www.runoob.com/lua/lua-tutorial.html
使用Redis命令調用腳本的常見命令可以是：

EVAL “redis.call(‘set’, ‘key’, ‘value’)” num

??上述命令解釋為EVAL是調用，后面雙引號中就是所調用的腳本語句，而最后的num即腳本語句中的KEYS類型參數的個數，num之外的就是ARGV(value)類型的參數。比如說，接下來這一個語句就代表著：setname為Rose，其中KEYS類型的參數有1個，就是num后面的第一個name，剩下的都是ARGV(value)類型的數據，其中調用的是KEYS[1]和ARGV[2]，也就是name和Rose

EVAL “redis.call(‘set’, ‘KEYS[1]’, ‘ARGV[2]’)” 1 name age Rose