緩存

在程序中，緩存是一個高速數據存儲層，其中存儲了數據子集，且通常是短暫性存儲，這樣日后再次請求此數據時，速度要比訪問數據的主存儲位置快。通過緩存，可以高效地重用之前檢索或計算的數據。

為什么要使用緩存

場景

在Java應用中，對于訪問頻率高，更新少的數據，通常的方案是將這類數據加入緩存中，相對從數據庫中讀取，讀緩存效率會有很大提升。

在集群環境下，常用的分布式緩存有Redis、Memcached等。但在某些業務場景上，可能不需要去搭建一套復雜的分布式緩存系統，在單機環境下，通常是會希望使用內部的緩存(LocalCache)。

Java內存緩存-通過Map定制簡單緩存 - 云+社區 - 騰訊云 (tencent.com)

本地緩存和分布式緩存

在本項目中，由于三級菜單的查詢是很常見的操作，有由于我們菜單的數據變動的機會很少，而且優化業務邏輯之后，吞吐量并沒有的到顯著的提高，此時我們就可以考慮使用緩存。

將我們的菜單數據，都放在緩存中，我們每次獲取三級菜單數據的時候，就可以減少對數據庫的訪問，主要是來減少I/O的消耗，我們每次訪問數據庫，都有需要建立TCP的連接的（雖然有線程池），但是TCP它是需要三次握手和四次揮手，依舊還是挺浪費我們的時間的。

基于此，我們要考慮使用緩存來保存我們的菜單數據，來減少性能的消耗。

緩存使用

為了系統的整體的提升，我們一般都會將部分數據放入到緩存中，加速訪問。而數據庫承擔數據落盤的工作。當然，緩存的使用，是需要有一定的條件。

那些數據適合做緩存來使用呢？

• 及時性、數據一致性要求不高的。
• 訪問量大且更新頻率不高的數據（讀多，寫少）

常用的應用場景：

• 電商類的應用，商品分類、商品列表等適合緩存并加一個失效時間（根據數據的更新的頻率），后臺如果發布一個商品，買家需要5分鐘才能看到新的商品一般還是可以接受的。

緩存的使用方式：

對應的偽代碼：

data = cache.loadd(id);//從緩存中讀取數據 if(data==null){ data = db.loadd(id);//從數據庫中加載 cache.put(id,data);//保存到緩存中 } return data;

需要注意的是：

在開發中，凡是放入緩存中的數據我們應該指定過期時間，使其可以在系統中即使沒有主動更新數據也能自動觸發數據加載進入緩存的流程。避免業務崩潰導致數據永久不一致的問題。

緩存的方式

使用Map<String,Object>。

Java 利用Map實現緩存 - zhoupan - 博客園 (cnblogs.com)

緩存工具類

package com.zsplat.yyzx.util;import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map;/*** 緩存機制*/ public class CacheUtil {private static CacheUtil cacheUtil;private static Map<String,Object> cacheMap;private CacheUtil(){cacheMap = new HashMap<String, Object>();}public static CacheUtil getInstance(){if (cacheUtil == null){cacheUtil = new CacheUtil();}return cacheUtil;}/*** 添加緩存* @param key* @param obj*/public void addCacheData(String key,Object obj){cacheMap.put(key,obj);}/*** 取出緩存* @param key* @return*/public Object getCacheData(String key){return cacheMap.get(key);}/*** 清楚緩存* @param key*/public void removeCacheData(String key){cacheMap.remove(key);} }

寫一個定時器 定時從數據庫里查出數據添加到緩存中

CacheUtil.getInstance().addCacheData("cacheYYZS100New", cacheMap);

取出緩存

Map<String , Object> cacheMap = (Map<String, Object>)CacheUtil.getInstance().getCacheData("cacheYYZS100New");存緩存和取緩存的方法，知道類型的話，強制轉換都沒問題

分頁 捋清楚startIndex和pageSize，count之間的關系就行 用截取list的方法實現分頁

if(StringUtils.isNotEmpty(startIndex)&&StringUtils.isNotEmpty(pageSize)){Integer start = Integer.parseInt(startIndex);Integer size = Integer.parseInt(pageSize);if(size*(start+1)<resultCount){dataList = dataList.subList(start*size,size*(start+1));}else if(start*size<resultCount){dataList = dataList.subList(start*size,resultCount);}else{dataList = new ArrayList<>();}}

在本項目中使用的流程圖：

在單體的項目中，這中方式確實能解決一定的問題，但是我們現在的項目都是部署在多臺的服務器上，也就是分布式，就是下面這種架構圖：

但是這樣會存在一個問題，假設現在有這一個場景，我們的商品服務，部署在3臺服務器中，用戶發起請求，負載均衡到我們的額一號服務器上，去獲取菜單數據，發現緩存為空，就去數據庫中查詢，查詢完之后，也給緩存中放一份，但是呢，又有其他人發起請求，負載均衡到了2號服務器，修改了菜單數據，又放在了2號服務器的緩存中，此時就會出現，1號服務器中的緩存數據和數據庫中就不一致了，那為什么呢？在我們的業務代碼中是判斷緩存有沒有，1號服務器之前已經查詢到數據了，并且放在自己的緩存中，那下一次在來查詢還是走一樣的緩存。此時就會出現數據庫和緩存中數據不一致的情況。

此時就需要針對分布式下如何使用緩存，來做進一步的思考了。

解決分布式下緩存和數據庫數據不一致的問題

當然針對于緩存，我們需要有一個共識：

• 緩存必須要有過期時間
• 保證數據庫跟緩存的最終一致性即可，不必追求強一致性

