更新時間:2022-11-11 來源:黑馬程序員 瀏覽量:
相信很多同學都聽說過分布式鎖,但也僅僅停留在概念的理解上,這篇文章會從分布式鎖的應用場景講起,從實現(xiàn)的角度上深度剖析redis如何實現(xiàn)分布式鎖。
一、超賣問題
我們先來看超賣的概念:
當寶貝庫存接近0時,如果多個買家同時付款購買此寶貝,或者店鋪后臺在架數(shù)量大于倉庫實際數(shù)量,將會出現(xiàn)超賣現(xiàn)象。超賣現(xiàn)象本質(zhì)上就是買到了比倉庫中數(shù)量更多的寶貝。
> 本文主要解決超賣問題的第一種,同時多人購買寶貝時,造成超賣。
測試代碼
那么超賣問題是如何產(chǎn)生的呢?我們準備一段代碼進行測試:
@Autowired private StringRedisTemplate stringRedisTemplate; /** * 第一種實現(xiàn),進程內(nèi)就存在線程安全問題 * 可以只啟動一個進程測試 */ @RequestMapping("/deduct_stock1") public void deductStock1(){ String stock = stringRedisTemplate.opsForValue().get("stock"); int stockNum = Integer.parseInt(stock); if(stockNum > 0){ //設置庫存減1 int realStock = stockNum - 1; stringRedisTemplate.opsForValue().set("stock",realStock + ""); System.out.println("設置庫存" + realStock); }else{ System.out.println("庫存不足"); } }
這段代碼中,使用redis先獲取庫存數(shù)量(當然實際場景中不會只保存一個全局庫存數(shù),應該根據(jù)每一個商品單元(sku)保存一份庫存數(shù))。
String stock = stringRedisTemplate.opsForValue().get("stock"); int stockNum = Integer.parseInt(stock);
接下來,判斷庫存數(shù)是否大于0:
- 如果大于0,將庫存數(shù)減一,通過set命令,寫回redis
>這里沒有使用redis的decrement命令,因為此命令在redis單線程模型下是線程安全的,而為了可以模擬線程不安全的情況將其拆成三步操作。
//設置庫存減1 int realStock = stockNum - 1; stringRedisTemplate.opsForValue().set("stock",realStock + ""); System.out.println("設置庫存" + realStock);
- 如果小于等于0,提示庫存不足
JMeter測試
通過JMeter進行并發(fā)測試,看下會不會出現(xiàn)超賣的問題:
1.啟動tomcat
這種情況下,只需要啟動一個tomcat就會出現(xiàn)超賣。我們先啟動一個tomcat在8080端口上。
2.下載JMeter
Apache JMeter是Apache組織開發(fā)的基于Java的壓力測試工具。
從官網(wǎng)上下載即可:
[https://jmeter.apache.org/download_jmeter.cgi](https://links.jianshu.com/go?to=https%3A%2F%2Fjmeter.apache.org%2Fdownload_jmeter.cgi)
下載完之后解壓,運行bin目錄下的jmeter.bat,顯示如下界面:
如果嫌字體太小,可以選擇放大:
3.配置JMeter
在Test Plan上點擊右鍵,創(chuàng)建`線程組(Thread Group)`
配置一下具體參數(shù):
- `Number of Threads` 同時并發(fā)線程數(shù)
- `Ramp-Up Period(in-seconds)` 代表隔多長時間執(zhí)行,0代表同時并發(fā)。假設線程數(shù)為100, 估計的點擊率為每秒10次, 那么估計的理想ramp-up period 就是 100/10 = 10 秒
- `Loop Count` 循環(huán)次數(shù)
> 這里給出500是為了直接測試并發(fā)500搶,看看能不能正好把500個貨物搶完。
添加Http請求:
添加請求URL:
添加聚合結果,用來顯示整體的運行情況:
到此為止JMeter的配置結束。
4.設置庫存量
啟動redis-server,使用redis-client連接:
把庫存數(shù)設置為500。
5.開始測試
點擊運行按鈕,啟動測試:
首先我們看到聚合報告里輸出的結果:
錯誤率0%,樣本數(shù)500,證明500個請求都已經(jīng)執(zhí)行,但是發(fā)現(xiàn)控制臺輸出如下:
<img src="assets/12.png" alt="img" style="zoom:67%;" />
很顯然,一份商品都被賣了多次,這顯然是不合理的。
原因分析
現(xiàn)在我們只啟動了一個tomcat,在單jvm進程的情況下,tomcat會使用線程池接收請求:
而由于每個線程可能同時獲取到庫存量,所以庫存量在兩個線程中顯示的都是500,然后兩個線程就繼續(xù)進行扣減庫存操作,得出499寫回redis中,在這個過程中,顯然存在線程安全的問題。同一個商品被賣出了2份,超賣問題就出現(xiàn)了。
二、加鎖優(yōu)化
synchronized鎖
要保證單jvm中線程安全,最簡單直接的方式就是添加synchronized關鍵字,那么這樣行不行呢,我們來做一個測試:
/** * 第二種實現(xiàn),使用synchronized加鎖 * 可以只啟動一個進程測試 */ @RequestMapping("/deduct_stock2") public void deductStock2(){ synchronized (this){ String stock = stringRedisTemplate.opsForValue().get("stock"); int stockNum = Integer.parseInt(stock); if(stockNum > 0){ //設置庫存減1 int realStock = stockNum - 1; stringRedisTemplate.opsForValue().set("stock",realStock + ""); System.out.println("設置庫存" + realStock); }else{ System.out.println("庫存不足"); } } }
在進行扣減庫存前,先通過synchronized關鍵字,對資源加鎖,這樣就只有一個線程能進入到扣減庫存的代碼塊中。來測試一下:
重置庫存
set stock 500
修改接口地址
測試
<img src="assets/15.png" alt="img" style="zoom:67%;" />
可以看到,庫存被扣減為0,并且沒有出現(xiàn)超賣的情況(設置了500庫存,并且500個人搶,正好搶完)。
但是這種方案顯然是不行的,在生產(chǎn)環(huán)境上如果部署多個tomcat實例,那么就會出現(xiàn)如下情況:
多個進程無法共享jvm內(nèi)存中的鎖,所以會出現(xiàn)多把鎖,這種情況下也會出現(xiàn)超賣問題。
三、分布式鎖的實現(xiàn)
多Tomcat實例下的超賣演示
接下來我們演示一下如何在多個Tomcat情況下,演示超賣的問題:
1.啟動兩個tomcat服務
在IDEA中配置兩個spring boot的啟動項,使用vm參數(shù)指定不同的端口號
```undefined
-Dserver.port=8080
```
2.配置nginx
編寫~/nginx_redis/conf/nginx.conf如下:
user nginx; worker_processes 1; error_log /var/log/nginx/error.log warn; pid /var/run/nginx.pid; events { worker_connections 1024; } http { include /etc/nginx/mime.types; default_type application/octet-stream; log_format main '$remote_addr - $remote_user [$time_local] "$request" ' '$status $body_bytes_sent "$http_referer" ' '"$http_user_agent" "$http_x_forwarded_for"'; upstream redislock{ server 192.168.226.1:8080 weight=1; server 192.168.226.1:8081 weight=1; } server { listen 80; server_name localhost; location /{ root html; proxy_pass http://redislock; } } access_log /var/log/nginx/access.log main; sendfile on; #tcp_nopush on; keepalive_timeout 65; #gzip on; include /etc/nginx/conf.d/*.conf; }
> 192.168.226.1這是我宿主機的IP
準備一個虛擬機(也可以使用windows下的nginx),使用docker啟動nginx:
docker pull nginx docker run -di -p 10085:80 --name nginx-redis-hc -v ~/nginx_redis/html:/usr/share/nginx/html -v ~/nginx_redis/conf/nginx.conf:/etc/nginx/nginx.conf -v ~/nginx_redis/logs:/var/log/nginx nginx
在宿主機下使用`虛擬機的IP地址:10085`訪問nginx,如果出現(xiàn)如下頁面就代表成功:
3.測試
修改接口地址為nginx:
運行查看兩個tomcat的控制臺:
- tomcat1
<img src="assets/21.png" alt="img" style="zoom:67%;" />
- tomcat2
<img src="assets/22.png" alt="img" style="zoom:67%;" />
沒有將庫存清空,證明存在超賣問題。
手動實現(xiàn)分布式鎖
使用redis手動實現(xiàn)分布式鎖,需要用到命令`setnx`。先來介紹一下setnx:
SETNX key value[]
> 可用版本: >= 1.0.0
>
> 時間復雜度: O(1)
只在鍵 `key` 不存在的情況下, 將鍵 `key` 的值設置為 `value` 。
若鍵 `key` 已經(jīng)存在, 則 `SETNX` 命令不做任何動作。
`SETNX` 是『SET if Not eXists』(如果不存在,則 SET)的簡寫。
返回值
命令在設置成功時返回 `1` , 設置失敗時返回 `0` 。
代碼示例
redis> EXISTS job # job 不存在 # job 不存在 (integer) 0 redis> SETNX job "programmer" # job 設置成功 (integer) 1 redis> SETNX job "code-farmer" # 嘗試覆蓋 job ,失敗 (integer) 0 redis> GET job # 沒有被覆蓋
使用redis構建分布式鎖流程如下:
image.png
- 線程1申請鎖(`setnx`),拿到了鎖。
- 線程2申請鎖,由于線程1已經(jīng)擁有了鎖,`setnx`返回0失敗,這一步用戶操作會失敗。
- 線程1執(zhí)行扣減庫存操作并釋放鎖。
- 線程2再次申請鎖,獲取到鎖并執(zhí)行扣減庫存,然后釋放鎖。
> 注意這里線程沒有拿到鎖,如果不嘗試while(true)重新獲取鎖,這個操作就直接失敗了。
代碼實現(xiàn)
/** * 第三種實現(xiàn),使用redis中的setIfAbsent(setnx命令)實現(xiàn)分布式鎖 */ @RequestMapping("/deduct_stock3") public void deductStock3(){ //在獲取到鎖的時候,給鎖分配一個id String opId = UUID.randomUUID().toString(); Boolean stockLock = stringRedisTemplate .opsForValue().setIfAbsent("stockLock", opId, Duration.ofSeconds(30); if(stockLock){ try{ String stock = stringRedisTemplate.opsForValue().get("stock"); int stockNum = Integer.parseInt(stock); if(stockNum > 0){ //設置庫存減1 int realStock = stockNum - 1; stringRedisTemplate.opsForValue().set("stock",realStock + ""); System.out.println("設置庫存" + realStock); }else{ System.out.println("庫存不足"); } }catch(Exception e){ e.printStackTrace(); }finally { if(opId.equals(stringRedisTemplate .opsForValue().get("stockLock"))){ stringRedisTemplate.delete("stockLock"); } } } }
測試略過,這里有幾個知識點需要說明
setIfAbsent設置超時
如果setIfAbsent不設置超時時間,假設線程執(zhí)行業(yè)務代碼時間時死鎖或者其他原因導致長時間不釋放,那么會影響其他線程獲取到鎖,這個時候整體業(yè)務就會出現(xiàn)不可用。
Boolean stockLock = stringRedisTemplate .opsForValue().setIfAbsent("stockLock", opId, Duration.ofSeconds(30);
設置超時時間為30秒,該時間一般大于業(yè)務執(zhí)行的最大時間。
每次獲取到鎖,設置唯一ID
考慮這樣的場景
- 線程1獲取鎖扣減庫存,但是由于操作不當,長時間卡住,這樣會觸發(fā)超時時間鎖被釋放。
- 線程2獲取到鎖,扣減庫存。
- 線程1的代碼拋出異常,執(zhí)行finally釋放鎖,但是釋放的是進程B的鎖。
解決方案就是在**加鎖前生成UUID**,釋放的時候校驗UUID是否正確,如果不正確,說明加鎖線程不是當前線程。
使用Redisson實現(xiàn)分布式鎖
setnx雖好,但是實現(xiàn)起來畢竟太過麻煩,一不小心就可能陷入并發(fā)編程的陷阱中,那么有沒有更加簡單的實現(xiàn)方式呢?答案就是`redisson`。
> Redisson是架設在[Redis](https://links.jianshu.com/go?to=http%3A%2F%2Fwww.oschina.net%2Fp%2Fredis)基礎上的一個Java駐內(nèi)存數(shù)據(jù)網(wǎng)格(In-Memory Data Grid)。【[Redis官方推薦](https://links.jianshu.com/go?to=http%3A%2F%2Fwww.redis.io%2Fclients)】
> Redisson在基于NIO的[Netty](https://links.jianshu.com/go?to=http%3A%2F%2Fnetty.io%2F)框架上,充分的利用了Redis鍵值數(shù)據(jù)庫提供的一系列優(yōu)勢,在Java實用工具包中常用接口的基礎上,為使用者提供了一系列具有分布式特性的常用工具類。使得原本作為協(xié)調(diào)單機多線程并發(fā)程序的工具包獲得了協(xié)調(diào)分布式多機多線程并發(fā)系統(tǒng)的能力,大大降低了設計和研發(fā)大規(guī)模分布式系統(tǒng)的難度。同時結合各富特色的分布式服務,更進一步簡化了分布式環(huán)境中程序相互之間的協(xié)作。
總而言之,`redisson`提供了一系列較為完善的工具類,其中就包含了分布式鎖。用`redisson`實現(xiàn)分布式鎖的流程極為簡單。
引入依賴
<dependency> <groupId>org.redisson</groupId> <artifactId>redisson</artifactId> <version>3.14.0</version> </dependency>
創(chuàng)建Redisson實例
@Bean public RedissonClient redisson(){ // 1. Create config object Config config = new Config(); config.useSingleServer().setAddress("redis://127.0.0.1:6379"); // config.useClusterServers() // // use "rediss://" for SSL connection // .addNodeAddress("redis://127.0.0.1:7181"); return Redisson.create(config); }
編寫分布式鎖代碼
@Autowired private RedissonClient redissonClient; /** * 第四種實現(xiàn),使用redisson實現(xiàn) */ @RequestMapping("/deduct_stock4") public void deductStock4(){ RLock lock = redissonClient.getLock("redisson:stockLock"); try{ //加鎖 lock.lock(); String stock = stringRedisTemplate.opsForValue().get("stock"); int stockNum = Integer.parseInt(stock); if(stockNum > 0){ //設置庫存減1 int realStock = stockNum - 1; stringRedisTemplate.opsForValue().set("stock",realStock + ""); System.out.println("設置庫存" + realStock); }else{ System.out.println("庫存不足"); } }catch(Exception e){ e.printStackTrace(); }finally { lock.unlock(); } }
其中加鎖代碼基本與進程內(nèi)加鎖一致,就不再詳細解讀,讀者自行實踐即可。
Redisson分布式鎖原理
`Redisson分布式鎖`的主要原理非常簡單,利用了lua腳本的原子性。
在分布式環(huán)境下產(chǎn)生并發(fā)問題的主要原因是三個操作并不是原子操作:
- 獲取庫存
- 扣減庫存
- 寫入庫存
那么如果我們把三個操作合并為一個操作,在默認單線程的Redis中運行,是不會產(chǎn)生并發(fā)問題的。源碼如下:
<T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) { internalLockLeaseTime = unit.toMillis(leaseTime); return evalWriteAsync(getName(), LongCodec.INSTANCE, command, "if (redis.call('exists', KEYS[1]) == 0) then " + "redis.call('hincrby', KEYS[1], ARGV[2], 1); " + "redis.call('pexpire', KEYS[1], ARGV[1]); " + "return nil; " + "end; " + "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " + "redis.call('hincrby', KEYS[1], ARGV[2], 1); " + "redis.call('pexpire', KEYS[1], ARGV[1]); " + "return nil; " + "end; " + "return redis.call('pttl', KEYS[1]);", Collections.singletonList(getName()), internalLockLeaseTime, getLockName(threadId)); }
這一段源碼中,`redisson`利用了lua腳本的原子性,校驗key是否存在,如果不存在就創(chuàng)建key并利用incrby加一操作(這步操作主要是為了實現(xiàn)可重入性)。`redisson`實現(xiàn)的分布式鎖具備如下特性:
- 鎖失效
- 鎖續(xù)租
> 執(zhí)行時間長的鎖快要到期時會自動續(xù)租
- 可重入
- 操作原子性
鎖續(xù)租原理
使用如下代碼進行測試鎖續(xù)租的情況
@Test void test() throws InterruptedException { RLock testlock1111 = redissonClient.getLock("testlock"); testlock1111.lock(); try{ Thread thread = new Thread(() -> { while(true){ Long testlock = redisTemplate.getExpire("testlock"); System.out.println(LocalDateTime.now().format(DateTimeFormatter.ISO_DATE_TIME) + " ttl:" + testlock); try { Thread.sleep(1000L); } catch (InterruptedException e) { e.printStackTrace(); } } }); thread.start(); thread.join(); }finally { if(testlock1111.isHeldByCurrentThread()){ testlock1111.unlock(); } } }
我們會發(fā)現(xiàn),每隔10秒會自動續(xù)租一次,保證鎖不被釋放。
<img src="assets/25.png" alt="image-20220721153251169" style="zoom:67%;" />
那么這種續(xù)租的行為是如何實現(xiàn)的呢?考慮這種情況:如果線程加鎖之后,進程宕機,線程無法執(zhí)行解鎖代碼,那么這個鎖就無法得到釋放(注意,不是加鎖線程不允許亂解鎖),為了避免這種情況的發(fā)生,鎖都會設置一個過期時間。比如使用**lock**無參命令會默認設置30秒的過期時間。那么30秒之后呢?如果線程還在工作,自動釋放依然會產(chǎn)生線程安全的問題。所以Redisson使用了watch dog看門狗機制來實現(xiàn)自動續(xù)租。
核心代碼及注釋:
private <T> RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) { RFuture<Long> ttlRemainingFuture; //lock()無參方法leaseTime為-1,所以進else分值 if (leaseTime > 0) { ttlRemainingFuture = tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG); } else { //通過lua腳本加鎖 ttlRemainingFuture = tryLockInnerAsync(waitTime, internalLockLeaseTime, TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG); } CompletionStage<Long> f = ttlRemainingFuture.thenApply(ttlRemaining -> { // 異步方法,等到加鎖成功會回調(diào),第一次加鎖ttlRemaining為空,leaseTime為-1 if (ttlRemaining == null) { if (leaseTime > 0) { internalLockLeaseTime = unit.toMillis(leaseTime); } else { //設置延時任務 scheduleExpirationRenewal(threadId); } } return ttlRemaining; }); return new CompletableFutureWrapper<>(f); }
接下來分析scheduleExpirationRenewal的過程:
private void renewExpiration() { ExpirationEntry ee = EXPIRATION_RENEWAL_MAP.get(getEntryName()); if (ee == null) { return; } //創(chuàng)建一個延遲任務 Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() { @Override public void run(Timeout timeout) throws Exception { ExpirationEntry ent = EXPIRATION_RENEWAL_MAP.get(getEntryName()); if (ent == null) { return; } Long threadId = ent.getFirstThreadId(); if (threadId == null) { return; } //執(zhí)行l(wèi)ua腳本進行續(xù)租 CompletionStage<Boolean> future = renewExpirationAsync(threadId); future.whenComplete((res, e) -> { if (e != null) { log.error("Can't update lock " + getRawName() + " expiration", e); EXPIRATION_RENEWAL_MAP.remove(getEntryName()); return; } //執(zhí)行l(wèi)ua續(xù)租,鎖還在就續(xù)租,鎖不在返回false就取消續(xù)租的行為 if (res) { // reschedule itself renewExpiration(); } else { cancelExpirationRenewal(null); } }); } }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS); //internalLockLeaseTime默認值30,所以每10秒會續(xù)租一次,續(xù)租到30秒 ee.setTimeout(task); }
其中,renewExpirationAsync執(zhí)行的lua腳本如下:
protected CompletionStage<Boolean> renewExpirationAsync(long threadId) { return evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN, "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " + "redis.call('pexpire', KEYS[1], ARGV[1]); " + "return 1; " + "end; " + "return 0;", Collections.singletonList(getRawName()), internalLockLeaseTime, getLockName(threadId)); }
判斷hash中是否存在鎖,如果存在就設置過期時間為30秒,返回1。如果不存在就返回0。
總結
本文介紹了超賣問題產(chǎn)生的原因:操作不具備原子性,同時提出了集中解決思路。
- `synchronized鎖`,無法保證多實例下的線程安全
- `setnx`手動實現(xiàn),坑很多、代碼較為復雜
- `redisson`實現(xiàn),能夠保證多實例下線程安全,代碼簡單可靠