如果程序運(yùn)行的極慢（硬件處理慢或者進(jìn)行了GC），導(dǎo)致30秒已經(jīng)到了，鎖已經(jīng)失效了，程序還沒有運(yùn)行完成，這時(shí)候，就會(huì)有另一個(gè)線程總想鉆個(gè)空子，導(dǎo)致票的超賣問題。
哈嘍，大家好，我是指北君。
今天帶大家深入剖析一下Redis分布式鎖，徹底搞懂它。
場(chǎng)景
既然要搞懂Redis分布式鎖，那肯定要有一個(gè)需要它的場(chǎng)景。
高并發(fā)售票問題就是一個(gè)經(jīng)典案例。
搭建環(huán)境
- 準(zhǔn)備redis服務(wù)，設(shè)置redis的鍵值對(duì)：set ticket 10
- 準(zhǔn)備 postman、JMeter 等模擬高并發(fā)請(qǐng)求的工具
- 核心代碼
@Service
public class TicketServiceImpl implements TicketService {
@Autowired
private StringRedisTemplate stringRedisTemplate;
private Logger logger = LoggerFactory.getLogger(TicketServiceImpl.class);
@Override
public String sellTicket(){
String ticketStr = stringRedisTemplate.opsForValue().get("ticket");
int ticket = 0;
if (null != ticketStr) {
ticket = Integer.parseInt(ticketStr);
}
if (ticket > 0) {
int ticketNew = ticket - 1;
stringRedisTemplate.opsForValue().set("ticket", String.valueOf(ticketNew));
logger.info("當(dāng)前票的庫(kù)存為：" + ticketNew);
} else {
logger.info("手速不夠呀，票已經(jīng)賣光了...");
}
return "搶票成功...";
}
}
分析解決問題
以上代碼沒有做任何的加鎖操作，在高并發(fā)情況下，票的超賣情況很嚴(yán)重，根本無法正常使用
分析1
既然要加分布式鎖，那么我們可以使用Redis中的setnx命令來模擬一個(gè)鎖。
redis> EXISTS job # job 不存在
(integer) 0
redis> SETNX job "programmer" # job 設(shè)置成功
(integer) 1
redis> SETNX job "code-farmer" # 嘗試覆蓋 job ，失敗
(integer) 0
當(dāng)一個(gè)線程進(jìn)入到當(dāng)前方法中，使用 setnx 設(shè)置一個(gè)鍵，如果設(shè)置成功，就允許繼續(xù)訪問，設(shè)置失敗，就不能訪問該方法；
當(dāng)方法運(yùn)行完畢時(shí)，將這個(gè)鍵刪除，下一次再有線程來訪問時(shí)，就重新執(zhí)行該操作。
public String sellTicket(){
String lock="lock";
// 如果成功設(shè)置這個(gè)值，證明目前該方法并沒有被操作，可以進(jìn)行賣票操作
Boolean tag = stringRedisTemplate.opsForValue().setIfAbsent(lock, "");
if (!tag) { // 如果設(shè)置失敗，證明當(dāng)前方法正在被執(zhí)行，不允許再次執(zhí)行
// 實(shí)際開發(fā)環(huán)境應(yīng)該使用隊(duì)列來完成訪問操作，這里主要探究分布式鎖的問題，所以僅僅模擬了場(chǎng)景
// 這里使用自旋的方式，防止訪問信息丟失
sellTicket();
return "當(dāng)前訪問人數(shù)過多，請(qǐng)稍后訪問...";
}
String ticketStr = stringRedisTemplate.opsForValue().get("ticket");
int ticket = 0;
if (null != ticketStr) {
ticket = Integer.parseInt(ticketStr);
}
if (ticket > 0) {
int ticketNew = ticket - 1;
stringRedisTemplate.opsForValue().set("ticket", String.valueOf(ticketNew));
logger.info("當(dāng)前票的庫(kù)存為：" + ticketNew);
} else {
logger.info("手速不夠呀，票已經(jīng)賣光了...");
}
stringRedisTemplate.delete(lock);
return "搶票成功...";
}分析2
上述的代碼在程序正常運(yùn)行下不會(huì)出現(xiàn)票超賣的問題，但是我們需要考慮：
- 如果程序運(yùn)行中系統(tǒng)出現(xiàn)了異常，導(dǎo)致無法刪除lock?，就會(huì)造成死鎖的問題。也許有人馬上就會(huì)想到，使用 try{} finally {} ，在finally中進(jìn)行刪除鎖的操作。
但是，如果是分布式架構(gòu)，第一個(gè)服務(wù)器接收到請(qǐng)求，加了鎖，此時(shí)第二個(gè)服務(wù)器也接收到請(qǐng)求，setnx 命令失敗，需要執(zhí)行return操作，根據(jù)finally的特性，執(zhí)行return之前，需要先執(zhí)行finally里的代碼，于是，第二個(gè)服務(wù)器把鎖給刪除了，程序中鎖失效了，肯定會(huì)出現(xiàn)票超賣等一系列問題。
- 如果程序在運(yùn)行中直接徹底死了（比如，程序員閑著沒事兒，來了個(gè) kill -9；或者斷電），就算加了finally，finally也不能執(zhí)行，還是會(huì)出現(xiàn)死鎖問題
解決方法：
給鎖加一個(gè)標(biāo)識(shí)符，只允許自己來操作鎖，其他訪問程序不能操作鎖
還要給鎖加一個(gè)過期時(shí)間，這樣就算程序死了，當(dāng)時(shí)間過期后，還是能夠繼續(xù)執(zhí)行
public String sellTicket(){
String lock="lock"; // 鎖的鍵
String lockId = UUID.randomUUID().toString(); // 鎖的值：唯一標(biāo)識(shí)
try{
// 如果成功設(shè)置這個(gè)值，證明目前該方法并沒有被操作，可以進(jìn)行賣票操作
// 添加一個(gè)過期時(shí)間，暫定為 30秒，這里的操作具有原子性，如果過期時(shí)間設(shè)置失敗，鍵也會(huì)設(shè)置失敗
Boolean tag = stringRedisTemplate.opsForValue().setIfAbsent(lock, lockId, 30, TimeUnit.SECONDS);
if (!tag) { // 如果設(shè)置失敗，證明當(dāng)前方法正在被執(zhí)行，不允許再次執(zhí)行
// 實(shí)際開發(fā)環(huán)境應(yīng)該使用隊(duì)列來完成訪問操作，這里主要探究分布式鎖的問題，所以僅僅模擬了場(chǎng)景
// 不設(shè)置回調(diào)的話，訪問信息會(huì)丟失
sellTicket();
return "當(dāng)前訪問人數(shù)過多，請(qǐng)稍后訪問...";
}
String ticketStr = stringRedisTemplate.opsForValue().get("ticket");
int ticket = 0;
if (null != ticketStr) {
ticket = Integer.parseInt(ticketStr);
}
if (ticket > 0) {
int ticketNew = ticket - 1;
stringRedisTemplate.opsForValue().set("ticket", String.valueOf(ticketNew));
logger.info("當(dāng)前票的庫(kù)存為：" + ticketNew);
} else {
logger.info("手速不夠呀，票已經(jīng)賣光了...");
}
} finally {
// 如果redis中的值，和當(dāng)前的值一致，才允許刪除鎖。
if (lockId.equals(stringRedisTemplate.opsForValue().get(lock))) {
stringRedisTemplate.delete(lock);
}
}
return "搶票成功...";
}分析3
寫到這里已經(jīng)可以解決大部分問題了，但是還需要考慮一個(gè)問題：
如果程序運(yùn)行的極慢（硬件處理慢或者進(jìn)行了GC），導(dǎo)致30秒已經(jīng)到了，鎖已經(jīng)失效了，程序還沒有運(yùn)行完成，這時(shí)候，就會(huì)有另一個(gè)線程總想鉆個(gè)空子，導(dǎo)致票的超賣問題。
這里我們可以使用 sleep 模擬一下
......
if (ticket > 0) {
try {
// 為了測(cè)試方便，過期時(shí)間和線程暫停時(shí)間都改成了3秒
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
int ticketNew = ticket - 1;
stringRedisTemplate.opsForValue().set("ticket", String.valueOf(ticketNew));
......
- 這樣運(yùn)行就會(huì)出現(xiàn)極其嚴(yán)重的超賣問題
那么該如何設(shè)置這個(gè)過期時(shí)間呢？繼續(xù)加大？這顯然是不合適的，因?yàn)闊o論多么大，總有可能出現(xiàn)問題。
解決方法
我們可以使用守護(hù)線程，來保證這個(gè)時(shí)間永不過期
public String sellTicket(){
String lock="lock"; // 鎖的鍵
String lockId = UUID.randomUUID().toString(); // 鎖的值：唯一標(biāo)識(shí)
MyThread myThread = null; // 鎖的守護(hù)線程
try{
// 如果成功設(shè)置這個(gè)值，證明目前該方法并沒有被操作，可以進(jìn)行賣票操作
// 添加一個(gè)過期時(shí)間，暫定為 3 秒，這里的操作具有原子性，如果過期時(shí)間設(shè)置失敗，鍵也會(huì)設(shè)置失敗
Boolean tag = stringRedisTemplate.opsForValue().setIfAbsent(lock, lockId, 3, TimeUnit.SECONDS);
if (!tag) { // 如果設(shè)置失敗，證明當(dāng)前方法正在被執(zhí)行，不允許再次執(zhí)行
// 實(shí)際開發(fā)環(huán)境應(yīng)該使用隊(duì)列來完成訪問操作，這里主要探究分布式鎖的問題，所以僅僅模擬了場(chǎng)景
// 不設(shè)置回調(diào)的話，訪問信息會(huì)丟失
sellTicket();
return "當(dāng)前訪問人數(shù)過多，請(qǐng)稍后訪問...";
}
// 開啟守護(hù)線程, 每隔三分之一的時(shí)間，給鎖續(xù)命
myThread = new MyThread(lock);
myThread.setDaemon(true);
myThread.start();
String ticketStr = stringRedisTemplate.opsForValue().get("ticket");
int ticket = 0;
if (null != ticketStr) {
ticket = Integer.parseInt(ticketStr);
}
if (ticket > 0) {
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
int ticketNew = ticket - 1;
stringRedisTemplate.opsForValue().set("ticket", String.valueOf(ticketNew));
logger.info("當(dāng)前票的庫(kù)存為：" + ticketNew);
} else {
logger.info("手速不夠呀，票已經(jīng)賣光了...");
}
} finally {
// 如果redis中的值，和當(dāng)前的值一致，才允許刪除鎖。
if (lockId.equals(stringRedisTemplate.opsForValue().get(lock))) {
// 程序運(yùn)行結(jié)束，需要關(guān)閉守護(hù)線程
myThread.stop();
stringRedisTemplate.delete(lock);
logger.info("釋放鎖成功...");
}
}
return "搶票成功...";
}
/** 使用后臺(tái)線程進(jìn)行續(xù)命
* 守護(hù)線程
* 在主線程下 如果有一個(gè)守護(hù)線程 這個(gè)守護(hù)線程的生命周期 跟主線程是同生死的
*/
class MyThread extends Thread{
String lock;
MyThread (String lock) {
this.lock = lock;
}
@Override
public void run(){
while (true) {
try {
// 三分之一的時(shí)間
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
// 假設(shè)線程還活著，就要給鎖續(xù)命
logger.info("線程續(xù)命ing...");
stringRedisTemplate.expire(lock, 3, TimeUnit.SECONDS);
}
}
}總結(jié)
到這里，我們已經(jīng)基本實(shí)現(xiàn)了redis分布式鎖，并且可以在高并發(fā)場(chǎng)景下正常運(yùn)行。
需要注意的是，實(shí)現(xiàn)分布式鎖的代碼肯定不是最佳的，重要的是了解分布式鎖的實(shí)現(xiàn)原理，以及發(fā)現(xiàn)問題并解決問題的思路。
