集群環境中使用Redis實現分布式鎖兩種方式

一、介紹

互聯網的應用場景中，為了支持高並發的請求，服務都是執行的分布式部署，相同的任務可以在集群中不同的服務器上執行，並且現在的服務容器都是支持多線程，相同的任務也可能會被同一個容器多次執行，都要求執行結果都滿足冪等性的設計原則。

分布式鎖，就是為了確保在分布式的環境下，相同任務只會執行成功的執行一次，后續的執行不會對這些已經產生了變化的業務再次產生影響。

分布式鎖的實現有不少的方式，如：

1. 使用RDBMS數據庫本身的表鎖或行鎖特性；
2. 使用Redis做為分布式鎖；
3. 使用Zookeeper做為分布式鎖；

使用RDBMS數據庫做為鎖不是筆者要討論的范疇，因為其本身的特性，不太符合在高並發下鎖的應用場景，這里以Redis作為分布式鎖做為介紹。

二、Redis

Redis本身有一些命令組支持原子性的操作，如getset、setns，這些命令可以用於分布式鎖的場景中。

1、使用getset作為分布式鎖控制實現

getSet本身是支持原子性的，在寫入新值的同時會返回舊的值，用這個寫入新值並獲取舊值做為分布式鎖的控制實現，如果返回的值不為空，那就說明前面已經有其它線程（這里的其它線程可以指當前容器中當前服務的其它線程，也指由部署在其它服務器上的應用中的線程）修改了該值，則可以認為已經有線程在對該請求正在處理，因而可以放棄后面的處理邏輯。

可以將當前系統的時間作為分布式鎖key的值，后續其它線程的請求時，將其請求的時間與獲取到的鎖對應的key的舊值進行比較，比較是否已經超過了一定的時間控制閥值，如果超過了則可以認為（這里存在誤判的可能性，因為后續邏輯的數據處理，恰好超過了這個時間比較閥值，就會導致重復執行，因而這里時間控制閥值要設置的比較合理，另外也需要合適的熔斷機制用於保證）前面的交易處理失敗（如服務恰好在設置了用於分布式鎖的key后，立即就掛了，沒有執行到后面的刪除操作）導致用於分布式鎖的key沒有被刪除掉，可以繼續處理該請求的后續交易邏輯。

這個是非常輕量級的事務控制，不會對Redis產生外部事務（應用與Redis之間的交互事務），只是需要對Redis多進一次getset操作，流程圖如下：

​

其中藍色部分表示獲取鎖的邏輯。

Java代碼實現如下：

@Resource
private RedisTemplate<String, String> redisTemplate;
// 超時時間，以毫秒為單位
private final long timeout = 2000;

@Test
public void testLock() {
    // 用於判斷交易唯一性和合法性的Token，在交易執行之前先保存在服務端，
    // 並且下發給客戶端，客戶端會在執行交易之前把Token帶上，沒帶Token的
    // 請求、Token不存在的服務端的請求、Token不正確的請求都視為非法請求
    String token = "...";
    String key = MD5Util.md5Of32(token);
    String lockKey = new StringBuilder(key).append("_lock").toString();
    boolean isGetKey = getLock(lockKey);
    if (!isGetKey) {
        log.warn("當前交易正在被處理中");
    }
    boolean handleSuccess = false;
    try {
        log.info("處理交易開始");
        String storedToken = redisTemplate.opsForValue().get(key);
        // 判斷Token是否存在且合法
        if (!token.equals(storedToken)) {
            log.warn("指定的Token不存在.");
            return;
        }
        handleSuccess = true;
        log.info("處理交易結束");
    } catch (Exception e) {
        log.info("處理發生異常", e);
    } finally {
        if (handleSuccess) {
            // 限制了單個Token只能夠執行一筆記交易，因而執行成功后將其刪除
            List<String> keys = new ArrayList<String>();
            keys.add(key);// 限制了單個Token只能夠執行一筆記交易，因而執行成功后將其刪除
            keys.add(lockKey);// 用於表示鎖的key刪除，表示釋放掉鎖
            redisTemplate.delete(keys);
        } else {
            // 刪除用於鎖定的key
            redisTemplate.delete(lockKey);
        }
    }
}

/**
 * 原理是從redis中獲取到的lockKey的值是不是存在，如果不存在表示寫入的是當前值，表示鎖獲取成功；
 * 如果獲取到的值存在，再判斷是否已經超過了指定的期限，如果超過了指定的期限，則認為鎖獲取成功，否則認為鎖獲取失敗；
 * 
 * @param lockKey 用於獲取鎖定的key
 * @return true表示獲取到鎖，false表示未獲取到鎖
 */
public boolean getLock(String lockKey) {
    long now = System.currentTimeMillis();
    // Redis的GetSet返回的值必須是字符串，否則會拋異常，因而將其轉換為字符串
    String nowTime = String.valueOf(now);
    String oldTime = null;
    // 判斷用於鎖定的key是否已經被設置了值，如果被設置了值，則用於控制后續的處理邏輯不再進行
    if ((oldTime = redisTemplate.opsForValue().getAndSet(lockKey, nowTime)) != null) {
        // 檢查鎖lockKey的值是不是超過了設定的時間，如2秒鍾，沒有超過則返回，不繼續處理后續的任務；
        // 注：這個邏輯有個問題，就是客戶端在2秒鍾之內不停的重試，就永遠不會進入到后面的處理環節。
        // 不過針對正常的業務請求這個是可以約定的，針對非正常的請求，被攔截也很正常，所以這個問題不是問題。
        if (now - Long.parseLong(oldTime) < timeout) {
            return false;
        }
        return true;
    }
    return true;
}

2、使用setnx作為分布式鎖控制實現

setnx和getset的執行邏輯不同，getset是設置新值並返回舊值，setnx如果存在舊值時可以通過參數控制不設置值並返回0,不存在舊值時才設值並返回1。

二者處理流程上都是相同的，不同之處在於獲取鎖的實現，setnx的實現邏輯如下：

​

其中綠色部分為setnx獲取鎖的邏輯，這個和getset是不同的實現邏輯。

setnx獲取鎖的Java代碼實現如下：

public boolean getLock(String lockKey) {
    RedisConnection connection = redisTemplate.getConnectionFactory().getConnection();
    JedisCommands commands = (JedisCommands) connection.getNativeConnection();
    boolean con = false;
    do {
        long now = System.currentTimeMillis();
        // Redis的GetSet返回的值必須是字符串，否則會拋異常，因而將其轉換為字符串
        String nowTime = String.valueOf(now);
        con = false;
        // 返回1表示鎖獲取成功，返回0表示鎖取失敗
        String result = commands.set(lockKey, nowTime, "NX", "PX", expire);
        if ("1".equals(result)) {
            return true;
        } else {
            String oldTime = redisTemplate.opsForValue().get(lockKey);
            if (null != oldTime) {
                // 檢查鎖lockKey的值是不是超過了設定的時間，如2秒鍾，如果超過了則繼續嘗試獲取鎖，
                // 直到獲取到鎖，或者數據未超期時退出，循環判斷可以解決死鎖的問題
                if (now - Long.parseLong(oldTime) >= expire) {// 數據已經過期了
                    con = true;
                }
            }
        }
    } while (con);
    return false;
}