修正迷思：這才是SpringBoot Redis 分散式鎖的正確實作方式

我是碼哥，可以叫我靚仔。

在說分散式鎖之前，我們先說為什麼需要分散式鎖。

在單機部署的時候，我們可以使用Java 中提供的JUC 鎖機制來避免多執行緒同時操作一個共享變數產生的安全性問題。JUC 鎖定機制只能保證同一個JVM 進程中的相同時刻只有一個執行緒操作共享資源。

一個應用部署多個節點，多個進程如果要修改同一個共享資源，為了避免操作亂序導致的並發安全問題，這個時候就需要引入分散式鎖，分散式鎖就是用來控制同一時刻，只有一個JVM 進程中的一個執行緒可以存取被保護的資源。

分散式鎖很重要，然而許多公司的系統可能還在跑有缺陷的分散式鎖方案，其中不乏一些大型公司。

所以，碼哥今天分享一個正確Redis 分散式鎖定程式碼實戰，讓你一飛沖天，程式碼可直接用於生產，不是簡單的demo。

小提醒：如果你只想看程式碼實戰部分，可直接翻到SpringBoot 實戰章節。

錯誤的分散式鎖

在說正確方案之前，先來一個錯誤的，知道錯在哪，才能意識到如何寫正確。

在銀行工作的小白老師，使用Redis SET 指令實現加鎖，指令滿足了當key 不存在則設定value，同時設定超時時間，並且滿足原子語意。

SET lockKey 1 NX PX expireTime1.

lockKey 表示鎖的資源，value 設定成1。
NX：表示只有 lockKey 不存在的時候才能 SET 成功，從而確保只有一個客戶端可以獲得鎖定。
PX expireTime設定鎖的逾時時間，單位是毫秒；也可以使用 EX seconds以秒為單位設定逾時時間。

至於解鎖操作，小白老師果決的使用 DEL指令刪除。一個分散式鎖方案出來了，一氣呵成，組員不明覺厲，紛紛豎起大拇指，偽代碼如下。

//加锁成功
if（jedis.set(lockKey, 1, "NX", "EX", 10) == 1）{
  try {
      do work //执行业务

  } finally {
    //释放锁
     jedis.del(key);
  }
}1.
2.
3.
4.
5.
6.
7.
8.
9.
10.

然而，這是一個錯誤的分散式鎖定。問題在於解鎖的操作有可能出現釋放別人的鎖的情況。

有可能出現釋放別人的鎖的情況。

客戶端A 取得鎖成功，設定超時時間10 秒。
客戶端A 執行業務邏輯，但因為某些原因（網路問題、FullGC、程式碼垃圾效能差）執行很慢，時間超過10 秒，鎖因為逾時自動釋放了。
客戶端B 加鎖成功。
客戶端A 執行 DEL 釋放鎖，相當於釋放客戶端B 的鎖了。

原因很簡單：客戶端加鎖時，沒有設定一個唯一識別。釋放鎖的邏輯並不會檢查這把鎖的歸屬，直接刪除。

殘血版分散式鎖

小白老師：“碼哥，怎麼解決釋放別人的鎖的情況呢？”

解決方法：客戶端加鎖時設定一個“唯一標識”，可以讓value 儲存客戶端的唯一標識，例如隨機數、 UUID 等；釋放鎖時判斷鎖的唯一標識與客戶端的標識是否匹配，匹配才能刪除。

加鎖

SET lockKey randomValue NX PX 30001.

解鎖

刪除鎖定的時候判斷唯一識別是否符合偽代碼如下。

if (jedis.get(lockKey).equals(randomValue)) {
    jedis.del(lockKey);
}1.
2.
3.

加鎖、解鎖的偽代碼如下圖所示。

