今天帶大家深入剖析一下Redis分布式鎖，徹底搞懂它。

場(chǎng)景

既然要搞懂Redis分布式鎖，那肯定要有一個(gè)需要它的場(chǎng)景。

高并發(fā)售票問題就是一個(gè)經(jīng)典案例。

搭建環(huán)境

1. 準(zhǔn)備redis服務(wù)，設(shè)置redis的鍵值對(duì)：set ticket 10
2. 準(zhǔn)備 postman、JMeter 等模擬高并發(fā)請(qǐng)求的工具
3. 核心代碼

@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)前票的庫存為：" + 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)前票的庫存為：" + ticketNew);
    } else {
        logger.info("手速不夠呀，票已經(jīng)賣光了...");
    }
    stringRedisTemplate.delete(lock);
    return "搶票成功...";
}

分析2

上述的代碼在程序正常運(yùn)行下不會(huì)出現(xiàn)票超賣的問題，但是我們需要考慮：

1. 如果程序運(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)票超賣等一系列問題。
2. 如果程序在運(yùn)行中直接徹底死了（比如，程序員閑著沒事兒，來了個(gè) kill -9；或者斷電），就算加了finally，finally也不能執(zhí)行，還是會(huì)出現(xiàn)死鎖問題

解決方法：

1. 給鎖加一個(gè)標(biāo)識(shí)符，只允許自己來操作鎖，其他訪問程序不能操作鎖
2. 還要給鎖加一個(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)前票的庫存為：" + 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)前票的庫存為：" + 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)問題并解決問題的思路。