如何使用Redis解決高並發

NoSQL

Not Only SQL的簡稱。 NoSQL是解決傳統的RDBMS在處理某些問題時比較乏力而提出的。

即非關係型資料庫，它們不保證關係資料的ACID特性，資料之間一般沒有關聯，在擴展上就非常容易實現，並且擁有較高的效能。

Redis

redis是nosql的典型代表，也是目前網路公司的必用技術。

Redis主要採用哈希表來實現鍵值對儲存。大多時候是直接以快取的形式被使用，使得請求不直接存取到磁碟，所以效率方面是很不錯的，完全能滿足中小型企業的使用需求。

常用資料型別

• 字串string

• 雜湊hash

• #清單list

• 集合sets

• #有序集合sort set

##使用頻率上string和hash會高一些，各類型有各自的操作指令，無非增刪改查，具體的指令後面我會整理一份。

痛點

web應用程式在眾多請求同時發生時，可能會導致資料讀取、儲存體上出現錯誤，即發生髒讀、髒資料產生。

在分散式專案下，會出現更多的問題。

思路

並發時，本質其實就是多個請求同時進來了，沒辦法正確的去進行處理。

可以將所有的請求放在 一個佇列，讓請求們依照一個順序，挨個進來執行業務邏輯。使用訊息佇列是目前可行的解決方案，我將在下次整理一篇關於如何處理高並發的訊息佇列文章

還有一個方法是直接將並行轉為串行，Java提供了synchronized，即同步，不過這個在效率要求比較苛刻的地方或者分散式專案下還是不太合適的方案，這裡就引出了使用redis來實現分散式鎖，從而解決並發問題。

分散式鎖定

在分散式項目中，使用一個唯一、通用、效率高的標識，來表示鎖定和解鎖。

redis實作起來很簡單，也就是對一個key是否存在來表示是否上鎖、是否解鎖。

以string型別舉例：

Integer stock = goodsMapper.getStock();
if (stock > 0) {
    stock =- 1;
    goodsMapper.updateStock(stock);
}

以上是最簡單的秒殺偽程式碼，我們嘗試用redis實作分散式鎖定。

// 这里是错误代码，只是一个思考过程，请耐心看完哦
String key = "REDIS_DISTRIBUTION_LOCKER"; // 分布式锁名称
String value = jedisUtils.get(key);
if (value != null) { // 未上锁
    // wingzingliu
    jedisUtils.set(key, 1); // 上锁
    Integer stock = goodsMapper.getStock();
    if (stock > 0) {
        stock =- 1;
        goodsMapper.updateStock(stock);
        jedisUtils.del(key); // 释放锁
    }
}

以上程式碼可能會出現一個問題，就是當同時多個請求進來，某次多個請求都拿到value為空，線程A進入if 走到// wingzingliu這裡的時候，還未上鎖，其他請求也進來了，這樣就會出現髒資料了。

這裡的程式碼問題就是出在沒有考慮原子性問題。

所以我們要使用到redis的一個setNx指令，本質也是設定值，但是這是一個原子操作，執行之後會回傳是否設定成功。

redis> SETNX job "programmer"    # job 设置成功
(integer) 1
 
redis> SETNX job "code-farmer"   # 尝试覆盖 job ，失败
(integer) 0
 
redis> GET job                   # 没有被覆盖
"programmer"

重點關注 當有值時，會失敗，回傳0。所以我們的程式碼會改造成以下這個樣子。

// 这里是错误代码，只是一个思考过程，请耐心看完哦
String key = "REDIS_DISTRIBUTION_LOCKER"; // 分布式锁名称
Long result = jedisUtils.setNx(key, 1);
if (result > 0) { // 上锁成功，进入逻辑
    // wingzingliu1
    Integer stock = goodsMapper.getStock();
    if (stock > 0) {
        stock =- 1;
        goodsMapper.updateStock(stock);
 
        System.out.println("购买成功!");
    } else {
        System.out.println("没有库存了!");
    }
    // wingzingliu2
    jedisUtils.del(key); // 释放锁
}

以上我們就可以保證原子性，能正確的依序去處理。

可是還有一個隱藏的問題，就是當某個執行緒執行上鎖成功後，在wingzingliu1到wingzingliu2之間時，程式拋異常了，那麼程式終止了，就無法釋放鎖，其他執行緒也都進不來了。

解決方案是加上try catch finally區塊，在finally裡面去釋放鎖定。

可是那如果是宕機呢？上鎖之後宕機了，finally裡面的依然不會執行，鎖沒有得到釋放，不手動處理的情況下，以後所有線程也無法進入。

所以引入了redis的過期時間，到了某個時間自動解鎖。

// 这里是不够完善的代码，请耐心看完哦
try {
    String key = "REDIS_DISTRIBUTION_LOCKER"; // 分布式锁名称
    Long result = jedisUtils.setNx(key, 1, 30); // 假设处理逻辑需要20s左右，设置了30秒自动过期
    if (result > 0) { // 上锁成功，进入逻辑
        Integer stock = goodsMapper.getStock();
        if (stock > 0) {
            stock =- 1;
            goodsMapper.updateStock(stock);
 
            System.out.println("购买成功!");
        } else {
            System.out.println("没有库存了!");
        }
    }
} catch (Exception e) {
    
} finally {
    jedisUtils.del(key); // 释放锁
}

以上是比較完善的分散式鎖了，但是還有一個小瑕疵，就是假設某一次請求A處理的很慢，預計20s但是跑了35s，到了30s的時候鎖過期了，其他請求就自然進來了。

這不僅會導致一次並發執行，而且在請求A處理完後，仍會繼續執行釋放鎖定操作，從而實際上將鎖定交給了下一個執行緒。以此類推，整個並發控制就亂了。

理論上可以設定一個更大的key過期時間，但並不是最好的解決方案。這裡就引出一個概念：鎖續命。

鎖續命

如其名，給鎖續命。實現就是 當鎖快過期的時候，去延長鎖的時間。假設使用一個30秒的鎖，每10秒進行一次檢測以確認鎖是否仍然存在。如果鎖依然存在，則將鎖繼續保持在30秒。這樣就避免掉了上面的這個可能出現的問題。

這裡使用一個定時任務，週期性的呼叫即可。

擴充

剛剛對key設定的value是1，其實能使用請求ID來進行保存，這樣就能知道鎖是由哪個請求上的，在解鎖的時候也可以避免解鎖了其他線程上的鎖。具體由前端傳遞，或是由服務端以某種規則產生都可以。

以上是如何使用Redis解決高並發的詳細內容。更多資訊請關注PHP中文網其他相關文章！