try (Jedis jedis = pool.getResource()) {
  //加锁成功
  if（jedis.set(lockKey, randomValue, "NX", "PX", 3000) == 1）{
    do work //执行业务
  }
} finally {
  //判断是不是当前线程加的锁,是才释放
  if (randomValue.equals(jedis.get(keylockKey {
     jedis.del(lockKey); //释放锁
  }
}1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.

到這裡，很多公司可能都是用這個方式來實現分散式鎖。

小白：“碼哥，還有問題。判斷鎖的唯一標識是否與當前客戶端匹配和刪除操作不是原子操作。”

聰明。這個方案還存在原子性問題，存在其他客戶端把鎖給釋放的問題。

客戶端A 執行唯一識別匹配成功，還來不及執行DEL釋放鎖定操作，鎖定過期被釋放。
客戶端B 取得鎖定成功，value 設定了自己的客戶端唯一識別。
客戶端A 繼續執行 DEL刪除鎖定操作，相當於把客戶端B 的鎖給刪了。

青銅版分散式鎖

雖然叫做青銅版，這也是我們最常用的分散式鎖方案之一了，這個版本沒有太大的硬傷，而且比較簡單。

小白老師：“碼哥，這如何是好，如何解決解鎖不是原子操作的問題？分佈式鎖這麼多門道，是我膚淺了。”

解決方案很簡單，解鎖的邏輯我們可以透過 Lua 腳本來實現判斷和刪除的過程。

KEYS[1]是lockKey。
ARGV[1] 表示客戶端的唯一識別requestId。

回傳 nil 表示鎖不存在，已經被刪除了。只有回傳值是1 才表示加鎖成功。

// key 不存在，返回 null
if (redis.call('exists', KEYS[1]) == 0) then
   return nil;
end;
// 获取 KEY[1] 中的 value 与 ARGV[1] 匹配，匹配则 del，返回 1。不匹配 return 0 解锁失败
if redis.call("get",KEYS[1]) == ARGV[1] then
    return redis.call("del",KEYS[1]);
else
    return 0;
end;1.
2.
3.
4.
5.
6.
7.
8.
9.
10.

使用上面的腳本，每個鎖都用一個隨機值作為唯一標識，當刪除鎖的客戶端的「唯一標識」與鎖的value 匹配的時候，才能執行刪除操作。這個方案已經相對完美，我們用的最多的可能就是這個方案了。

理論知識學完了，上實戰。

Spring Boot 環境準備

接下來碼哥，給你一個以Spring Boot 為基礎並且能用來生產實戰的程式碼。在上實戰程式碼之前，先把Spring Boot 整合Redis 的環境搞定。

新增依賴

程式碼基於Spring Boot 2.7.18 ，使用lettuce 用戶端來操作Redis。新增 spring-boot-starter-data-redis依賴。

<dependencyManagement>
    <dependencies>
        <dependency>
            <!-- Import dependency management from Spring Boot -->
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-dependencies</artifactId>
            <version>2.7.18</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.28</version>
            <optional>true</optional>
        </dependency>
    </dependencies>
</dependencyManagement>

<dependencies>
    <!--redis依赖-->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-redis</artifactId>
    </dependency>

    <!--Jackson依赖-->
    <dependency>
        <groupId>com.fasterxml.jackson.core</groupId>
        <artifactId>jackson-databind</artifactId>
    </dependency>
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <optional>true</optional>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.
40.
41.
42.

SpringBoot 配置

先配置yaml。

server:
  servlet:
    context-path: /redis
  port: 9011
spring:
  application:
    name: redis
  redis:
    host: 127.0.0.1
    port: 6379
    password: magebyte
    timeout: 6000
    client-type: lettuce
    lettuce:
      pool:
        max-active: 300
        max-idle: 100
        max-wait: 1000ms
        min-idle: 51.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.

RedisTemplate 預設序列化方式不具備可讀性，我們改下配置，使用JSON 序列化。注意了，這一步是附加操作，與分散式鎖沒有關係，是碼哥順帶給你的彩蛋。

@Configuration
public class RedisConfig {

    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory connectionFactory) {
        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<String, Object>();
        redisTemplate.setConnectionFactory(connectionFactory);

        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();

        redisTemplate.setKeySerializer(stringRedisSerializer); // key的序列化类型

        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
        ObjectMapper objectMapper = new ObjectMapper();
        objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        // 方法过期，改为下面代码
//        objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        objectMapper.activateDefaultTyping(LaissezFaireSubTypeValidator.instance,
                ObjectMapper.DefaultTyping.NON_FINAL, JsonTypeInfo.As.PROPERTY);
        jackson2JsonRedisSerializer.setObjectMapper(objectMapper);
        jackson2JsonRedisSerializer.setObjectMapper(objectMapper);

        redisTemplate.setValueSerializer(jackson2JsonRedisSerializer); // value的序列化类型
        redisTemplate.setHashKeySerializer(stringRedisSerializer);
        redisTemplate.setHashValueSerializer(jackson2JsonRedisSerializer);
        redisTemplate.afterPropertiesSet();
        return redisTemplate;
    }
}1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.

把分散式鎖介面定義出來，所謂面向介面和物件編程，程式碼如有神。順帶用英文顯擺下什麼叫做專業。

/**
 * 分布式锁
 */
public interface Lock {

    /**
     * Tries to acquire the lock with defined <code>leaseTime</code>.
     * Waits up to defined <code>waitTime</code> if necessary until the lock became available.
     * <p>
     * Lock will be released automatically after defined <code>leaseTime</code> interval.
     *
     * @param waitTime  the maximum time to acquire the lock
     * @param leaseTime lease time
     * @param unit      time unit
     * @return <code>true</code> if lock is successfully acquired,
     * otherwise <code>false</code> if lock is already set.
     * @throws InterruptedException - if the thread is interrupted
     */
    boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException;

    /**
     * Acquires the lock with defined <code>leaseTime</code>.
     * Waits if necessary until lock became available.
     * <p>
     * Lock will be released automatically after defined <code>leaseTime</code> interval.
     *
     * @param leaseTime the maximum time to hold the lock after it's acquisition,
     *                  if it hasn't already been released by invoking <code>unlock</code>.
     *                  If leaseTime is -1, hold the lock until explicitly unlocked.
     * @param unit      the time unit
     */
    void lock(long leaseTime, TimeUnit unit);

    /**
     * Releases the lock.
     *
     * <p><b>Implementation Considerations</b>
     *
     * <p>A {@code Lock} implementation will usually impose
     * restrictions on which thread can release a lock (typically only the
     * holder of the lock can release it) and may throw
     * an (unchecked) exception if the restriction is violated.
     * Any restrictions and the exception
     * type must be documented by that {@code Lock} implementation.
     */
    void unlock();
}1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.
40.
41.
42.
43.
44.
45.
46.
47.

青銅分散式鎖實戰

DistributedLock 實作 Lock 接口，建構方法實作 resourceName 和 StringRedisTemplate 的屬性設定。客戶端唯一識別使用uuid:threadId 組成。

分散式鎖

public class DistributedLock implements Lock {

    /**
     * 标识 id
     */
    private final String id = UUID.randomUUID().toString();

    /**
     * 资源名称
     */
    private final String resourceName;

    private final List<String> keys = new ArrayList<>(1);


    /**
     * redis 客户端
     */
    private final StringRedisTemplate redisTemplate;

    public DistributedLock(String resourceName, StringRedisTemplate redisTemplate) {
        this.resourceName = resourceName;
        this.redisTemplate = redisTemplate;
        keys.add(resourceName);
    }

    private String getRequestId(long threadId) {
        return id + ":" + threadId;
    }
}1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.

加鎖tryLock、lock

tryLock 以阻塞等待 waitTime 時間的方式來嘗試取得鎖。獲取成功則回傳true，反之false。tryAcquire 方法相當於執行了Redis 的SET resourceName uuid:threadID NX PX {leaseTime} 指令。

與 tryLock不同的是， lock 一直嘗試自旋阻塞等待取得分散式鎖，直到取得成功為止。而 tryLock 只會阻塞等待 waitTime 時間。

此外，為了讓程序更加健壯，碼哥實現了阻塞等待獲取分散式鎖，讓你用的更加開心，面試不慌加薪不難。如果你不需要自旋阻塞等待取得鎖，那把 while 程式碼區塊刪除即可。

@Override
public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
    long time = unit.toMillis(waitTime);
    long current = System.currentTimeMillis();
    long threadId = Thread.currentThread().getId();
    // 获取锁
    Boolean isAcquire = tryAcquire(leaseTime, unit, threadId);
    // lock acquired
    if (Boolean.TRUE.equals(isAcquire)) {
        return true;
    }

    time -= System.currentTimeMillis() - current;
    // 等待时间用完，获取锁失败
    if (time <= 0) {
        return false;
    }
    // 自旋获取锁
    while (true) {
        long currentTime = System.currentTimeMillis();
        isAcquire = tryAcquire(leaseTime, unit, threadId);
        // lock acquired
        if (Boolean.TRUE.equals(isAcquire)) {
            return true;
        }

        time -= System.currentTimeMillis() - currentTime;
        if (time <= 0) {
            return false;
        }
    }
}

