[[386548]]

今天我們就來把基于Redis實現的分布式鎖，集成到我們的項目中，分布式鎖歷來都受到大家的關注。不管是工作中、面試中，分布式鎖永遠是個不老的話題，也希望大家能掌握此技能，便于大家日后能"升官發財"。

分布式鎖

背景

為什么要有分布式鎖呢?不是已經有synchronized、ReantrantLock等相關鎖了嗎?

是的，我們在開發應用的時候，如果需要對某一個共享變量進行多線程同步訪問的時候，可以使用我們學到的鎖進行處理，并且可以完美的運行，毫無Bug!

注意：這是單機應用，后來業務發展，需要做集群，一個應用需要部署到幾臺機器上然后做負載均衡 ：

 

上圖可以看到，變量A存在三個服務器內存中(這個變量A主要體現是在一個類中的一個成員變量，是一個有狀態的對象)，如果不加任何控制的話，變量A同時都會在分配一塊內存，三個請求發過來同時對這個變量操作，顯然結果是不對的!即使不是同時發過來，三個請求分別操作三個不同內存區域的數據，變量A之間不存在共享，也不具有可見性，處理的結果也是不對的!

如果我們業務中確實存在這個場景的話，我們就需要一種方法解決這個問題!

為了保證一個方法或屬性在高并發情況下的同一時間只能被同一個線程執行，在傳統單體應用單機部署的情況下，可以使用并發處理相關的功能進行互斥控制。但是，隨著業務發展的需要，原單體單機部署的系統被演化成分布式集群系統后，由于分布式系統多線程、多進程并且分布在不同機器上，這將使原單機部署情況下的并發控制鎖策略失效，單純的應用并不能提供分布式鎖的能力。為了解決這個問題就需要一種跨機器的互斥機制來控制共享資源的訪問，這就是分布式鎖要解決的問題!

分布鎖的基本原理

分布式環境下，多臺機器上多個進程對同一個共享資源(數據、文件等)進行操作，如果不做互斥，就有可能出現“余額扣成負數”，或者“商品超賣”的情況。

為了解決這個問題，需要分布式鎖服務。首先，來看一下分布式鎖應該具備哪些條件。

• 互斥性：在任意時刻，對于同一個鎖，只有一個客戶端能持有，從而保證一個共享資源同一時間只能被一個客戶端操作;
• 安全性：即不會形成死鎖，當一個客戶端在持有鎖的期間崩潰而沒有主動解鎖的情況下，其持有的鎖也能夠被正確釋放，并保證后續其它客戶端能加鎖;
• 可用性：當提供鎖服務的節點發生宕機等不可恢復性故障時，“熱備” 節點能夠接替故障的節點繼續提供服務，并保證自身持有的數據與故障節點一致。
• 對稱性：對于任意一個鎖，其加鎖和解鎖必須是同一個客戶端，即客戶端 A 不能把客戶端 B 加的鎖給解了。

目前市面上，分布式鎖的實現方案大致有三種：

1. 數據庫樂觀鎖;
2. 基于分布式緩存實現的鎖服務，典型代表有 Redis 和基于 Redis 的 RedLock;
3. 基于分布式一致性算法實現的鎖服務，典型代表有 ZooKeeper、Chubby 和 ETCD。

項目中，使用的最多的是后兩種，其實每種方案都各有利弊。

預防死鎖

我們看下面這個典型死鎖場景。

一個客戶端獲取鎖成功，但是在釋放鎖之前崩潰了，此時該客戶端實際上已經失去了對公共資源的操作權，但卻沒有辦法請求解鎖(刪除 Key-Value 鍵值對)，那么，它就會一直持有這個鎖，而其它客戶端永遠無法獲得鎖。

我們的解決方案是：在加鎖時為鎖設置過期時間，當過期時間到達，Redis 會自動刪除對應的 Key-Value，從而避免死鎖。需要注意的是，這個過期時間需要結合具體業務綜合評估設置，以保證鎖的持有者能夠在過期時間之內執行完相關操作并釋放鎖。

