單體架構(gòu)下鎖的實(shí)現(xiàn)方案
1. ReentrantLock全局鎖
ReentrantLock（可重入鎖），指的是一個(gè)線程再次對(duì)已持有的鎖保護(hù)的臨界資源時(shí)，重入請(qǐng)求將會(huì)成功。
簡(jiǎn)單的與我們常用的Synchronized進(jìn)行比較：
| ReentrantLock | Synchronized | |
|---|---|---|
| 鎖實(shí)現(xiàn)機(jī)制 | 依賴(lài)AQS | 監(jiān)視器模式 |
| 靈活性 | 支持響應(yīng)超時(shí)、中斷、嘗試獲取鎖 | 不靈活 |
| 釋放形式 | 必須顯示調(diào)用unlock()釋放鎖 | 自動(dòng)釋放監(jiān)視器 |
| 鎖類(lèi)型 | 公平鎖 & 非公平鎖 | 非公平鎖 |
| 條件隊(duì)列 | 可關(guān)聯(lián)多個(gè)條件隊(duì)列 | 關(guān)聯(lián)一個(gè)條件隊(duì)列 |
| 可重入性 | 可重入 | 可重入 |
AQS機(jī)制：如果被請(qǐng)求的共享資源空閑，那么就當(dāng)前請(qǐng)求資源的線程設(shè)置為有效的工作線程，將共享資源通過(guò)CAScompareAndSetState設(shè)置為鎖定狀態(tài)；如果共享資源被占用，就采用一定的阻塞等待喚醒機(jī)制（CLH變體的FIFO雙端隊(duì)列）來(lái)保證鎖分配。
可重入性：無(wú)論是公平鎖還是非公平鎖的情況，加鎖過(guò)程會(huì)利用一個(gè)state值
private volatile int state
- state值初始化的時(shí)候?yàn)?，表示沒(méi)有任何線程持有鎖
- 當(dāng)有線程來(lái)請(qǐng)求該鎖時(shí)，state值會(huì)自增1，同一個(gè)線程多次獲取鎖，就會(huì)多次+1，這就是可重入的概念
- 解鎖也是對(duì)state值自減1，一直到0，此線程對(duì)鎖釋放。
public class LockExample {
static int count = 0;
static ReentrantLock lock = new ReentrantLock();
public static void main(String[] args) throws InterruptedException {
Runnable runnable = new Runnable() {
@Override
public void run() {
try {
// 加鎖
lock.lock();
for (int i = 0; i < 10000; i++) {
count++;
}
} catch (Exception e) {
e.printStackTrace();
}
finally {
// 解鎖，放在finally子句中，保證鎖的釋放
lock.unlock();
}
}
};
Thread thread1 = new Thread(runnable);
Thread thread2 = new Thread(runnable);
thread1.start();
thread2.start();
thread1.join();
thread2.join();
System.out.println("count: " + count);
}
}
/**
* 輸出
* count: 20000
*/
2. Mysql行鎖、樂(lè)觀鎖
樂(lè)觀鎖即是無(wú)鎖思想，一般都是基于CAS思想實(shí)現(xiàn)的，而在MySQL中通過(guò)version版本號(hào) + CAS無(wú)鎖形式實(shí)現(xiàn)樂(lè)觀鎖；例如T1，T2兩個(gè)事務(wù)一起并發(fā)執(zhí)行時(shí)，當(dāng)T2事務(wù)執(zhí)行成功提交后，會(huì)對(duì)version+1，所以T1事務(wù)執(zhí)行的version條件就無(wú)法成立了。
對(duì)sql語(yǔ)句進(jìn)行加鎖以及狀態(tài)機(jī)的操作，也可以避免不同線程同時(shí)對(duì)count值訪問(wèn)導(dǎo)致的數(shù)據(jù)不一致問(wèn)題。
// 樂(lè)觀鎖 + 狀態(tài)機(jī)
update
table_name
set
version = version + 1,
count = count + 1
where
id = id AND version = version AND count = [修改前的count值];
// 行鎖 + 狀態(tài)機(jī)
update
table_name
set
count = count + 1
where
id = id AND count = [修改前的count值]
for update;
3. 細(xì)粒度的ReetrantLock鎖
如果我們直接采用ReentrantLock全局加鎖，那么這種情況是一條線程獲取到鎖，整個(gè)程序全部的線程來(lái)到這里都會(huì)阻塞；但是我們?cè)陧?xiàng)目里面想要針對(duì)每個(gè)用戶(hù)在操作的時(shí)候?qū)崿F(xiàn)互斥邏輯，所以我們需要更加細(xì)粒度的鎖。
public class LockExample {
private static Map<String, Lock> lockMap = new ConcurrentHashMap<>();
public static void lock(String userId) {
// Map中添加細(xì)粒度的鎖資源
lockMap.putIfAbsent(userId, new ReentrantLock());
// 從容器中拿鎖并實(shí)現(xiàn)加鎖
lockMap.get(userId).lock();
}
public static void unlock(String userId) {
// 先從容器中拿鎖，確保鎖的存在
Lock locak = lockMap.get(userId);
// 釋放鎖
lock.unlock();
}
}
弊端：如果每一個(gè)用戶(hù)請(qǐng)求共享資源，就會(huì)加鎖一次，后續(xù)該用戶(hù)就沒(méi)有在登錄過(guò)平臺(tái)，但是鎖對(duì)象會(huì)一直存在于內(nèi)存中，這等價(jià)于發(fā)生了內(nèi)存泄漏，所以鎖的超時(shí)和淘汰機(jī)制機(jī)制需要實(shí)現(xiàn)。
4. 細(xì)粒度的Synchronized全局鎖
上面的加鎖機(jī)制使用到了鎖容器ConcurrentHashMap，該容易為了線程安全的情況，多以底層還是會(huì)用到Synchronized機(jī)制，所以有些情況，使用lockMap需要加上兩層鎖。
那么我們是不是可以直接使用Synchronized來(lái)實(shí)現(xiàn)細(xì)粒度的鎖機(jī)制
public class LockExample {
public static void syncFunc1(Long accountId) {
String lock = new String(accountId + "").intern();
synchronized (lock) {
System.out.println(Thread.currentThread().getName() + "拿到鎖了");
// 模擬業(yè)務(wù)耗時(shí)
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
System.out.println(Thread.currentThread().getName() + "釋放鎖了");
}
}
public static void syncFunc2(Long accountId) {
String lock = new String(accountId + "").intern();
synchronized (lock) {
System.out.println(Thread.currentThread().getName() + "拿到鎖了");
// 模擬業(yè)務(wù)耗時(shí)
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
System.out.println(Thread.currentThread().getName() + "釋放鎖了");
}
}
// 使用 Synchronized 來(lái)實(shí)現(xiàn)更加細(xì)粒度的鎖
public static void main(String[] args) {
new Thread(()-> syncFunc1(123456L), "Thread-1").start();
new Thread(()-> syncFunc2(123456L), "Thread-2").start();
}
}
/**
* 打印
* Thread-1拿到鎖了
* Thread-1釋放鎖了
* Thread-2拿到鎖了
* Thread-2釋放鎖了
*/
- 從代碼中我們發(fā)現(xiàn)實(shí)現(xiàn)加鎖的對(duì)象其實(shí)就是一個(gè)與用戶(hù)ID相關(guān)的一個(gè)字符串對(duì)象，這里可能會(huì)有疑問(wèn)，我每一個(gè)新的線程進(jìn)來(lái)，new的都是一個(gè)新的字符串對(duì)象，只不過(guò)字符串內(nèi)容一樣，怎么能夠保證可以安全的鎖住共享資源呢；
- 這其實(shí)需要?dú)w功于后面的
intern()函數(shù)的功能； -
intern()函數(shù)用于在運(yùn)行時(shí)將字符串添加到堆空間中的字符串常量池中，如果字符串已經(jīng)存在，返回字符串常量池中的引用。
分布式架構(gòu)下鎖的實(shí)現(xiàn)方案
核心問(wèn)題：我們需要找到一個(gè)多個(gè)進(jìn)程之間所有線程可見(jiàn)的區(qū)域來(lái)定義這個(gè)互斥量。
一個(gè)優(yōu)秀的分布式鎖的實(shí)現(xiàn)方案應(yīng)該滿(mǎn)足如下幾個(gè)特性：
- 分布式環(huán)境下，可以保證不同進(jìn)程之間的線程互斥
- 同一時(shí)刻，同時(shí)只允許一條線程成功獲取到鎖資源
- 保證互斥量的地方需要保證高可用性
- 要保證可以高性能的獲取鎖和釋放鎖
- 可以支持同一線程的鎖重入性
- 具備合理的阻塞機(jī)制，競(jìng)爭(zhēng)鎖失敗的線程要有相應(yīng)的處理方案
- 支持非阻塞式的獲取鎖。獲取鎖失敗的線程可以直接返回
- 具備合理的鎖失效機(jī)制，如超時(shí)失效等，可以確保避免死鎖情況出現(xiàn)
Redis實(shí)現(xiàn)分布式鎖
- redis屬于中間件，可獨(dú)立部署；
- 對(duì)于不同的Java進(jìn)程來(lái)說(shuō)都是可見(jiàn)的，同時(shí)性能也非?？捎^
- 依賴(lài)與redis本身提供的指令
setnx key value來(lái)實(shí)現(xiàn)分布式鎖；區(qū)別于普通set指令的是只有當(dāng)key不存在時(shí)才會(huì)設(shè)置成功，key存在時(shí)會(huì)返回設(shè)置失敗
代碼實(shí)例：
// 扣庫(kù)存接口
@RequestMapping("/minusInventory")
public String minusInventory(Inventory inventory) {
// 獲取鎖
String lockKey = "lock-" + inventory.getInventoryId();
int timeOut = 100;
Boolean flag = stringRedisTemplate.opsForValue()
.setIfAbsent(lockKey, "竹子-熊貓",timeOut,TimeUnit.SECONDS);
// 加上過(guò)期時(shí)間，可以保證死鎖也會(huì)在一定時(shí)間內(nèi)釋放鎖
stringRedisTemplate.expire(lockKey,timeOut,TimeUnit.SECONDS);
if(!flag){
// 非阻塞式實(shí)現(xiàn)
return "服務(wù)器繁忙...請(qǐng)稍后重試?。?！";
}
// ----只有獲取鎖成功才能執(zhí)行下述的減庫(kù)存業(yè)務(wù)----
try{
// 查詢(xún)庫(kù)存信息
Inventory inventoryResult =
inventoryService.selectByPrimaryKey(inventory.getInventoryId());
if (inventoryResult.getShopCount() <= 0) {
return "庫(kù)存不足，請(qǐng)聯(lián)系賣(mài)家....";
}
// 扣減庫(kù)存
inventoryResult.setShopCount(inventoryResult.getShopCount() - 1);
int n = inventoryService.updateByPrimaryKeySelective(inventoryResult);
} catch (Exception e) { // 確保業(yè)務(wù)出現(xiàn)異常也可以釋放鎖，避免死鎖
// 釋放鎖
stringRedisTemplate.delete(lockKey);
}
if (n > 0)
return "端口-" + port + ",庫(kù)存扣減成功?。?！";
return "端口-" + port + ",庫(kù)存扣減失?。。?！";
}
作者：竹子愛(ài)熊貓
鏈接：https://juejin.cn/post/7038473714970656775
過(guò)期時(shí)間的合理性分析：
因?yàn)閷?duì)于不同的業(yè)務(wù)，我們?cè)O(shè)置的過(guò)期時(shí)間的長(zhǎng)短都會(huì)不一樣，太長(zhǎng)了不合適，太短了也不合適；
所以我們想到的解決方案是設(shè)置一條子線程，給當(dāng)前鎖資源續(xù)命。具體實(shí)現(xiàn)是，子線程間隔2-3s去查詢(xún)一次key是否過(guò)期，如果還沒(méi)有過(guò)期則代表業(yè)務(wù)線程還在執(zhí)行業(yè)務(wù)，那么則為該key的過(guò)期時(shí)間加上5s。
但是為了避免主線程意外死亡后，子線程會(huì)一直為其續(xù)命，造成“長(zhǎng)生鎖”的現(xiàn)象，所以將子線程變?yōu)橹鳎I(yè)務(wù)）線程的守護(hù)線程，這樣子線程就會(huì)跟著主線程一起死亡。
// 續(xù)命子線程
public class GuardThread extends Thread {
private static boolean flag = true;
public GuardThread(String lockKey,
int timeOut, StringRedisTemplate stringRedisTemplate){
……
}
@Override
public void run() {
// 開(kāi)啟循環(huán)續(xù)命
while (flag){
try {
// 先休眠一半的時(shí)間
Thread.sleep(timeOut / 2 * 1000);
}catch (Exception e){
e.printStackTrace();
}
// 時(shí)間過(guò)了一半之后再去續(xù)命
// 先查看key是否過(guò)期
Long expire = stringRedisTemplate.getExpire(
lockKey, TimeUnit.SECONDS);
// 如果過(guò)期了，代表主線程釋放了鎖
if (expire <= 0){
// 停止循環(huán)
flag = false;
}
// 如果還未過(guò)期
// 再為則續(xù)命一半的時(shí)間
stringRedisTemplate.expire(lockKey,expire
+ timeOut/2,TimeUnit.SECONDS);
}
}
}
// 創(chuàng)建子線程為鎖續(xù)命
GuardThread guardThread = new GuardThread(lockKey,timeOut,stringRedisTemplate);
// 設(shè)置為當(dāng)前 業(yè)務(wù)線程 的守護(hù)線程
guardThread.setDaemon(true);
guardThread.start();
作者：竹子愛(ài)熊貓
鏈接：https://juejin.cn/post/7038473714970656775
Redis主從架構(gòu)下鎖失效的問(wèn)題
為了在開(kāi)發(fā)過(guò)程保證Redis的高可用，會(huì)采用主從復(fù)制架構(gòu)做讀寫(xiě)分離，從而提升Redis的吞吐量以及可用性。但是如果一條線程在redis主節(jié)點(diǎn)上獲取鎖成功之后，主節(jié)點(diǎn)還沒(méi)有來(lái)得及復(fù)制給從節(jié)點(diǎn)就宕機(jī)了，此時(shí)另一條線程訪問(wèn)redis就會(huì)在從節(jié)點(diǎn)上面訪問(wèn)，同時(shí)也獲取鎖成功，這時(shí)候臨界資源的訪問(wèn)就會(huì)出現(xiàn)安全性問(wèn)題了。
解決辦法：
- 紅鎖算法（官方提出的解決方案）：多臺(tái)獨(dú)立的Redis同時(shí)寫(xiě)入數(shù)據(jù)，在鎖失效時(shí)間之內(nèi)，一半以上的機(jī)器寫(xiě)成功則返回獲取鎖成功，失敗的時(shí)候釋放掉那些成功的機(jī)器上的鎖。但這種做法缺點(diǎn)是成本高需要獨(dú)立部署多臺(tái)Redis節(jié)點(diǎn)。
- 額外記錄鎖狀態(tài)：再額外通過(guò)其他獨(dú)立部署的中間件（比如DB）來(lái)記錄鎖狀態(tài)，在新線程獲取鎖之前需要先查詢(xún)DB中的鎖持有記錄，只要當(dāng)鎖狀態(tài)為未持有時(shí)再?lài)L試獲取分布式鎖。但是這種情況缺點(diǎn)顯而易見(jiàn)，獲取鎖的過(guò)程實(shí)現(xiàn)難度復(fù)雜，性能開(kāi)銷(xiāo)也非常大；另外還需要配合定時(shí)器功能更新DB中的鎖狀態(tài)，保證鎖的合理失效機(jī)制。
- 使用Zookepper實(shí)現(xiàn)
Zookeeper實(shí)現(xiàn)分布式鎖
Zookeeper數(shù)據(jù)區(qū)別于redis的數(shù)據(jù)，數(shù)據(jù)是實(shí)時(shí)同步的，主節(jié)點(diǎn)寫(xiě)入后需要一半以上的節(jié)點(diǎn)都寫(xiě)入才會(huì)返回成功。所以如果像電商、教育等類(lèi)型的項(xiàng)目追求高性能，可以放棄一定的穩(wěn)定性，推薦使用redis實(shí)現(xiàn)；例如像金融、銀行、政府等類(lèi)型的項(xiàng)目，追求高穩(wěn)定性，可以犧牲一部分性能，推薦使用Zookeeper實(shí)現(xiàn)。
分布式鎖性能優(yōu)化
上面加鎖確實(shí)解決了并發(fā)情況下線程安全的問(wèn)題，但是我們面對(duì)100w個(gè)用戶(hù)同時(shí)去搶購(gòu)1000個(gè)商品的場(chǎng)景該如何解決呢？
- 可與將共享資源做一下提前預(yù)熱，分段分散存儲(chǔ)一份。搶購(gòu)時(shí)間為下午15：00，提前再14：30左右將商品數(shù)量分成10份，并將每一塊數(shù)據(jù)進(jìn)行分別加鎖，來(lái)防止并發(fā)異常。
- 另外也需要在redis中寫(xiě)入10個(gè)key，每一個(gè)新的線程進(jìn)來(lái)先隨機(jī)的分配一把鎖，然后進(jìn)行后面的減庫(kù)存邏輯，完成之后釋放鎖，以便之后的線程使用。
- 這種分布式鎖的思想就是，將原先一把鎖就可以實(shí)現(xiàn)的多線程同步訪問(wèn)共享資源的功能，為了提高瞬時(shí)情況下多線程的訪問(wèn)速度，還需要保證并發(fā)安全的情況下一種實(shí)現(xiàn)方式。
參考文章：
-
https://juejin.cn/post/7236213437800890423
-
https://juejin.cn/post/7038473714970656775
-
https://tech.meituan.com/2019/12/05/aqs-theory-and-apply.html
作者：京東科技 焦?jié)杀?mark hidden color="red">文章來(lái)源：http://www.zghlxwxcb.cn/news/detail-748548.html
來(lái)源：京東云開(kāi)發(fā)者社區(qū) 轉(zhuǎn)載請(qǐng)注明來(lái)源文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-748548.html
到了這里，關(guān)于并發(fā)情況如何實(shí)現(xiàn)加鎖來(lái)保證數(shù)據(jù)一致性？的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