@Override
public void lock(long leaseTime, TimeUnit unit) {
    long threadId = Thread.currentThread().getId();
    Boolean acquired;
    do {
        acquired = tryAcquire(leaseTime, unit, threadId);
    } while (Boolean.TRUE.equals(acquired));
}

private Boolean tryAcquire(long leaseTime, TimeUnit unit, long threadId) {
    return redisTemplate.opsForValue().setIfAbsent(resourceName, getRequestId(threadId), leaseTime, unit);
}1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.
40.
41.
42.
43.
44.
45.

解鎖unlock

解鎖的邏輯是透過執行lua 腳本實現。

@Override
public void unlock() {
    long threadId = Thread.currentThread().getId();

    // 执行 lua 脚本
    DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>(LuaScript.unlockScript(), Long.class);
    Long opStatus = redisTemplate.execute(redisScript, keys, getRequestId(threadId));

    if (opStatus == null) {
        throw new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: "
                + id + " thread-id: " + threadId);
    }


}1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.

Lua腳本

其實這個腳本就是在講解青銅板分散式鎖原理的那段程式碼，具體邏輯已經解釋過，這裡就不再重複分析。

public class LuaScript {

    private LuaScript() {

    }

    /**
     * 分布式锁解锁脚本
     *
     * @return 当且仅当返回 `1`才表示加锁成功.
     */
    public static String unlockScript() {
        return "if (redis.call('exists', KEYS[1]) == 0) then " +
                "return nil;" +
                "end; "+
                "if redis.call('get',KEYS[1]) == ARGV[1] then" +
                "   return redis.call('del',KEYS[1]);" +
                "else" +
                "   return 0;" +
                "end;";
    }
}1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.

