在公司的項目中用到了分布式鎖，但只會用卻不明白其中的規則所以寫一篇文章來記錄使用場景：交易服務，使用redis分布式鎖，防止重復提交訂單，出現超賣問題

分布式鎖的實現方式

redis是如何實現加鎖的？

在redis中，有一條命令，實現鎖

SETNX key value

該命令的作用是將 key 的值設為 value ，當且僅當 key 不存在。若給定的 key 已經存在，則 SETNX 不做任何動作。設置成功，返回 1 ；設置失敗，返回 0

使用 redis 來實現鎖的邏輯就是這樣的

線程 1 獲取鎖 -- > setnx lockKey lockvalue -- > 1 獲取鎖成功線程 2 獲取鎖 -- > setnx lockKey lockvalue -- > 0 獲取鎖失敗 （繼續等待，或者其他邏輯）線程 1 釋放鎖 -- > 線程 2 獲取鎖 -- > setnx lockKey lockvalue -- > 1 獲取成功

接下來我們將基于springboot實現redis分布式鎖

1. 引入redis、springmvc、lombok依賴

<?xml version='1.0' encoding='UTF-8'?><project xmlns='http://maven.apache.org/POM/4.0.0' xmlns:xsi='http://www.w3.org/2001/XMLSchema-instance' xsi:schemaLocation='http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd'> <modelVersion>4.0.0</modelVersion> <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>2.4.0</version> <relativePath/> <!-- lookup parent from repository --> </parent> <groupId>cn.miao.redis</groupId> <artifactId>springboot-caffeine-demo</artifactId> <version>0.0.1-SNAPSHOT</version> <name>springboot-redis-lock-demo</name> <description>Demo project for Redis Distribute Lock</description> <properties> <java.version>1.8</java.version> </properties> <dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter</artifactId> </dependency> <!--redis--> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-redis</artifactId> <version>2.1.4.RELEASE</version> </dependency> <!--springMvc--> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> <version>2.3.3.RELEASE</version> </dependency> <!-- https://mvnrepository.com/artifact/org.projectlombok/lombok --> <dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> <version>1.18.12</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <scope>test</scope> </dependency> </dependencies> <build> <plugins> <plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId> </plugin> </plugins> </build></project>

2. 新建RedisDistributedLock.java并書寫加鎖解鎖邏輯

import lombok.extern.slf4j.Slf4j;import org.springframework.data.redis.connection.RedisStringCommands;import org.springframework.data.redis.connection.ReturnType;import org.springframework.data.redis.core.RedisCallback;import org.springframework.data.redis.core.RedisTemplate;import org.springframework.data.redis.core.types.Expiration;import java.nio.charset.StandardCharsets;/** * @author miao * redis 加鎖工具類 */@Slf4jpublic class RedisDistributedLock { /** * 超時時間 */ private static final long TIMEOUT_MILLIS = 15000; /** * 重試次數 */ private static final int RETRY_TIMES = 10; /*** * 睡眠時間 */ private static final long SLEEP_MILLIS = 500; /** * 用來加鎖的lua腳本 * 因為新版的redis加鎖操作已經為原子性操作 * 所以放棄使用lua腳本 */ private static final String LOCK_LUA = 'if redis.call('setnx',KEYS[1],ARGV[1]) == 1 ' + 'then ' + ' return redis.call(’expire’,KEYS[1],ARGV[2]) ' + 'else ' + ' return 0 ' + 'end'; /** * 用來釋放分布式鎖的lua腳本 * 如果redis.get(KEYS[1]) == ARGV[1],則redis delete KEYS[1] * 否則返回0 * KEYS[1] , ARGV[1] 是參數，我們只調用的時候 傳遞這兩個參數就可以了 * KEYS[1] 主要用?澩?菰?edis 中用作key值的參數 * ARGV[1] 主要用來傳遞在redis中用做 value值的參數 */ private static final String UNLOCK_LUA = 'if redis.call('get',KEYS[1]) == ARGV[1] ' + 'then ' + ' return redis.call('del',KEYS[1]) ' + 'else ' + ' return 0 ' + 'end '; /** * 檢查 redisKey 是否上鎖 * * @param redisKey redisKey * @param template template * @return Boolean */ public static Boolean isLock(String redisKey, String value, RedisTemplate<Object, Object> template) { return lock(redisKey, value, template, RETRY_TIMES); } private static Boolean lock(String redisKey,String value,RedisTemplate<Object, Object> template,int retryTimes) { boolean result = lockKey(redisKey, value, template); while (!(result) && retryTimes-- > 0) { try {log.debug('lock failed, retrying...{}', retryTimes);Thread.sleep(RedisDistributedLock.SLEEP_MILLIS); } catch (InterruptedException e) {return false; } result = lockKey(redisKey, value, template); } return result; } private static Boolean lockKey(final String key, final String value, RedisTemplate<Object, Object> template) { try { RedisCallback<Boolean> callback = (connection) -> connection.set( key.getBytes(StandardCharsets.UTF_8), value.getBytes(StandardCharsets.UTF_8), Expiration.milliseconds(RedisDistributedLock.TIMEOUT_MILLIS), RedisStringCommands.SetOption.SET_IF_ABSENT ); return template.execute(callback); } catch (Exception e) { log.info('lock key fail because of ', e); } return false; } /** * 釋放分布式鎖資源 * * @param redisKey key * @param value value * @param template redis * @return Boolean */ public static Boolean releaseLock(String redisKey, String value, RedisTemplate<Object, Object> template) { try { RedisCallback<Boolean> callback = (connection) -> connection.eval( UNLOCK_LUA.getBytes(), ReturnType.BOOLEAN, 1, redisKey.getBytes(StandardCharsets.UTF_8), value.getBytes(StandardCharsets.UTF_8) ); return template.execute(callback); } catch (Exception e) { log.info('release lock fail because of ', e); } return false; }}

