Redis實現並發阻塞鎖方案
由於用戶同時訪問線上的下訂單介面，導致在扣減庫存時出現了異常，這是一個很典型的並發問題，本篇文章為解決並發問題而生，採用的技術為Redis鎖機制+多執行緒的阻塞喚醒方法。
在實現Redis鎖機制之前，我們需要了解一下前置知識。
一、前置知識
1、多執行緒
將wait()、notifyAll()歸為到多執行緒的方法中略有一些不恰當，這兩個方法是Object中的方法。
① 當調用了wait()方法後，讓當前執行緒進入等待狀態，並且讓當前執行緒釋放對象鎖，等待既為阻塞狀態，等待notifyAll()方法的喚醒。
wait()方法和sleep()方法有一些相似之處，都是使當前執行緒阻塞，但他們實際是有一些區別的。
- 執行wait() 方法之前需要請求鎖，wait()方法執行的時候會釋放鎖，等待被喚醒的時候競爭鎖。
- sleep()只是讓當前執行緒休眠一段時間，無視鎖的存在。
- wait() 是Object類的方法 sleep()是Thread的靜態方法
② notifyAll()方法為喚醒wait()中的執行緒。
notifyAll() 和 notify() 方法都是可以喚醒調用了wait()方法，而陷入阻塞的執行緒。
但是notify()是隨機喚醒這個阻塞隊列中隨機的一個執行緒，而notifyAll()是喚醒所用的調用了wait()方法而陷入阻塞的執行緒，讓他們自己去搶佔對象鎖。
notifyAll() 和 notify() 也都是必須在加鎖的同步程式碼塊中被調用，它們起的是喚醒的作用，不是釋放鎖的作用，只用在當前同步程式碼塊中的程式執行完，也就是對象鎖自然釋放了，notifyAll() 和 notify()方法才會起作用，去喚醒執行緒。
wait()方法一般是和notify() 或者 notifyAll() 方法一起連用的。
以上為掌握本篇部落格必備的多執行緒知識，如果系統學習多執行緒的相關知識可查閱部落格 程式設計師田同學
2、Redis
加鎖的過程本質上就是往Redis中set值，當別的進程也來set值時候，發現裡面已經有值了，就只能放棄獲取稍後再試。
Redis提供了一個天然實現鎖機制的方法。
在Redis客戶端的命令為 setnx(set if not exists)
在集成Springboot中採用的方法為：
redisTemplate.opsForValue().setIfAbsent(key, value);
如果裡面set值成功會返回True，如果裡面已經存在值就會返回False。
在我們實際使用的時候，setIfAbsent()方法並不是總是返回True和False。
如果我們的業務中加了事務，該方法會返回null，不知道這是一個bug還是什麼，這是Redis的一個巨坑，浪費了很長時間才發現了這個問題，如果解決此問題可以跳轉到第四章。
二、實現原理
分散式鎖本質上要實現的目標就是在 Redis 裡面佔一個位置，當別的進程也要來占時，發現已經有人佔在那裡了，就只好放棄或者稍後再試。佔位一般是使用 setnx(set if not exists) 指令，只允許被一個客戶端佔位。先來先佔， 事辦完了，再調用 del 指令釋放茅坑。
其中，發現Redis中已經有值了，當前執行緒是直接放棄還是稍後再試分別就代表著，非阻塞鎖和阻塞鎖。
在我們的業務場景中肯定是要稍後再試（阻塞鎖），如果是直接放棄（非阻塞鎖）在資料庫層面就可以直接做，就不需要我們在程式碼大費周章了。
非阻塞鎖只能保存數據的正確性，在高並發的情況下會拋出大量的異常，當一百個並發請求到來時，只有一個請求成功，其他均會拋出異常。
Redis非阻塞鎖和 MySQL的樂觀鎖，最終達到的效果是一樣的，樂觀鎖是採用CAS的思想。
樂觀鎖方法：表欄位 加一個版本號，或者別的欄位也可以！加版本號，可以知道控制順序而已！在update 的時候可以where後面加上version= oldVersion。資料庫，在任何並發的情況下，update 成功就是 1 失敗就是 0 .可以根據返回的 1 ，0 做相應的處理！
我們更推薦大家使用阻塞鎖的方式。
當獲取不到鎖時候，我們讓當前執行緒使用wait()方法喚醒，當持有鎖的執行緒使用完成後，調用notifyAll()喚醒所有等待的方法。
三、具體實現
以下程式碼為阻塞鎖的實現方式。
業務層：
public String test() throws InterruptedException {
lock("lockKey");
System.out.println("11");
System.out.println("22");
System.out.println(Thread.currentThread().getName()+"***********");
Thread.sleep(2000);
System.out.println("33");
System.out.println("44");
System.out.println("55");
unlock("lockKey");
return "String";
}
鎖的工具類:
主要是加鎖和解鎖的兩個方法。
//每一個redis的key對應一個阻塞對象
private static HashMap<String, Object> blockers = new HashMap<>();
//當前獲得鎖的執行緒
private static Thread curThread;
public static RedisTemplate redisTemplate = (RedisTemplate) SpringUtils.getBean("redisTemplate") ;
/**
* 加鎖
* @param key
* @throws InterruptedException
*/
public static void lock(String key) {
//循環判斷是否能夠創建key， 不能則直接wait釋放CPU執行權
//放不進指說明鎖正在被佔用
System.out.println(key+"**");
while (!RedisUtil.setLock(key,"1",3)){
synchronized (key) {
blockers.put(key, key);
//wait釋放CPU執行權
try {
key.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
blockers.put(key, key);
//能夠成功創建，獲取鎖成功記錄當前獲取鎖執行緒
curThread = Thread.currentThread();
}
/**
* 解鎖
* @param key
*/
public static void unlock(String key) {
//判斷是否為加鎖的執行緒執行解鎖， 不是則直接忽略
if( curThread == Thread.currentThread()) {
RedisUtil.delete(key);
//刪除key之後需要notifyAll所有的應用， 所以這裡採用發訂閱消息給所有的應用
// RedisUtil.publish("lock", key);
//notifllall其他執行緒
Object lock = blockers.get(key);
if(lock != null) {
synchronized (lock) {
lock.notifyAll();
}
}
}
}
當我們在不加鎖時候，使用介面測試工具測試時，12345並不能都是順序執行的，會造成輸出順序不一致，如果是在我們的實際場景中，這是輸入換成了資料庫的select和update，數據出現錯亂也是很正常的情況了。
當我們加上鎖以後，12345都是順序輸出，並發問題順利解決了。
四、附錄
1、Redis存在的bug
本來lock()方法是直接調用 “Redis.setIfAbsent()” 方法，但是在使用時候一直報空指針異常，最終定位問題為Redis.setIfAbsent()方法存在問題。
在我的實際業務中，下訂單的方法使用了@Transflastion增加了事務，導致該方法返回null，我們手寫一個實現setIfAbsent()的作用。
/**
* 只有key不存在時,才設置值, 返回true, 否則返回false
*
* @param key key 不能為null
* @param value value 不能為null
* @param timeout 過期時長, 單位為妙
* @return
*/
public static Boolean setLock(String key,String value, long timeout) {
SessionCallback<Boolean> sessionCallback = new SessionCallback<Boolean>() {
List<Object> exec = null;
@Override
@SuppressWarnings("unchecked")
public Boolean execute(RedisOperations operations) throws DataAccessException {
operations.multi();
redisTemplate.opsForValue().setIfAbsent(key, value);
redisTemplate.expire(key,timeout, TimeUnit.SECONDS);
exec = operations.exec();
if(exec.size() > 0) {
return (Boolean) exec.get(0);
}
return false;
}
};
return (Boolean) redisTemplate.execute(sessionCallback);
}
方便對比，以下貼上原本的setIfAbsent()方法。
/**
* 只有key不存在時,才設置值, 返回true, 否則返回false [警告：事務或者管道情況下會報錯-可使用 setLock方法]
*
* @param key key 不能為null
* @param value value 不能為null
* @param timeout 過期時長, 單位為妙
* @return
*/
@Deprecated
public static <T> Boolean setIfAbsent(String key, T value, long timeout) {
// redisTemplate.multi();
ValueOperations<String, T> valueOperations = redisTemplate.opsForValue();
Boolean aBoolean = valueOperations.setIfAbsent(key, value, timeout, TimeUnit.SECONDS);
// redisTemplate.exec();
return aBoolean;
}
2、MySQL的鎖機制
在並發場景下MySQL會報錯，報錯資訊如下：
### Cause: com.mysql.cj.jdbc.exceptions.MySQLTransactionRollbackException: Lock wait timeout exceeded; try restarting transaction
; SQL []; Lock wait timeout exceeded; try restarting transaction; nested exception is com.mysql.cj.jdbc.exceptions.MySQLTransactionRollbackException: Lock wait timeout exceeded; try restarting transaction
問題出現的原因是，某一種表頻繁被鎖表，導致另外一個事務超時，出現問題的原因是MySQL的機制。
MySQL更新時如果where欄位存在索引會使用行鎖，否則會使用表鎖。
我們使用navichat在where欄位上加上索引，問題順利的迎刃而解。