RedisLock客戶端

最後，還需要提供一個客戶端給方便使用。

@Component
public class RedisLockClient {

    @Autowired
    private StringRedisTemplate redisTemplate;


    public Lock getLock(String name) {
        return new DistributedLock(name, redisTemplate);
    }

}1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.

單元測試來一個。

@Slf4j
@SpringBootTest(classes = RedisApplication.class)
public class RedisLockTest {

    @Autowired
    private RedisLockClient redisLockClient;


    @Test
    public void testLockSuccess() throws InterruptedException {
        Lock lock = redisLockClient.getLock("order:pay");
        try {
            boolean isLock = lock.tryLock(10, 30, TimeUnit.SECONDS);
            if (!isLock) {
                log.warn("加锁失败");
                return;
            }
            TimeUnit.SECONDS.sleep(3);
            log.info("业务逻辑执行完成");
        } finally {
            lock.unlock();
        }

    }

}1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.

有兩個點要注意。

釋放鎖的程式碼一定要放在 finally{} 區塊中。否則一旦執行業務邏輯過程中拋出異常，程式就無法執行釋放鎖的流程。只能乾等鎖逾時釋放。
加鎖的程式碼應該寫在try {} 程式碼中，放在try 外面的話，如果執行加鎖異常（客戶端網路連線逾時），但是實際指令已經傳送到服務端並執行，就會導致沒有機會執行解鎖的代碼。

小白：“碼哥，這個方案你管它叫青銅級別而已，這麼說還有王者、超神版？我們公司還用錯誤版分佈式鎖，難怪有時候出現重複訂單，是我膚淺了。”

趕緊將這個方案替換原來的錯誤或是殘血版的Redis 分散式鎖吧。

新聞