淺談緩存與分布式鎖（分布式緩存的分布式鎖lock是否會造成死鎖）
對于一個大型網站而言，每天的訪問量是巨大的，尤其遇到某些特定的時間點，比如電商平臺的購物節、教育平臺開學季。當在某個時間點遇到過量的并發時，往往會壓垮服務器導致網站崩潰，因此，網站對于高并發的處理是至關重要的，其中緩存起著舉足輕重的作用。對于一些不經常變化，或者熱度很高的數據，可以將其存入緩存，此時當用戶訪問時將直接讀取緩存而不查詢數據庫，從而大大提高了網站的吞吐量。
緩存的使用
首先來搭建一個簡單的測試環境，創建一個SpringBoot應用，并編寫一個控制器：
@RestController public class TestController { @Autowired private UserService userService; @GetMapping("/test") public List
1
2
3
4
5
6
7
8
9
10
11
訪問 http://localhost:8080/test 可以得到所有的用戶信息：
我們使用 jmeter 對該應用進行壓力測試，來到官網：http://jmeter.apache.org/download_jmeter.cgi
將zip壓縮包下載到本地，然后解壓縮，雙擊執行bin目錄下的 jmeter.bat 即可啟動jmeter：
這里模擬了1秒內2000次請求的并發，看看應用的吞吐量有多少：
發現吞吐量為421，可以想象當數據表中的數據量非常龐大時，若是所有的請求都需要查詢一次數據庫，那么效率就會大打折扣，所以，我們可以加入緩存來進行優化：
@RestController public class TestController { // 緩存 Map
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
這里使用HashMap簡答地模擬了一個緩存，那么接下來這個接口的執行過程如下所示：
當請求到來時，首先要從緩存中讀取數據，若是讀取到了數據，則直接返回；若是沒有讀取到，則查詢數據庫，并將得到的數據存入緩存，這樣下次請求就可以讀取到緩存中的數據了。
現在測試一下該應用的吞吐量：
不難發現，吞吐量得到了顯著的提升。
本地緩存與分布式緩存
剛才我們使用緩存提升了應用的整體性能，但緩存是被定義在應用內部的，這種緩存稱之為 本地緩存。本地緩存對于單機應用確實可以解決問題，但在分布式應用中，一個應用往往會被部署多份以實現高可用：
此時每份應用中都會保存一份自己的緩存，當修改數據時，相應地需要修改緩存中的數據，然而因為緩存有多份，這樣會導致其它的緩存沒有被修改，進而導致數據發生錯亂。
由此，我們需要將緩存抽取出去，形成一個獨立于所有應用，但又與所有應用有聯系的緩存中間件：
當前較為流行的緩存中間件就是 Redis 了。
SpringBoot整合Redis
接下來改造一下剛才的應用，讓其使用Redis緩存，首先下載redis的鏡像：
docker pull redis
1
創建目錄結構：
mkdir -p /mydata/redis/conf touch /mydata/redis/conf/redis.conf
1
2
來到/mydata/redis/conf目錄下，修改redis.conf文件：
appendonly yes # 持久化配置
1
創建redis的實例并啟動：
docker run -p 6379:6379 --name redis\ -v /mydata/redis/data:/data\ -v /mydata/redis/conf/redis.conf:/etc/redis/redis.conf\ -d redis redis-server /etc/redis/redis.conf
1
2
3
4
配置一下使redis隨著Docker的啟動而啟動：
docker update redis --restart=always
1
到這里Redis就準備好了，然后在項目中引入redis的依賴：
1
2
3
4
在application.yml中配置Redis：
spring: redis: host: 192.168.66.10
1
2
3
修改控制器代碼：
@RestController public class TestController { @Autowired private UserService userService; @Autowired private StringRedisTemplate redisTemplate; @GetMapping("/test") public String test() { // 從Redis中獲取數據 String usersJson = redisTemplate.opsForValue().get("users"); if (StringUtils.isEmpty(usersJson)) { // 未命中緩存，查詢數據庫 List
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
緩存中存在的一些問題
使用了Redis緩存并不是說就高枕無憂了，它仍然有很多的問題需要解決，以下是緩存中間件經常面臨的三個問題：
緩存穿透
緩存雪崩
緩存擊穿
緩存穿透
緩存穿透指的是查詢一個一定不存在的數據，由于緩存是不命中時需要從數據庫查詢，查不到數據則不寫入緩存，這將導致這個不存在的數據每次請求都要到數據庫去查詢，進而給數據庫帶來壓力。
因為緩存是幫助數據庫分擔壓力的，但若是讓某些人知道了系統中哪些數據是一定不存在的，那么它就可以利用這個數據不停地發送大量請求，從而擊垮我們的系統。
解決方案是不管這個數據是否存在，都對其進行存儲，比如某個請求需要的數據是不存在的，那么仍然將這個數據的key進行存儲，這樣下次請求時就可以從緩存中獲取，但若是每次請求數據的key均不同，那么Redis中就會存儲大量無用的key，所以應該為這些key設置一個指定的過期時間，到期自動刪除即可。
緩存雪崩
緩存雪崩是指緩存中數據大批量地同時過期，而查詢數據量巨大，引起數據庫壓力過大甚至宕機。
解決的辦法是在數據原有的過期時間上增加一個隨機值，這樣可以使數據之間的過期時間不一致，也就不會出現數據大批量同時過期的情況。
緩存擊穿
緩存擊穿是指熱點key在某個時間點過期的時候，而恰好在這個時間點對這個Key有大量的并發請求過來，從而大量的請求打到db。
解決的辦法是加鎖，當某個熱點key過期時，大量的請求會進行資源競爭，當某個請求成功執行時，其它請求就需要等待，此時該請求執行完成后就會將數據放入緩存，這樣別的請求就可以直接從緩存中獲取數據了。
解決緩存擊穿問題
對于緩存穿透和緩存雪崩，我們都能夠非常輕松地解決，然而緩存擊穿問題需要加鎖來解決，我們就來探究一下如何加鎖解決緩存擊穿問題。
@GetMapping("/test") public String test() { String usersJson = redisTemplate.opsForValue().get("users"); if (StringUtils.isEmpty(usersJson)) { synchronized (this){ // 再次確認緩存中是否有數據 String json = redisTemplate.opsForValue().get("users"); if(StringUtils.isEmpty(json)){ List
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
首先仍然需要從緩存中獲取數據，若未命中緩存，則執行同步代碼塊，在同步代碼塊中又進行了緩存數據的確認。這是因為當大量的請求同時進入了最外層的if語句中，此時某個請求開始執行，并成功查詢了數據庫，但是在該請求將數據放入Redis之后，如果不再次進行判斷，那么這些請求仍然還是會去查詢數據庫，其執行原理如下所示：
使用jmeter模擬1秒2000次的并發后，結果如下：
查詢了數據庫......
1
控制臺只輸出了一個 查詢了數據庫...... ，說明2000次的請求中確實只有一次查詢了數據庫，但隨之而來的是性能的急劇下降：
這種情況對于單機的應用是沒有問題的，因為SpringBoot中默認Bean是單例的，通過this鎖住代碼塊沒有任何問題，但在分布式應用中，一個應用往往被部署多份，this就無法鎖住每個應用的請求了，此時就需要使用 分布式鎖 。
分布式鎖
和緩存中間件一樣，我們可以將鎖抽取到外面，獨立于所有的服務，但又與每個服務聯系起來，如下所示：
每個服務想要加鎖，都需要去一個公共的地方進行占用，這樣就保證了即使在分布式的環境下，每個服務的鎖仍然是同一把，這個公共的地方可以有很多種選擇，可以使用Redis實現分布式鎖。
Redis中有一個指令非常適合實現分布式鎖，它就是 setnx ，來看看官網是如何介紹它的：
只有當key不存在的時候，setnx才會將值設置進去，否則什么也不做，那么對于每個服務，我們都可以讓其執行 setnx lock 1 ，因為這一操作是原子性的，即使有百萬的并發，也只能有一個請求設置成功，其它請求都會因為key已經存在而設置失敗。對于設置成功的，就表明占用鎖成功了；而設置失敗的，占用鎖也就失敗了。
代碼如下：
@RestController public class TestController { @Autowired private UserService userService; @Autowired private StringRedisTemplate redisTemplate; @GetMapping("/test") public String test() throws InterruptedException { String usersJson = redisTemplate.opsForValue().get("users"); if (StringUtils.isEmpty(usersJson)) { usersJson = getUsersJson(); } return usersJson; } public String getUsersJson() throws InterruptedException { String usersJson = ""; // 搶占分布式鎖 Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock", "1"); if (lock) { // 占鎖成功 // 再次確認緩存中是否有數據 String json = redisTemplate.opsForValue().get("users"); if (StringUtils.isEmpty(json)) { List
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
當然了，這里還是有很大問題的，如果在釋放鎖之前，程序就出現了異常，導致代碼終止，鎖沒有被及時釋放，就會出現死鎖問題，解決方案是在占用鎖的同時設置鎖的過期時間，這樣即使程序沒有及時釋放鎖，Redis也會等鎖過期后自動將其刪除。
即使設置了鎖的過期時間，仍然會有新的問題出現，當業務的執行時間大于了鎖的過期時間時，業務此時并沒有處理完成，但鎖卻被Redis刪除了，這樣別的請求就能夠重新占用鎖，并執行業務方法，解決方案是讓每個請求占用的鎖都是獨有的，某個請求不能隨意地去刪除其它請求的鎖，代碼如下：
public String getUsersJson() throws InterruptedException { String usersJson = ""; // 搶占分布式鎖 String uuid = UUID.randomUUID().toString(); Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock", uuid,300, TimeUnit.SECONDS); if (lock) { // 占鎖成功 // 再次確認緩存中是否有數據 String json = redisTemplate.opsForValue().get("users"); if (StringUtils.isEmpty(json)) { List
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
仔細想想，這里仍然是有問題存在的，因為在釋放鎖時，Java程序會向Redis發送指令，Redis執行完成后并將結果返回給Java程序，在網絡傳輸過程中都會消耗時間。假設此時Java程序向Redis獲取lock的值，Redis成功將值返回，但在返回過程中鎖過期了，此時別的請求將可以占有鎖，這時候Java程序接收到了lock的值，比較發現是自己的鎖，于是執行刪除操作，但此時Redis中的鎖已經是別的請求的鎖了，這樣還是出現了某個請求刪除了其它請求的鎖的問題。
為此，Redis官網也給出了解決方案：
通過執行這樣的一個Lua腳本即可解決剛才的問題，代碼如下：
public String getUsersJson() throws InterruptedException { String usersJson = ""; // 搶占分布式鎖 String uuid = UUID.randomUUID().toString(); Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock", uuid,300, TimeUnit.SECONDS); if (lock) { // 占鎖成功 // 再次確認緩存中是否有數據 String json = redisTemplate.opsForValue().get("users"); if (StringUtils.isEmpty(json)) { List
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
Redisson
Redisson是一個在Redis的基礎上實現的Java駐內存數據網格，我們可以使用它來輕松實現分布式鎖。
首先引入Redisson的依賴：
1
2
3
4
5
編寫配置類：
@Configuration public class MyRedissonConfig { @Bean public RedissonClient redissonClient() { Config config = new Config(); config.useSingleServer().setAddress("redis://192.168.66.10:6379"); return Redisson.create(config); } }
1
2
3
4
5
6
7
8
9
10
編寫一個控制器來體驗一下Redisson：
@RestController public class TestController { @Autowired private RedissonClient redissonClient; @GetMapping("/test") public String test() { // 占用鎖 RLock lock = redissonClient.getLock("my_lock"); // 加鎖 lock.lock(); try { // 模擬業務處理 Thread.sleep(1000 * 10); } catch (Exception e) { e.printStackTrace(); } finally { // 釋放鎖 lock.unlock(); } return "test"; } }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
就簡簡單單地聲明一下加鎖和釋放鎖操作即可，前面的所有問題都將迎刃而解，Redisson會自動為鎖設置過期時間，并且提供了一個監控鎖的看門狗，它的作用是在Redisson實例被關閉之前，不斷地延長鎖的過期時間，如果該鎖的線程還沒有處理完業務的話（默認情況下看門狗的續期時間為30秒）。
也可以指定鎖的過期時間：
lock.lock(15, TimeUnit.SECONDS);
1
在加鎖時設置好時間即可。
當設置了鎖的過期時間為15秒，若是業務執行耗時不止15秒，還會出現Redis自動刪除了鎖，別的請求搶占鎖的情況嗎？其實這種情況還是會有的，所以我們應該避免設置過小的過期時間，一定要讓鎖的過期時間大于業務的執行時間。
使用Redisson也能輕松實現讀寫鎖，比如：
@RestController public class TestController { @Autowired private StringRedisTemplate redisTemplate; @Autowired private RedissonClient redissonClient; @GetMapping("/write") public String write() { RReadWriteLock wrLock = redissonClient.getReadWriteLock("wr_lock"); // 獲取寫鎖 RLock wLock = wrLock.writeLock(); // 加鎖 wLock.lock(); String uuid = ""; try { uuid = UUID.randomUUID().toString(); Thread.sleep(20 * 1000); // 存入redis redisTemplate.opsForValue().set("uuid", uuid); } catch (InterruptedException e) { e.printStackTrace(); } finally { // 釋放鎖 wLock.unlock(); } return uuid; } @GetMapping("/read") public String read() { RReadWriteLock wrLock = redissonClient.getReadWriteLock("wr_lock"); // 獲取讀鎖 RLock rLock = wrLock.readLock(); // 加鎖 rLock.lock(); String uuid = ""; try { // 讀取uuid uuid = redisTemplate.opsForValue().get("uuid"); } finally { // 釋放鎖 rLock.unlock(); } return uuid; } }
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
48
只要讀寫鎖使用的是同一把鎖，那么在寫操作時，讀操作就必須等待，而且寫鎖是一個互斥鎖，當某個線程正在進行寫操作時，其它線程就必須排隊等待；讀寫是一個共享鎖，所有線程都可以直接進行讀操作，這樣便能夠保證每次讀取到的都是最新數據。
緩存一致性
使用緩存雖然提高了系統的吞吐量，但也隨之帶來了一個問題，當緩存中有了數據之后，都會從緩存中直接取出數據，但若是此時數據庫中的數據被修改了，用戶讀取到的仍然還是緩存中的數據，這就出現了數據不一致的問題，對于這一情況，一般有兩種解決方案：
雙寫模式：在修改數據庫的同時也去修改一下緩存
失效模式：在修改數據庫之后直接將緩存刪除
雙寫模式會導致臟數據問題，如下所示：
管理員A、B在修改一個商品的價格，管理員A先提交，管理員B后提交，按理應該是管理員B的寫緩存操作生效，但由于網絡波動等未知情況，導致管理員A的寫緩存操作先生效后，而管理員B的寫緩存操作后生效，最后緩存中的數據就變為了2000，這樣就導致了臟數據的產生，但這種臟數據只是暫時的，因為數據庫中的數據是正確的，所以等緩存過期后，重新查詢數據庫，緩存中的數據也就正常了。
問題轉化為如何保證雙寫模式下的數據一致性，解決辦法就是加鎖，對修改數據庫與修改緩存的操作加鎖，使其成為一個原子操作。
失效模式也是會導致臟數據產生的，所以對于經常修改的數據，應該直接查詢數據庫，而不是走緩存。
綜上所述，一般的解決方案為：對所有的緩存數據都需要設置過期時間，這樣可以使緩存在過期時觸發一次數據庫查詢從而更新緩存；讀寫數據的時候，使用Redisson添加讀寫鎖，保證寫操作的原子性。
Redis 分布式
版權聲明：本文內容由網絡用戶投稿，版權歸原作者所有，本站不擁有其著作權，亦不承擔相應法律責任。如果您發現本站中有涉嫌抄襲或描述失實的內容，請聯系我們jiasou666@gmail.com 處理，核實后本網站將在24小時內刪除侵權內容。
相關文章