另外，前面已經說了，在實際項目中我們都是使用后兩種方案，所以我們重點在后兩種方案上。說明 此文章是基于前面我們搞的項目繼續開展，同時把Redis已經集成到項目中了，所以此文中分布式鎖是基于Redis的實現。

基于Redis實現分布式鎖

先說一下使用Redis實現方案的思路：

• setnx +expire+delete
• setnx+lua
• set key value px milliseconds nx

簡單版

創建分布鎖

import org.slf4j.Logger; 
import org.slf4j.LoggerFactory; 
import org.springframework.data.redis.core.RedisTemplate; 
import org.springframework.stereotype.Component; 
 
import javax.annotation.Resource; 
import java.util.UUID; 
import java.util.concurrent.TimeUnit; 
 
@Component 
public class DistributedLockV1 { 
 
    private Logger logger = LoggerFactory.getLogger(DistributedLockV1.class); 
 
    @Resource 
    private RedisTemplate redisTemplate; 
 
    public boolean lock(String businessKey) { 
        boolean result = false; 
        String uniqueValue = UUID.randomUUID().toString(); 
        try { 
            //@see <a href="http://redis.io/commands/setnx">Redis Documentation: SETNX</a> 
            result = redisTemplate.opsForValue().setIfAbsent(businessKey, uniqueValue); 
            if (!result) { 
                return false; 
            } 
            //設置key的有效期 
            redisTemplate.expire(businessKey, 10, TimeUnit.SECONDS); 
            return result; 
        } catch (Exception ex) { 
            logger.error("獲取鎖失敗", ex); 
        } 
        return result; 
    } 
 
    public void unlock(String businessKey) { 
        try { 
            //delete 
            redisTemplate.delete(businessKey); 
        } catch (Exception ex) { 
            logger.error("釋放鎖失敗", ex); 
        } 
    } 
} 

就這樣，一個簡單的分布式鎖就實現了，但是這里會存在問題，問題也不是一定會出現，在特定的時刻還是會出現的。

下面，我們就來把這個分布式鎖應用到用戶賬戶余額扣減的功能中。

我們來創建一張用戶賬戶表，表中主要有userId和余額：