補充：1. spring-data-redis 有StringRedisTempla和RedisTemplate兩種，但是我選擇了RedisTemplate，因為他比較萬能。他們的區別是：當你的redis數據庫里面本來存的是字符串數據或者你要存取的數據就是字符串類型數據的時候，那么你就使用StringRedisTemplate即可， 但是如果你的數據是復雜的對象類型，而取出的時候又不想做任何的數據轉換，直接從Redis里面取出一個對象，那么使用RedisTemplate是 更好的選擇。2. 選擇lua腳本是因為，腳本運行是原子性的，在腳本運行期間沒有客戶端可以操作，所以在釋放鎖的時候用了lua腳本，而redis最新版加鎖時保證了Redis值和自動過期時間的原子性，所用沒用lua腳本

3. 創建測試類 TestController

import lombok.extern.slf4j.Slf4j;import org.springframework.data.redis.core.RedisTemplate;import org.springframework.web.bind.annotation.PostMapping;import org.springframework.web.bind.annotation.RestController;import javax.annotation.Resource;/** * @author miao */@RestController@Slf4jpublic class TestController { @Resource private RedisTemplate<Object, Object> redisTemplate; @PostMapping('/order') public String createOrder() throws InterruptedException { log.info('開始創建訂單'); Boolean isLock = RedisDistributedLock.isLock('testLock', '456789', redisTemplate); if (!isLock) { log.info('鎖已經被占用'); return 'fail'; } else { //.....處理邏輯 } Thread.sleep(10000); //一定要記得釋放鎖，否則會出現問題 RedisDistributedLock.releaseLock('testLock', '456789', redisTemplate); return 'success'; }}

4. 使用postman進行測試

5. redis分布式鎖的缺點

上面我們說的是redis，是單點的情況。如果是在redis sentinel集群中情況就有所不同了。在redis sentinel集群中，我們具有多臺redis，他們之間有著主從的關系，例如一主二從。我們的set命令對應的數據寫到主庫，然后同步到從庫。當我們申請一個鎖的時候，對應就是一條命令 setnx mykey myvalue ，在redis sentinel集群中，這條命令先是落到了主庫。假設這時主庫down了，而這條數據還沒來得及同步到從庫，sentinel將從庫中的一臺選舉為主庫了。這時，我們的新主庫中并沒有mykey這條數據，若此時另外一個client執行 setnx mykey hisvalue , 也會成功，即也能得到鎖。這就意味著，此時有兩個client獲得了鎖。這不是我們希望看到的，雖然這個情況發生的記錄很小，只會在主從failover的時候才會發生，大多數情況下、大多數系統都可以容忍，但是不是所有的系統都能容忍這種瑕疵。

6.redis分布式鎖的優化

為了解決故障轉移情況下的缺陷，Antirez 發明了 Redlock 算法，使用redlock算法，需要多個redis實例，加鎖的時候，它會想多半節點發送 setex mykey myvalue 命令，只要過半節點成功了，那么就算加鎖成功了。釋放鎖的時候需要想所有節點發送del命令。這是一種基于【大多數都同意】的一種機制。感興趣的可以查詢相關資料。在實際工作中使用的時候，我們可以選擇已有的開源實現，python有redlock-py，java 中有Redisson redlock。

redlock確實解決了上面所說的“不靠譜的情況”。但是，它解決問題的同時，也帶來了代價。你需要多個redis實例，你需要引入新的庫 代碼也得調整，性能上也會有損。所以，果然是不存在“完美的解決方案”，我們更需要的是能夠根據實際的情況和條件把問題解決了就好。

我大致講清楚了redis分布式鎖方面的問題（日后如果有新的領悟就繼續更新）。更多相關springboot redis分布式鎖內容請搜索好吧啦網以前的文章或繼續瀏覽下面的相關文章希望大家以后多多支持好吧啦網！