分布式下緩存的解決方案的架構圖：

在分布式下的緩存，我們應該使用一個中間件，將所有商品服務（當然還有很多的服務）的緩存都放到緩存的中間件Redis。

商品服務的部署的其他服務器，都給一個地方（緩存中間件）放數據。而不是放在自己的服務進程中，而是讓大家來共享這個緩存。

這個緩存中間件有很多，Memcache、Redis等等。

之所以選擇緩存中間件，是因為，如果我們后期的數據很大，我們的緩存中間件，放不下，我們可以對緩存中間件，搭集群，將數據分片存儲（也就是一人分一點）。理論上我們的緩存的容量的無限提升，打破了我們本地緩存的局限性。

整合Redis（作為我們各個微服務的緩存中間件）

整合的步驟：

導入依賴

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId> </dependency> <!--redis版本有管理-->

配置Redis服務器

spring:redis:host: 192.168.1.115port: 6379

如何使用Redis（單元測試）

//Redis實際上也是Map<key,value>@SpringBootTest @Slf4j class RedisTest {@AutowiredStringRedisTemplate stringRedisTemplate;/*** 測試stringRedisTemplate** @author wanglufei* @date 2022/5/11 9:27 PM*/@Testpublic void test01() {ValueOperations<String, String> ops = stringRedisTemplate.opsForValue();//存值的操作ops.set("uin", "niupi_" + UUID.randomUUID());//查詢操作System.out.println(ops.get("uin"));} }

接下來就使用Redis來保存的我們菜單數據。

改造三級分類的業務（使用Redis來存儲我們的菜單數據）

業務邏輯