CREATE TABLE `user_account` ( 
  `id` bigint NOT NULL AUTO_INCREMENT, 
  `user_id` bigint DEFAULT NULL, 
  `balance` decimal(10,2) DEFAULT NULL, 
  `create_time` timestamp NULL DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP, 
  `update_time` timestamp NULL DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP, 
  PRIMARY KEY (`id`) 
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8; 

然后創建UserAccountRepository接口。

import com.tian.user.entity.UserAccount; 
import org.springframework.data.jpa.repository.JpaRepository; 
import org.springframework.data.jpa.repository.Modifying; 
import org.springframework.data.jpa.repository.Query; 
import org.springframework.stereotype.Repository; 
 
import java.math.BigDecimal; 
 
@Repository 
public interface UserAccountRepository extends JpaRepository<UserAccount, Long> { 
 
    UserAccount findByUserId(Long userId); 
    //通過userId更新余額 
    @Modifying 
    @Query("update UserAccount u set u.balance=?1 where  u.userId=?2") 
    void updateBalanceByUserId(BigDecimal balance, Long userId); 
} 

創建UserAccountService和實現類，并實現其扣減方法：

import com.tian.user.entity.UserAccount; 
 
import java.math.BigDecimal; 
 
public interface UserAccountService {  
    /** 
     * 扣減余額 
     * @param userId 當前用戶userId 
     * @param balance 當前需要減余額 
     * @return 是否扣減成功 
     */ 
    boolean reduceBalance(Long userId, BigDecimal balance); 
} 
import com.tian.user.entity.UserAccount; 
import com.tian.user.lock.DistributedLockV1; 
import com.tian.user.repository.UserAccountRepository; 
import com.tian.user.service.UserAccountService; 
import org.springframework.stereotype.Service; 
import org.springframework.transaction.annotation.Transactional; 
 
import javax.annotation.Resource; 
import java.math.BigDecimal; 
import java.util.Date; 
 
@Service 
@Transactional(rollbackFor = Exception.class) 
public class UserAccountServiceImpl implements UserAccountService { 
    private Logger logger= LoggerFactory.getLogger(getClass()); 
    @Resource 
    private UserAccountRepository userAccountRepository; 
    @Resource 
    private DistributedLockV1 distributedLockV1;  
     
    @Override 
    public boolean reduceBalance(Long userId, BigDecimal balance) { 
        try { 
            //把該賬戶給鎖住，使用userId作為key。 
            boolean lock = distributedLockV1.lock(userId.toString()); 
            //獲取鎖失敗，則直接返回扣減失敗 
            if (!lock) { 
                return false; 
            } 
            UserAccount userAccount = userAccountRepository.findByUserId(userId); 
            BigDecimal currBalance = userAccount.getBalance(); 
            //校驗余額是否足夠扣減 
            if (currBalance.compareTo(balance) > 0) { 
                BigDecimal newBalance = currBalance.subtract(balance); 
                //扣減余額 
                userAccountRepository.updateBalanceByUserId(newBalance, userId); 
                return true; 
            } 
        }catch(Exception ex){ 
            logger.error("余額扣減失敗", ex); 
        } finally { 
            //釋放鎖 
            distributedLockV1.unlock(userId.toString()); 
        } 
        return false; 
    } 
} 

到此，簡單版的分布式鎖，以及如何使用，這里就已經搞完了。下面我們來理一下思路：

1.使用setnx(set not exist),就是如何set的這個key在redis不存在就返回true，否則返回false。

2.對已經set的key設置有效期，使用expire設置有效期。

3.校驗我們的可用余額是否足夠扣減，不夠就直接結束并使用delete刪除redis中的key。

4.扣減余額，更新數據庫余額值。

5.刪除key，delete redis中key。

那么，問題來了，第一步、第二步都成功了。但假如第三步查詢余額、扣減余額耗時20秒了，上面我們對Redis中key的有效期設置的10秒，也就是超時了，key過期了，并且在10秒到20秒之間又有其他線程來獲取到鎖了，然后此時把其他線程拿到的鎖給刪了，把其他線程的鎖給解了。此時，不就亂了嗎?

這也是面試中常被問使用redis做分布式鎖，業務超時了怎么辦?

升級版

我們可以把每次key對應的value返回，當釋放鎖的時候，判斷當前key對應的value是否是當前手里持有的value。

然后，我們針對上面的進行修改一版。

import org.slf4j.Logger; 
import org.slf4j.LoggerFactory; 
import org.springframework.data.redis.core.RedisTemplate; 
import org.springframework.stereotype.Component; 
 
import javax.annotation.Resource; 
import java.util.UUID; 
import java.util.concurrent.TimeUnit; 
 
@Component 
public class DistributedLockV1 { 
    private Logger logger = LoggerFactory.getLogger(DistributedLockV1.class); 
    @Resource 
    private RedisTemplate redisTemplate; 
 
    public String lockV2(String businessKey) { 
        boolean result = false; 
        String uniqueValue = UUID.randomUUID().toString(); 
        try { 
            result = redisTemplate.opsForValue().setIfAbsent(businessKey, uniqueValue); 
            if (!result) { 
                return null; 
            } 
            redisTemplate.expire(businessKey, 100, TimeUnit.SECONDS); 
            return uniqueValue; 
        } catch (Exception ex) { 
            logger.error("獲取鎖失敗", ex); 
        } 
        return null; 
    } 
 
    public void unlockV2(String businessKey, String businessValue) { 
        try { 
            Object value = redisTemplate.opsForValue().get(businessKey); 
            if (value == null) { 
                return false; 
            } 
            //當前key在redis中value和當前線程手里持有的是否一致 
            if (!businessValue.equals(value)) { 
                //不一致，證明被其他線程獲取了 
                logger.info("key={}釋放鎖失敗嗎，該鎖已被其他線程獲取",businessKey); 
                return false; 
            } 
            redisTemplate.delete(businessKey); 
            logger.info("key={}釋放鎖成功",businessKey); 
        } catch (Exception ex) { 
            logger.error("釋放鎖失敗", ex); 
        } 
    } 
} 

這里比簡單版多了一個判斷，判斷持有鎖的線程是否為當前線程。盡管使用隨機字符串的 value來判斷是否為當前線程，但是在釋放鎖時(delete方法)，還是無法做到原子操作，比如進程 A 執行完業務邏輯，在準備釋放鎖時，恰好這時候進程 A 的鎖自動過期時間到了，而另一個進程 B 獲得鎖成功，然后 B 還沒來得及執行，進程 A 就執行了 delete(key) ，釋放了進程 B 的鎖.... ，因此需要配合 Lua 腳本釋放鎖。

setnx+Lua腳本Lua 是一種輕量小巧的腳本語言，用標準 C 語言編寫并以源代碼形式開放， 其設計目的是為了嵌入應用程序中，從而為應用程序提供靈活的擴展和定制功能。

Lua 提供了交互式編程模式。我們可以在命令行中輸入程序并立即查看效果。

lua腳本優點：

• 減少網絡開銷：原先多次請求的邏輯放在 redis 服務器上完成。使用腳本，減少了網絡往返時延
• 原子操作：Redis會將整個腳本作為一個整體執行，中間不會被其他命令插入(想象為事務)
• 復用：客戶端發送的腳本會永久存儲在Redis中，意味著其他客戶端可以復用這一腳本而不需要使用代碼完成同樣的邏輯

在resources目錄下創建一個redis-lock.lua文件。填入內容：

if redis.call('get', KEYS[1]) == ARGV[1] 
    then 
        return redis.call('del', KEYS[1]) 
    else 
        return 0 
end 

這段代碼的意思。就是通過key獲得其在redis中value，然后使用當前線程手里的value與之對比，一樣則刪除redis這個key。刪除返回1，否則返回0表示什么沒做。

Redis鎖代碼塊如下：

import org.slf4j.Logger; 
import org.slf4j.LoggerFactory; 
import org.springframework.core.io.ClassPathResource; 
import org.springframework.data.redis.core.RedisTemplate; 
import org.springframework.data.redis.core.script.DefaultRedisScript; 
import org.springframework.scripting.support.ResourceScriptSource; 
import org.springframework.stereotype.Component; 
 
import javax.annotation.PostConstruct; 
import javax.annotation.Resource; 
import java.util.ArrayList; 
import java.util.List; 
import java.util.UUID; 
import java.util.concurrent.TimeUnit; 
 
@Component 
public class DistributedLockV1 { 
 
    private Logger logger = LoggerFactory.getLogger(DistributedLockV1.class); 
 
    @Resource 
    private RedisTemplate redisTemplate; 
 
    private DefaultRedisScript<Long> script; 
 
    @PostConstruct 
    public void init() { 
        script = new DefaultRedisScript<Long>(); 
        script.setResultType(Long.class); 
        script.setScriptSource(new ResourceScriptSource(new ClassPathResource("redis-lock.lua"))); 
    } 
 
    public String lockV3(String key) { 
        String value = UUID.randomUUID().toString().replace("-", ""); 
 
        /* 
         * setIfAbsent <=> SET key value [NX] [XX] [EX <seconds>] [PX [millseconds]] 
         * set expire time 5 mins 
         */ 
        Boolean flag = redisTemplate.opsForValue().setIfAbsent(key, value, 10000, TimeUnit.MILLISECONDS); 
        if (flag) { 
            return value; 
        } 
        return null; 
    } 
 
    public void unlockV3(String key, String value) { 
        /** 業務邏輯處理完畢，釋放鎖 **/ 
        String lockValue = (String) redisTemplate.opsForValue().get(key); 
        if (lockValue != null && lockValue.equals(value)) { 
            System.out.println("lockValue========：" + lockValue); 
            List<String> keys = new ArrayList<>(); 
            keys.add(key); 
            Object execute = redisTemplate.execute(script, keys, lockValue); 
            System.out.println("execute執行結果，1表示執行del，0表示未執行 ===== " + execute); 
            logger.info("{} 解鎖成功，結束處理業務", key); 
            return; 
        } 
        logger.info("key={}釋放鎖失敗", key); 
    } 
} 

最后我們再次執行罰款扣減，日志輸出：

lockValue========：199740e62c184a6a9897f9c95e720b4d 
execute執行結果，1表示執行del，0表示未執行 ===== 1 
2021-03-09 19:03:51.592  INFO 6692 --- [nio-8080-exec-4] com.tian.user.lock.DistributedLockV1     : 1 解鎖成功，結束處理業務 

到此，setnx+Lua 這種方案我們已經實現了。如果對此有懷疑的，是好事，建議創建多個線程去調用罰款扣減這個service方法，看看器是否會出現問題。

「注意」

setnx在redis較低的版本里是沒有的，后面才引入的。其實我們也可以使用set命令來解決setnx，另外還可以加過期時間，整體命令為

set key value nx px xxx 

value 最好是隨機字符串，這樣可以防止業務代碼執行時間超過設置的鎖自動過期時間，而導致再次釋放鎖時出現釋放其他進程鎖的情況。

setnx 瑣最大的缺點就是它加鎖時只作用在一個 Redis 節點上，即使 Redis 通過 Sentinel(哨崗、哨兵) 保證高可用，如果這個 master 節點由于某些原因發生了主從切換，那么就會出現鎖丟失的情況，下面是個例子：

1. 在 Redis 的 master 節點上拿到了鎖;
2. 但是這個加鎖的 key 還沒有同步到 slave 節點;
3. master 故障，發生故障轉移，slave 節點升級為 master節點;
4. 上邊 master 節點上的鎖丟失。

有的時候甚至不單單是鎖丟失這么簡單，新選出來的 master 節點可以重新獲取同樣的鎖，出現一把鎖被拿兩次的場景。

由此可知，鎖被獲取兩次，肯定不能滿足安全性了。

盡管前兩種方案不是很如意，總是有些問題，但也有被部分企業采用，下面我們就來看基于Redis來實現分布式鎖更高級的版本。

高級版 Redisson + RedLock

Redisson 是 java 的 Redis 客戶端之一，是 Redis 官網推薦的 java 語言實現分布式鎖的項目。

Redisson 提供了一些 api 方便操作 Redis。因為本文主要以鎖為主，所以接下來我們主要關注鎖相關的類，以下是 Redisson 中提供的多樣化的鎖：

• 可重入鎖(Reentrant Lock)
• 公平鎖(Fair Lock)
• 聯鎖(MultiLock)
• 紅鎖(RedLock)
• 讀寫鎖(ReadWriteLock)
• 信號量(Semaphore) 等等

總之，管你了解不了解，反正 Redisson 就是提供了一堆鎖... 也是目前大部分公司使用 Redis 分布式鎖最常用的一種方式。

整體加鎖和解鎖的代碼結構如下：

RLock lock = redissonClient.getLock("xxx"); 
 
lock.lock(); 
 
try { 
    ... 
} finally { 
    lock.unlock(); 
} 

其實，加鎖和解鎖的磁層也是使用Lua腳本來實現的，有興趣的朋友可以去翻看一下器底層源碼。

由于篇幅問題，還涉及到Redis集群，所以這里就給出加鎖和解鎖的流程圖，僅供參考。

「加鎖過程」

 

「解鎖過程」

 

總結

通常我們為了實現 Redis 的高可用，一般都會搭建 Redis 的集群模式，比如給 Redis 節點掛載一個或多個 slave 從節點，然后采用哨兵模式進行主從切換。但由于 Redis 的主從模式是異步的，所以可能會在數據同步過程中，master 主節點宕機，slave 從節點來不及數據同步就被選舉為 master 主節點，從而導致數據丟失，大致過程如下：

1. 用戶在 Redis 的 master 主節點上獲取了鎖;
2. master 主節點宕機了，存儲鎖的 key 還沒有來得及同步到 slave 從節點上;
3. slave 從節點升級為 master 主節點;
4. 用戶從新的 master 主節點獲取到了對應同一個資源的鎖，同把鎖獲取兩次。

ok，然后為了解決這個問題，Redis 作者提出了 RedLock 算法，步驟如下(五步)：

在下面的示例中，我們假設有 5 個完全獨立的 Redis Master 節點，他們分別運行在 5 臺服務器中，可以保證他們不會同時宕機。

獲取當前 Unix 時間，以毫秒為單位。

依次嘗試從 N 個實例，使用相同的 key 和隨機值獲取鎖。在步驟 2，當向 Redis 設置鎖時，客戶端應該設置一個網絡連接和響應超時時間，這個超時時間應該小于鎖的失效時間。例如你的鎖自動失效時間為 10 秒，則超時時間應該在 5-50 毫秒之間。這樣可以避免服務器端 Redis 已經掛掉的情況下，客戶端還在死死地等待響應結果。如果服務器端沒有在規定時間內響應，客戶端應該盡快嘗試另外一個 Redis 實例。

客戶端使用當前時間減去開始獲取鎖時間(步驟 1 記錄的時間)就得到獲取鎖使用的時間。當且僅當從大多數(這里是 3 個節點)的 Redis 節點都取到鎖，并且使用的時間小于鎖失效時間時，鎖才算獲取成功。

如果取到了鎖，key 的真正有效時間等于有效時間減去獲取鎖所使用的時間(步驟 3 計算的結果)。

如果因為某些原因，獲取鎖失敗(沒有在至少 N/2+1 個Redis實例取到鎖或者取鎖時間已經超過了有效時間)，客戶端應該在所有的 Redis 實例上進行解鎖(即便某些 Redis 實例根本就沒有加鎖成功)。

到這，基本看出來，只要是大多數的 Redis 節點可以正常工作，就可以保證 Redlock 的正常工作。這樣就可以解決前面單點 Redis 的情況下我們討論的節點掛掉，由于異步通信，導致鎖失效的問題。

但是細想后， Redlock 還是存在如下問題：

假設一共有5個Redis節點：A, B, C, D, E。設想發生了如下的事件序列：

1. 客戶端1 成功鎖住了 A, B, C，獲取鎖成功(但 D 和 E 沒有鎖住)。
2. 節點 C 崩潰重啟了，但客戶端1在 C 上加的鎖沒有持久化下來，丟失了。
3. 節點 C 重啟后，客戶端2 鎖住了 C, D, E，獲取鎖成功。
4. 這樣，客戶端1 和 客戶端2 同時獲得了鎖(針對同一資源)。

哎，還是不能解決故障重啟后帶來的鎖的安全性問題...

針對節點重后引發的鎖失效問題，Redis 作者又提出了 延遲重啟 的概念，大致就是說，一個節點崩潰后，不要立刻重啟他，而是等到一定的時間后再重啟，等待的時間應該大于鎖的過期時間，采用這種方式，就可以保證這個節點在重啟前所參與的鎖都過期，聽上去感覺 延遲重啟 解決了這個問題...

但是，還是有個問題，節點重啟后，在等待的時間內，這個節點對外是不工作的。那么如果大多數節點都掛了，進入了等待，就會導致系統的不可用，因為系統在過期時間內任何鎖都無法加鎖成功。

總之，在 Redis 分布式鎖的實現上還有很多問題等待解決，我們需要認識到這些問題并清楚如何正確實現一個 Redis 分布式鎖，然后在工作中合理的選擇和正確的使用分布式鎖。

但為什么又說，有一部分企業采用Redis來實現分布式鎖呢?其實實現分布式鎖，從中間件上來選，也有 Zookeeper 可選，并且 Zookeeper 可靠性比 Redis 強太多，但是效率是低了點，如果并發量不是特別大，追求可靠性，那么肯定首選 Zookeeper。

關于分布式鎖的實現方案，沒有絕對的好與壞，沒有最好的方案，只有最適合你的業務的方案。

以下兩種方案僅供參考：

• 如果為了效率，建議使用Redis來實現;
• 如果追求可靠性，建議使用Zookeeper來實現。

本文轉載自微信公眾號「Java后端技術全?！梗梢酝ㄟ^以下二維碼關注。轉載本文請聯系Java后端技術全棧公眾號。