@Override public Map<String, List<Catalog2Vo>> getCatalogJson() {//1.加入redis緩存 緩存中存的數據都是json數據格式//json數據跨平臺 兼容性好String catalogJson = stringRedisTemplate.opsForValue().get("catalogJson");if (StringUtils.isEmpty(catalogJson)) {//2.如果緩存中沒有的話 就去數據庫查詢Map<String, List<Catalog2Vo>> fromDB = getCatalogJsonFromDB();//將從數據庫查詢出來的數據轉為json放到緩存中stringRedisTemplate.opsForValue().set("catalogJson", JSON.toJSONString(fromDB));}//需要注意的是 給緩存中放的是json數據 但是我們要返回給前臺的是對象 所以還要轉換過來//其實這個操作也就是序列化和反序列化的過程Map<String, List<Catalog2Vo>> result = JSON.parseObject(catalogJson,new TypeReference<Map<String, List<Catalog2Vo>>>() {});return result; }

針對改造的三級分類壓測

編寫測試計劃

和上面的步驟類似。我直接將線程組的配置拉倒了5000一直循環，不到一會兒，就出現異常，JMeter卡死。

壓測過程出現的問題

高并發下緩存失效-緩存穿透

后臺服務也出現了連接超時的異常，出現的原因的是我們這一下估計好幾十w的請求，直接將我們的服務的打崩了。緩存直接給穿透了，將大量的請求落到我們的MySQL服務器上。

產生堆外內存溢出：OutOfDirectMemoryError

SpringBoot2.0之后默認使用Lettuce作為操作redis的客戶端，它使用netty進行網絡通信。主要是Lettuce的客戶端的問題導致的我們的堆外內存溢出。

解決方案：【性能調優】堆外內存溢出_種下星星的日子的博客-CSDN博客_lettuce 內存溢出

先來解釋一下，什么是緩存穿透？

緩存穿透，指查詢一個一定不存在的數據，由于緩存是不命中，將去查詢數據庫，但是數據庫也沒有此記錄，我們沒有將這次查詢的null寫入緩存，這將導致這個不存在的數據每次查詢都要到存儲層去查詢，失去了緩存的意義。

這樣的操作會有很大風險，容易被人利用不存在數據進行攻擊，數據庫瞬時壓力增大，最終導致崩潰。

我們知道，緩存的工作原理是先從緩存中獲取數據，如果有數據則直接返回給用戶，如果沒有數據則從慢速設備上讀取實際數據并且將數據放入緩存。同步緩存就像這樣：

高并發下緩存失效-緩存雪崩

緩存雪崩是指我們設置緩存時key采用了相同的過期時間，導致緩存在某一時刻同時失效，請求全部落到我們的數據庫上，數據庫瞬時壓力過重雪崩。

解決的辦法，在原有的失效時間基礎上增加一個隨機值，比如1-5分鐘的隨機，這樣每一個緩存的過期時間的重復率就會降低，就很難引發集體失效的事件。

高并發下緩存失效-緩存擊穿

對于一些設置了過期時間的key，如果這些key 可能會在某些時間點被超高并發地訪問，是一種非常熱點的數據。如果這個key在大量的請求同時進來前正好失效，那么所有對這個key的數據查詢都落到db，我們稱之為緩存擊穿。

解決的辦法：第一種是加鎖。大量并發只讓一個去查，其他人等待，查到以后釋放鎖，其他人獲取到鎖，先查緩存，就會有數據，不用去DB。

分布式下如何加鎖

本地鎖

本地鎖的意義是，在單進程的系統中，當存在多個線程可以同時改變某個變量（可變共享變量）時，就需要對變量或代碼塊做同步，使其在修改這種變量時能夠線性執行，以防止并發修改變量帶來數據不一致或者數據污染的現象。

而為了實現多個線程在一個時刻同一個代碼塊只能有一個線程可執行，那么需要在某個地方做個標記，這個標記必須每個線程都能看到，當標記不存在時可以設置該標記，其余后續線程發現已經有標記了則等待擁有標記的線程結束同步代碼塊取消標記后再去嘗試設置標記。這個標記可以理解為鎖。

常用的本地鎖

synchronized和lock。

本地鎖，對于普通同步方法，鎖是當前實例對象。它等同于同步方法塊，鎖是synchronized括號里的配置的對象。

對于靜態同步方法，鎖是當前類的Class類元信息，類似于字節碼。

使用本地鎖（在多線程的情況下）

所以在本項目中，使用本地鎖來控制資源，如果在我們分布式的環境下，會出現什么問題？

發現在每個服務，它都會去查詢數據庫，不是我們預想的結果，所以我們要考慮使用分布式鎖。

分布式鎖

https://zhuanlan.zhihu.com/p/42056183

如果是單機情況下（單JVM），線程之間共享內存，只要使用線程鎖就可以解決并發問題。

但如果是分布式情況下（多JVM），線程A和線程B很可能不是在同一JVM中，這樣線程鎖就無法起到作用了，這時候就要用到分布式鎖來解決。

分布式鎖是控制分布式系統同步訪問共享資源的一種方式。

在了解，什么是分布式鎖的基礎上，我們需要對鎖有一個概念。

什么情況下用鎖

在單進程的系統中，當存在多個線程可以同時改變某個變量（可變共享變量）時，就需要對變量或代碼塊做同步，使其在修改這種變量時能夠線性執行消除并發修改變量。

而同步的本質是通過鎖來實現的。為了實現多個線程在一個時刻同一個代碼塊只能有一個線程可執行，那么需要在某個地方做個標記，這個標記必須每個線程都能看到，當標記不存在時可以設置該標記，其余后續線程發現已經有標記了則等待擁有標記的線程結束同步代碼塊取消標記后再去嘗試設置標記。這個標記可以理解為鎖。

不同地方實現鎖的方式也不一樣，只要能滿足所有線程都能看得到標記即可。

如 Java 中 synchronized 是在對象頭設置標記，Lock 接口的實現類基本上都只是某一個 volitile 修飾的 int 型變量其保證每個線程都能擁有對該 int 的可見性和原子修改，linux 內核中也是利用互斥量或信號量等內存數據做標記。

除了利用內存數據做鎖其實任何互斥的都能做鎖（只考慮互斥情況），如流水表中流水號與時間結合做冪等校驗可以看作是一個不會釋放的鎖，或者使用某個文件是否存在作為鎖等。只需要滿足在對標記進行修改能保證原子性和內存可見性即可。

什么是分布式

分布式的 CAP 理論告訴我們:

任何一個分布式系統都無法同時滿足一致性（Consistency）、可用性（Availability）和分區容錯性（Partition tolerance），最多只能同時滿足兩項。

目前很多大型網站及應用都是分布式部署的，分布式場景中的數據一致性問題一直是一個比較重要的話題。

基于 CAP理論，很多系統在設計之初就要對這三者做出取舍。

在互聯網領域的絕大多數的場景中，都需要犧牲強一致性來換取系統的高可用性，系統往往只需要保證最終一致性。

分布式場景

此處主要指集群模式下，多個相同服務同時開啟.

在許多的場景中，我們為了保證數據的最終一致性，需要很多的技術方案來支持，比如分布式事務、分布式鎖等。

很多時候我們需要保證一個方法在同一時間內只能被同一個線程執行。在單機環境中，通過 Java 提供的并發 API 我們可以解決，但是在分布式環境下，就沒有那么簡單啦。

• 分布式與單機情況下最大的不同在于其不是多線程而是多進程。
• 多線程由于可以共享堆內存，因此可以簡單的采取內存作為標記存儲位置。而進程之間甚至可能都不在同一臺物理機上，因此需要將標記存儲在一個所有進程都能看到的地方。

什么是分布式鎖？

• 當在分布式模型下，數據只有一份（或有限制），此時需要利用鎖的技術控制某一時刻修改數據的進程數。
• 與單機模式下的鎖不僅需要保證進程可見，還需要考慮進程與鎖之間的網絡問題。（我覺得分布式情況下之所以問題變得復雜，主要就是需要考慮到網絡的延時和不可靠。。。一個大坑）
• 分布式鎖還是可以將標記存在內存，只是該內存不是某個進程分配的內存而是公共內存如 Redis、Memcache。至于利用數據庫、文件等做鎖與單機的實現是一樣的，只要保證標記能互斥就行。

我們需要怎樣的分布式鎖？

• 可以保證在分布式部署的應用集群中，同一個方法在同一時間只能被一臺機器上的一個線程執行。
• 這把鎖要是一把可重入鎖（避免死鎖）
• 這把鎖最好是一把阻塞鎖（根據業務需求考慮要不要這條）
• 這把鎖最好是一把公平鎖（根據業務需求考慮要不要這條）
• 有高可用的獲取鎖和釋放鎖功能
• 獲取鎖和釋放鎖的性能要好

數據庫實現分布式鎖

基于樂觀鎖

就是說特別樂觀，比如說每次去吃飯的時候，都認為窗口沒有人，只有到了吃飯的窗口才看有沒有人，如果有人則去別的地方吃飯。

就像系統認為數據的更新在大多數情況下是不會產生沖突的， 只在數據庫更新操作的提交的時候才對數據作沖突檢測。

如果檢測的結果出現了與預期數據不一致的情況，則返回失敗的信息。

樂觀鎖在大多數是基于數據版本（version）的記錄機制實現的。

為了更好的理解數據庫樂觀鎖在實際項目中的使用，下面舉一個典型的電商庫存的例子。

當用戶進行購買的時候就會對庫存進行操作（庫存減1代表已經賣出了一件）。

我們將這個庫存模型用下面的一張表optimistic_lock來表述，參考如下:

CREATE TABLE `optimistic_lock` ( `id` bigint(20) NOT NULL AUTO_INCREMENT, `resource` int(11) NOT NULL COMMENT '鎖定的資源', `version` int(11) NOT NULL COMMENT '版本信息', `created_time` datetime DEFAULT NULL COMMENT '創建時間', `updated_time` datetime DEFAULT NULL COMMENT '更新時間', `deleted_time` datetime DEFAULT NULL COMMENT '刪除時間', PRIMARY KEY (`id`), UNIQUE KEY `uiq_idx_resource` (`resource`) ) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8mb4 COMMENT='數據庫分布式鎖表';

其中：id表示主鍵；resource表示具體操作的資源，在這里也就是特指庫存；version表示版本號。

在使用樂觀鎖之前要確保表中有相應的數據，比如：

INSERT INTO `database1`.`optimistic_lock`(`id`, `resource`, `version`, `created_time`, `updated_time`, `deleted_time`) VALUES (1, 100, 1, '2020-04-10 22:05:52', '2020-04-10 22:05:52', '2020-04-10 22:05:52');

如果是單線程進行操作，數據庫本身就能保證操作的正確性。主要步驟如下：

• STEP1 - 獲取資源：SELECT resource FROM optimistic_lock WHERE id = 1
• STEP2 - 執行業務邏輯
• STEP3 - 更新資源：UPDATE optimistic_lock SET resource = resource -1 WHERE id = 1

然而大多數情況下是不會單線程的，要是單線程的話 ，公司豈不是要涼涼。

比如兩個以上的線程購買同一件商品，在數據庫層實際操作的時候應該是庫存（resource）減2，但是由于是高并發的情況，第一個線程執行之后（執行了STEP1、STEP2但是還沒有完成STEP3），第二個線程在購買相同的商品（執行STEP1），此時查詢出的庫存并沒有完成減1的動作，那么最終會導致2個線程購買的商品卻出現庫存只減1的情況。

在引入了version字段之后，那么具體的操作就會演變成下面的內容：

• STEP1 - 獲取資源： SELECT resource, version FROM optimistic_lock WHERE id= 1
• STEP2 - 執行業務邏輯
• STEP3 - 更新資源：UPDATE optimistic_lock SET resource = resource -1, version = version + 1 WHERE id = 1 AND version = oldVersion

其實，借助更新時間戳（updated_at）也可以實現樂觀鎖，和采用version字段的方式相似：

更新操作執行前線獲取記錄當前的更新時間，在提交更新時，檢測當前更新時間是否與更新開始時獲取的更新時間戳相等。

樂觀鎖的優點:

• 檢測數據沖突時不依賴數據庫庫本身的機制，所以不會影響請求的性能，當產生并發且并發量較小的時候只有少部分請求會失敗。

樂觀鎖的缺點：

• 對表的設計增加額外的字段，增加了數據庫的冗余，另外當并發量高的時候，version的值在頻繁變化，則會導致大量請求失敗，影響系統的可用性。

綜上所述，樂觀鎖比較適合并發量不高，并且寫操作不頻繁的場景

基于表主鍵唯一做分布式鎖

利用主鍵唯一的特性，如果有多個請求同時提交到數據庫的話，數據庫會保證只有一個操作可以成功，那么我們就可以認為操作成功的那個線程獲得了該方法的鎖，當方法執行完畢之后，想要釋放鎖的話，刪除這條數據庫記錄即可。

上面這種簡單的實現有以下幾個問題：

• 這把鎖強依賴數據庫的可用性，數據庫是一個單點，一旦數據庫掛掉，會導致業務系統不可用。
• 這把鎖沒有失效時間，一旦解鎖操作失敗，就會導致鎖記錄一直在數據庫中，其他線程無法再獲得到鎖。
• 這把鎖只能是非阻塞的，因為數據的 insert 操作，一旦插入失敗就會直接報錯。沒有獲得鎖的線程并不會進入排隊隊列，要想再次獲得鎖就要再次觸發獲得鎖操作。
• 這把鎖是非重入的，同一個線程在沒有釋放鎖之前無法再次獲得該鎖。因為數據中數據已經存在了。
• 這把鎖是非公平鎖，所有等待鎖的線程憑運氣去爭奪鎖。
• 在 MySQL 數據庫中采用主鍵沖突防重，在大并發情況下有可能會造成鎖表現象。

當然，我們也可以有其他方式解決上面的問題。

• 數據庫是單點？搞兩個數據庫，數據之前雙向同步，一旦掛掉快速切換到備庫上。
• 沒有失效時間？只要做一個定時任務，每隔一定時間把數據庫中的超時數據清理一遍。
• 非阻塞的？搞一個 while 循環，直到 insert 成功再返回成功。
• 非重入的？在數據庫表中加個字段，記錄當前獲得鎖的機器的主機信息和線程信息，那么下次再獲取鎖的時候先查詢數據庫，如果當前機器的主機信息和線程信息在數據庫可以查到的話，直接把鎖分配給他就可以了。
• 非公平的？再建一張中間表，將等待鎖的線程全記錄下來，并根據創建時間排序，只有最先創建的允許獲取鎖。
• 比較好的辦法是在程序中生產主鍵進行防重。

基于 Redis 做分布式鎖

流程圖：

/*** 使用分布式鎖* 1.使用setnx*/public Map<String, List<Catalog2Vo>> getCatalogJsonFromDBWithRedisSetnx() {/*** 怎么使用分布式鎖* 1.占坑*///Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent("lock", "111");/*** 5.刪除鎖直接刪除??? 如果由于業務時間很長，鎖自己過期了，我們 直接刪除，有可能把別人正在持有的鎖刪除了。* 解決辦法：占鎖的時候，值指定為uuid，每個人匹配是自己 的鎖才刪除。*/String token = UUID.randomUUID().toString();/*** 4.還有一種可能產生的問題，如果我們還沒來的及給鎖設置過期時間 就崩了 也會造成死鎖* 主要造成的原因是：我們的設置過期時間和加鎖 他不是一個原子性的操作* 在Redis總cli中有這樣一個命令：set lock 111 EX 300 NX* 意思就是set<lock,111> EX 過期時間300s 不存在才添加*/Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent("lock", token, 300, TimeUnit.SECONDS);if (lock) {System.out.println("獲取分布式鎖成功，執行業務中....");//2.加鎖成功 為了防止因為網路問題或者其他的問題 導致我們沒有釋放鎖 造成死鎖問題 我們需要設置鎖的過期時間stringRedisTemplate.expire("lock", 30, TimeUnit.SECONDS);Map<String, List<Catalog2Vo>> fromDB = null;try {fromDB = getFromDB();} finally {String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";Long result = stringRedisTemplate.execute(new DefaultRedisScript<Long>(script, Long.class), Arrays.asList("lock"), token);}//2.1 數據返回成功的我們還要解鎖//stringRedisTemplate.delete("lock");//我們需要根據獲取當前鎖的線程 獲取到它的value 來和一開始uuid來匹配 如果相等 說明在釋放自己的鎖//String value = stringRedisTemplate.opsForValue().get("lock");/*** 6.如果在比較的前面（也就是去查詢redis redis給我門返回數據的途中 我們redis中的數據過期了 那別人就有機會進來 重新占了個鎖）* 此時別人也叫lock，但是value，是不一樣的值 那我們此時 在走我們判斷的業務邏輯 進不去 就又造成了 死鎖* 造成的主要原因是：他們不是一個原子性的操作（獲取值和和值進行對比）* 解決辦法：刪除鎖必須保證原子性。使用redis+Lua腳本完成*///if (value.equals(token)) {//說明在釋放自己的鎖 那就放心釋放//stringRedisTemplate.delete("lock");//}//Lua腳本//String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call" +//"('del', KEYS[1]) else return 0 end";/*** 刪除成功返回 1 不成功0* Lua腳本解鎖保證原子性操作*///Integer result = stringRedisTemplate.execute(new DefaultRedisScript<Integer>(script,//Integer.class),//Arrays.asList("lock"), token);return fromDB;} else {//3.加鎖失敗（等一會兒，再去重試） 類似與自旋/*** 7.保證加鎖【占位+過期時間】和刪除鎖【判斷+刪除】的原子性。 更難的事情，鎖的自動續期* 也就是我們的業務還沒執行完 我們的鎖過期了 那不就bbq了。就好比 我們在網吧 我們正打團 機子給我們提示余額用完了* 給我們鎖機了。* 所以為了解決這個問題：我們需要解決在業務的執行期間 需要給鎖自動續期* 最簡單的方法就是過期時間 給多一點（合理的業務時間）使用try{} finally{}*/System.out.println("獲取分布式鎖失敗，自旋等待中....");//可以設置休眠100ms在重試try {Thread.sleep(300);} catch (InterruptedException e) {e.printStackTrace();}return getCatalogJsonFromDBWithRedisSetnx();}}

相關鏈接：https://zhuanlan.zhihu.com/p/42056183

基于 redis 的 setnx()、expire() 方法做分布式鎖

setnx()

setnx 的含義就是 SET if Not Exists，其主要有兩個參數 setnx(key, value)。該方法是原子的，如果 key 不存在，則設置當前 key 成功，返回 1；如果當前 key 已經存在，則設置當前 key 失敗，返回 0。

expire()

expire 設置過期時間，要注意的是 setnx 命令不能設置 key 的超時時間，只能通過 expire() 來對 key 設置。

使用步驟

1、setnx(lockkey, 1) 如果返回 0，則說明占位失敗；如果返回 1，則說明占位成功

2、expire() 命令對 lockkey 設置超時時間，為的是避免死鎖問題。

3、執行完業務代碼后，可以通過 delete 命令刪除 key。

這個方案其實是可以解決日常工作中的需求的，但從技術方案的探討上來說，可能還有一些可以完善的地方。比如，如果在第一步 setnx 執行成功后，在 expire() 命令執行成功前，發生了宕機的現象，那么就依然會出現死鎖的問題，所以如果要對其進行完善的話，可以使用 redis 的 setnx()、get() 和 getset() 方法來實現分布式鎖。

基于 redis 的 setnx()、get()、getset()方法做分布式鎖

這個方案的背景主要是在 setnx() 和 expire() 的方案上針對可能存在的死鎖問題，做了一些優化。

getset()

這個命令主要有兩個參數 getset(key，newValue)。該方法是原子的，對 key 設置 newValue 這個值，并且返回 key 原來的舊值。假設 key 原來是不存在的，那么多次執行這個命令，會出現下邊的效果：

getset(key, “value1”) 返回 null 此時 key 的值會被設置為 value1

getset(key, “value2”) 返回 value1 此時 key 的值會被設置為 value2

依次類推！

使用步驟

setnx(lockkey, 當前時間+過期超時時間)，如果返回 1，則獲取鎖成功；如果返回 0 則沒有獲取到鎖，轉向 2。

get(lockkey) 獲取值 oldExpireTime ，并將這個 value 值與當前的系統時間進行比較，如果小于當前系統時間，則認為這個鎖已經超時，可以允許別的請求重新獲取，轉向 3。

計算 newExpireTime = 當前時間+過期超時時間，然后 getset(lockkey, newExpireTime) 會返回當前 lockkey 的值currentExpireTime。

判斷 currentExpireTime 與 oldExpireTime 是否相等，如果相等，說明當前 getset 設置成功，獲取到了鎖。如果不相等，說明這個鎖又被別的請求獲取走了，那么當前請求可以直接返回失敗，或者繼續重試。

在獲取到鎖之后，當前線程可以開始自己的業務處理，當處理完畢后，比較自己的處理時間和對于鎖設置的超時時間，如果小于鎖設置的超時時間，則直接執行 delete 釋放鎖；如果大于鎖設置的超時時間，則不需要再鎖進行處理。

基于 Redlock 做分布式鎖

官方文檔：Redis SET 命令。 Distributed Locks with Redis | Redis

Distributed Locks with Redis

A Distributed Lock Pattern with Redis

一種帶有 Redis 的分布式鎖模式

Distributed locks are a very useful primitive in many environments where different processes must operate with shared resources in a mutually exclusive way.

在許多環境中，分布式鎖是非常有用的原語，在這些環境中，不同的進程必須以互斥的方式使用共享資源進行操作。

There are a number of libraries and blog posts describing how to implement a DLM (Distributed Lock Manager) with Redis, but every library uses a different approach, and many use a simple approach with lower guarantees compared to what can be achieved with slightly more complex designs.

有許多庫和博客文章描述了如何使用 Redis 實現 DLM (緩存同步/目錄) ，但是每個庫使用不同的方法，并且許多使用一種簡單的方法，與稍微復雜的設計相比，可以實現較低的保證。

This page describes a more canonical algorithm to implement distributed locks with Redis. We propose an algorithm, called Redlock, which implements a DLM which we believe to be safer than the vanilla single instance approach. We hope that the community will analyze it, provide feedback, and use it as a starting point for the implementations or more complex or alternative designs.

本頁描述了使用 Redis 實現分布式鎖的更規范的算法。我們提出了一個名為 Redlock 的算法，它實現了一個 DLM，我們相信它比普通的單實例方法更安全。我們希望社區能夠分析它，提供反饋，并將其作為實現或更復雜或可選設計的起點。

分布式鎖所需的保證

Safety property: Mutual exclusion. At any given moment, only one client can hold a lock.

安全特性: 互斥鎖。在任何時刻，只有一個客戶可以持有鎖

Liveness property A: Deadlock free. Eventually it is always possible to acquire a lock, even if the client that locked a resource crashes or gets partitioned. 活性屬性 a: 無死鎖。最終，即使鎖定資源的客戶機崩潰或分區，也總是有可能獲得鎖

Liveness property B: Fault tolerance. As long as the majority of Redis nodes are up, clients are able to acquire and release locks. 活性屬性 b: 容錯性。只要大多數 Redis 節點處于啟動狀態，客戶端就能夠獲取和釋放鎖

Redisson

redisson/redisson: Redisson - Redis Java client with features of In-Memory Data Grid. Over 50 Redis based Java objects and services: Set, Multimap, SortedSet, Map, List, Queue, Deque, Semaphore, Lock, AtomicLong, Map Reduce, Publish / Subscribe, Bloom filter, Spring Cache, Tomcat, Scheduler, JCache API, Hibernate, MyBatis, RPC, local cache … (github.com)

文檔：https://github.com/redisson/redisson/wiki/Table-of-Content

項目介紹中文文檔：https://github.com/redisson/redisson/wiki/Redisson%E9%A1%B9%E7%9B%AE%E4%BB%8B%E7%BB%8D

Redisson是架設在Redis基礎上的一個Java駐內存數據網格（In-Memory Data Grid）。充分的利用了Redis鍵值數據庫提供的一系列優勢，基于Java實用工具包中常用接口，為使用者提供了一系列具有分布式特性的常用工具類。使得原本作為協調單機多線程并發程序的工具包獲得了協調分布式多機多線程并發系統的能力，大大降低了設計和研發大規模分布式系統的難度。同時結合各富特色的分布式服務，更進一步簡化了分布式環境中程序相互之間的協作。

Redisson采用了基于NIO的Netty框架，不僅能作為Redis底層驅動客戶端，具備提供對Redis各種組態形式的連接功能，對Redis命令能以同步發送、異步形式發送、異步流形式發送或管道形式發送的功能，LUA腳本執行處理，以及處理返回結果的功能，還在此基礎上融入了更高級的應用方案，不但將原生的Redis Hash，List，Set，String，Geo，HyperLogLog等數據結構封裝為Java里大家最熟悉的映射（Map），列表（List），集（Set），通用對象桶（Object Bucket），地理空間對象桶（Geospatial Bucket），基數估計算法（HyperLogLog）等結構，在這基礎上還提供了分布式的多值映射（Multimap），本地緩存映射（LocalCachedMap），有序集（SortedSet），計分排序集（ScoredSortedSet），字典排序集（LexSortedSet），列隊（Queue），阻塞隊列（Blocking Queue），有界阻塞列隊（Bounded Blocking Queue），雙端隊列（Deque），阻塞雙端列隊（Blocking Deque），阻塞公平列隊（Blocking Fair Queue），延遲列隊（Delayed Queue），布隆過濾器（Bloom Filter），原子整長形（AtomicLong），原子雙精度浮點數（AtomicDouble），BitSet等Redis原本沒有的分布式數據結構。不僅如此，Redisson還實現了Redis文檔中提到像分布式鎖Lock這樣的更高階應用場景。事實上Redisson并沒有不止步于此，在分布式鎖的基礎上還提供了聯鎖（MultiLock），讀寫鎖（ReadWriteLock），公平鎖（Fair Lock），紅鎖（RedLock），信號量（Semaphore），可過期性信號量（PermitExpirableSemaphore）和閉鎖（CountDownLatch）這些實際當中對多線程高并發應用至關重要的基本部件。正是通過實現基于Redis的高階應用方案，使Redisson成為構建分布式系統的重要工具。

在提供這些工具的過程當中，Redisson廣泛的使用了承載于Redis訂閱發布功能之上的分布式話題（Topic）功能。使得即便是在復雜的分布式環境下，Redisson的各個實例仍然具有能夠保持相互溝通的能力。在以這為前提下，結合了自身獨有的功能完善的分布式工具，Redisson進而提供了像分布式遠程服務（Remote Service），分布式執行服務（Executor Service）和分布式調度任務服務（Scheduler Service）這樣適用于不同場景的分布式服務。使得Redisson成為了一個基于Redis的Java中間件（Middleware）。

Redisson Node的出現作為駐內存數據網格的重要特性之一，使Redisson能夠獨立作為一個任務處理節點，以系統服務的方式運行并自動加入Redisson集群，具備集群節點彈性增減的能力。然而在真正意義上讓Redisson發展成為一個完整的駐內存數據網格的，還是具有將基本上任何復雜、多維結構的對象都能變為分布式對象的分布式實時對象服務（Live Object Service），以及與之相結合的，在分布式環境中支持跨節點對象引用（Distributed Object Reference）的功能。這些特色功能使Redisson具備了在分布式環境中，為Java程序提供了堆外空間（Off-Heap Memory）儲存對象的能力。

Redisson提供了使用Redis的最簡單和最便捷的方法。Redisson的宗旨是促進使用者對Redis的關注分離（Separation of Concern），從而讓使用者能夠將精力更集中地放在處理業務邏輯上。如果您現在正在使用其他的Redis的Java客戶端，希望Redis命令和Redisson對象匹配列表 能夠幫助您輕松的將現有代碼遷徙到Redisson里來。如果目前Redis的應用場景還僅限于作為緩存使用，您也可以將Redisson輕松的整合到像Spring和Hibernate這樣的常用框架里。除此外您也可以間接的通過Java緩存標準規范JCache API (JSR-107)接口來使用Redisson。

Redisson生而具有的高性能，分布式特性和豐富的結構等特點恰巧與Tomcat這類服務程序對會話管理器（Session Manager）的要求相吻合。利用這樣的特點，Redisson專門為Tomcat提供了會話管理器（Tomcat Session Manager）。

在此不難看出，Redisson同其他Redis Java客戶端有著很大的區別，相比之下其他客戶端提供的功能還僅僅停留在作為數據庫驅動層面上，比如僅針對Redis提供連接方式，發送命令和處理返回結果等。像上面這些高層次的應用則只能依靠使用者自行實現。

Redisson支持Redis 2.8以上版本，支持Java1.6+以上版本。

Redisson的初次使用

導入依賴

<!--redisson--> <dependency><groupId>org.redisson</groupId><artifactId>redisson</artifactId><version>3.11.1</version> </dependency>

與第三方框架整合

官方文檔：https://github.com/redisson/redisson/wiki/14.-%E7%AC%AC%E4%B8%89%E6%96%B9%E6%A1%86%E6%9E%B6%E6%95%B4%E5%90%88

@Configuration public class MyRedissonConfig {/*** 所有對Redisson的使用都是通過RedissonClient對象來操作** @return org.redisson.api.RedissonClient* @author wanglufei* @date 2022/5/18 8:20 PM*/@Beanpublic RedissonClient redisson() {//1.創建配置對象Config config = new Config();//單集群模式config.useSingleServer().setAddress("redis://192.168.2.115:6379");//2.根據配置對象創建出RedissonClient實例對象RedissonClient redissonClient = Redisson.create(config);return redissonClient;} }

測試

@SpringBootTest public class RedissonClientTest {@AutowiredRedissonClient redissonClient;@Testpublic void test01() {System.out.println(redissonClient);}}

接下來，測試主要是針對分布式鎖來做簡單的測試，如需還有其他關于Redisson的理解，可以轉移到官方文檔。

地址：

可重入鎖（Reentrant Lock）

基于Redis的Redisson分布式可重入鎖RLock Java對象實現了java.util.concurrent.locks.Lock接口。同時還提供了異步（Async）、反射式（Reactive）和RxJava2標準的接口。

RLock lock = redisson.getLock("anyLock"); // 最常見的使用方法 lock.lock();

大家都知道，如果負責儲存這個分布式鎖的Redisson節點宕機以后，而且這個鎖正好處于鎖住的狀態時，這個鎖會出現鎖死的狀態。為了避免這種情況的發生，Redisson內部提供了一個監控鎖的看門狗，它的作用是在Redisson實例被關閉前，不斷的延長鎖的有效期。默認情況下，看門狗的檢查鎖的超時時間是30秒鐘，也可以通過修改Config.lockWatchdogTimeout來另行指定。

另外Redisson還通過加鎖的方法提供了leaseTime的參數來指定加鎖的時間。超過這個時間后鎖便自動解開了。

// 加鎖以后10秒鐘自動解鎖 // 無需調用unlock方法手動解鎖 lock.lock(10, TimeUnit.SECONDS);// 嘗試加鎖，最多等待100秒，上鎖以后10秒自動解鎖 boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS); if (res) {try {...} finally {lock.unlock();} }

Redisson同時還為分布式鎖提供了異步執行的相關方法：

RLock lock = redisson.getLock("anyLock"); lock.lockAsync(); lock.lockAsync(10, TimeUnit.SECONDS); Future<Boolean> res = lock.tryLockAsync(100, 10, TimeUnit.SECONDS);

RLock對象完全符合Java的Lock規范。也就是說只有擁有鎖的進程才能解鎖，其他進程解鎖則會拋出IllegalMonitorStateException錯誤。但是如果遇到需要其他進程也能解鎖的情況，請使用分布式信號量Semaphore 對象.

讀寫鎖（ReadWriteLock）

基于Redis的Redisson分布式可重入讀寫鎖RReadWriteLock Java對象實現了java.util.concurrent.locks.ReadWriteLock接口。其中讀鎖和寫鎖都繼承了RLock接口。

分布式可重入讀寫鎖允許同時有多個讀鎖和一個寫鎖處于加鎖狀態。

RReadWriteLock rwlock = redisson.getReadWriteLock("anyRWLock"); // 最常見的使用方法 rwlock.readLock().lock(); // 或 rwlock.writeLock().lock();

大家都知道，如果負責儲存這個分布式鎖的Redis節點宕機以后，而且這個鎖正好處于鎖住的狀態時，這個鎖會出現鎖死的狀態。為了避免這種情況的發生，Redisson內部提供了一個監控鎖的看門狗，它的作用是在Redisson實例被關閉前，不斷的延長鎖的有效期。默認情況下，看門狗的檢查鎖的超時時間是30秒鐘，也可以通過修改Config.lockWatchdogTimeout來另行指定。

另外Redisson還通過加鎖的方法提供了leaseTime的參數來指定加鎖的時間。超過這個時間后鎖便自動解開了。

// 10秒鐘以后自動解鎖 // 無需調用unlock方法手動解鎖 rwlock.readLock().lock(10, TimeUnit.SECONDS); // 或 rwlock.writeLock().lock(10, TimeUnit.SECONDS);// 嘗試加鎖，最多等待100秒，上鎖以后10秒自動解鎖 boolean res = rwlock.readLock().tryLock(100, 10, TimeUnit.SECONDS); // 或 boolean res = rwlock.writeLock().tryLock(100, 10, TimeUnit.SECONDS); ... lock.unlock();

信號量（Semaphore）

基于Redis的Redisson的分布式信號量（Semaphore）Java對象RSemaphore采用了與java.util.concurrent.Semaphore相似的接口和用法。同時還提供了異步（Async）、反射式（Reactive）和RxJava2標準的接口。

RSemaphore semaphore = redisson.getSemaphore("semaphore"); semaphore.acquire(); //或 semaphore.acquireAsync(); semaphore.acquire(23); semaphore.tryAcquire(); //或 semaphore.tryAcquireAsync(); semaphore.tryAcquire(23, TimeUnit.SECONDS); //或 semaphore.tryAcquireAsync(23, TimeUnit.SECONDS); semaphore.release(10); semaphore.release(); //或 semaphore.releaseAsync();

閉鎖（CountDownLatch）

基于Redisson的Redisson分布式閉鎖（CountDownLatch）Java對象RCountDownLatch采用了與java.util.concurrent.CountDownLatch相似的接口和用法。

RCountDownLatch latch = redisson.getCountDownLatch("anyCountDownLatch"); latch.trySetCount(1); latch.await();// 在其他線程或其他JVM里 RCountDownLatch latch = redisson.getCountDownLatch("anyCountDownLatch"); latch.countDown();

緩存一致性協議

/*** 使用redisson分布式鎖*/public Map<String, List<Catalog2Vo>> getCatalogJsonFromDBWithRedisson() {/*** 怎么使用分布式鎖* 1.占坑* 需要注意鎖的名字。鎖的粒度，越細越快* 鎖的粒度：具體緩存的是某個數據*/RLock lock = redissonClient.getLock("catalogJson-Lock");//加鎖lock.lock();Map<String, List<Catalog2Vo>> fromDB = null;try {fromDB = getFromDB();} finally {//釋放鎖lock.unlock();}return fromDB;}

假如我們有一天三級分類的數據被修改了，那我們從緩存獲取到的數據，就和我們真實數據庫的數據就產生數據不一致性。所以就牽扯到另外一個問題緩存一致性的問題。

最常用的解決緩存數據一致性的模式，分為兩種：

雙寫模式

失效模式

我們系統的解決方案：

緩存的所有數據都有過期時間，數據過期觸發主動更新

讀寫數據的時候，加上分布式的讀寫鎖，我們經常讀經常寫，會有影響。

優秀博客推薦：

深度剖析如何保證緩存與數據庫的一致性 - Virtuals - 博客園 (cnblogs.com)

緩存和數據庫一致性問題，看這篇就夠了 - 知乎 (zhihu.com)

Canal

總結

以上是生活随笔為你收集整理的Mall商城的高级篇的开发（三）缓存与分布式锁的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。