把redis作為緩存使用已經(jīng)是司空見慣，但是使用redis后也可能會碰到一系列的問題，尤其是數(shù)據(jù)量很大的時候，經(jīng)典的幾個問題如下：

(一)緩存和數(shù)據(jù)庫間數(shù)據(jù)一致性問題
分布式環(huán)境下（單機就不用說了）非常容易出現(xiàn)緩存和數(shù)據(jù)庫間的數(shù)據(jù)一致性問題，針對這一點的話，只能說，如果你的項目對緩存的要求是強一致性的，那么請不要使用緩存。我們只能采取合適的策略來降低緩存和數(shù)據(jù)庫間數(shù)據(jù)不一致的概率，而無法保證兩者間的強一致性。合適的策略包括 合適的緩存更新策略，更新數(shù)據(jù)庫后要及時更新緩存、緩存失敗時增加重試機制，例如MQ模式的消息隊列。

(二)緩存擊穿問題
緩存擊穿表示惡意用戶模擬請求很多緩存中不存在的數(shù)據(jù)，由于緩存中都沒有，導(dǎo)致這些請求短時間內(nèi)直接落在了數(shù)據(jù)庫上，導(dǎo)致數(shù)據(jù)庫異常。這個我們在實際項目就遇到了，有些搶購活動、秒殺活動的接口API被大量的惡意用戶刷，導(dǎo)致短時間內(nèi)數(shù)據(jù)庫c超時了，好在數(shù)據(jù)庫是讀寫分離，同時也有進行接口限流，hold住了。

解決方案的話：

方案1、使用互斥鎖排隊

業(yè)界比價普遍的一種做法，即根據(jù)key獲取value值為空時，鎖上，從數(shù)據(jù)庫中l(wèi)oad數(shù)據(jù)后再釋放鎖。若其它線程獲取鎖失敗，則等待一段時間后重試。這里要注意，分布式環(huán)境中要使用分布式鎖，單機的話用普通的鎖（synchronized、Lock）就夠了。

public String getWithLock(String key, Jedis jedis, String lockKey, String uniqueId, long expireTime) {
? ? // 通過key獲取value
? ? String value = redisService.get(key);
? ? if (StringUtil.isEmpty(value)) {
? ? ? ? // 分布式鎖，詳細可以參考https://blog.csdn.net/fanrenxiang/article/details/79803037
? ? ? ? //封裝的tryDistributedLock包括setnx和expire兩個功能，在低版本的redis中不支持
? ? ? ? try {
? ? ? ? ? ? boolean locked = redisService.tryDistributedLock(jedis, lockKey, uniqueId, expireTime);
? ? ? ? ? ? if (locked) {
? ? ? ? ? ? ? ? value = userService.getById(key);
? ? ? ? ? ? ? ? redisService.set(key, value);
? ? ? ? ? ? ? ? redisService.del(lockKey);
? ? ? ? ? ? ? ? return value;
? ? ? ? ? ? } else {
? ? ? ? ? ? ? ? // 其它線程進來了沒獲取到鎖便等待50ms后重試
? ? ? ? ? ? ? ? Thread.sleep(50);
? ? ? ? ? ? ? ? getWithLock(key, jedis, lockKey, uniqueId, expireTime);
? ? ? ? ? ? }
? ? ? ? } catch (Exception e) {
? ? ? ? ? ? log.error("getWithLock exception=" + e);
? ? ? ? ? ? return value;
? ? ? ? } finally {
? ? ? ? ? ? redisService.releaseDistributedLock(jedis, lockKey, uniqueId);
? ? ? ? }
? ? }
? ? return value;
}
這樣做思路比較清晰，也從一定程度上減輕數(shù)據(jù)庫壓力，但是鎖機制使得邏輯的復(fù)雜度增加，吞吐量也降低了，有點治標不治本。

方案2、接口限流與熔斷、降級

重要的接口一定要做好限流策略，防止用戶惡意刷接口，同時要降級準備，當接口中的某些服務(wù)不可用時候，進行熔斷，失敗快速返回機制。

方案3、布隆過濾器

bloomfilter就類似于一個hash set，用于快速判某個元素是否存在于集合中，其典型的應(yīng)用場景就是快速判斷一個key是否存在于某容器，不存在就直接返回。布隆過濾器的關(guān)鍵就在于hash算法和容器大小，下面先來簡單的實現(xiàn)下看看效果，我這里用guava實現(xiàn)的布隆過濾器：

<dependencies> ?
? ? ?<dependency> ?
? ? ? ? ?<groupId>com.google.guava</groupId> ?
? ? ? ? ?<artifactId>guava</artifactId> ?
? ? ? ? ?<version>23.0</version> ?
? ? ?</dependency> ?
</dependencies> ?
public class BloomFilterTest {
?
? ? private static final int capacity = 1000000;
? ? private static final int key = 999998;
?
? ? private static BloomFilter<Integer> bloomFilter = BloomFilter.create(Funnels.integerFunnel(), capacity);
?
? ? static {
? ? ? ? for (int i = 0; i < capacity; i++) {
? ? ? ? ? ? bloomFilter.put(i);
? ? ? ? }
? ? }
?
? ? public static void main(String[] args) {
? ? ? ? /*返回計算機最精確的時間，單位微妙*/
? ? ? ? long start = System.nanoTime();
?
? ? ? ? if (bloomFilter.mightContain(key)) {
? ? ? ? ? ? System.out.println("成功過濾到" + key);
? ? ? ? }
? ? ? ? long end = System.nanoTime();
? ? ? ? System.out.println("布隆過濾器消耗時間:" + (end - start));
? ? ? ? int sum = 0;
? ? ? ? for (int i = capacity + 20000; i < capacity + 30000; i++) {
? ? ? ? ? ? if (bloomFilter.mightContain(i)) {
? ? ? ? ? ? ? ? sum = sum + 1;
? ? ? ? ? ? }
? ? ? ? }
? ? ? ? System.out.println("錯判率為:" + sum);
? ? }
}
成功過濾到999998
布隆過濾器消耗時間:215518
錯判率為:318
可以看到，100w個數(shù)據(jù)中只消耗了約0.2毫秒就匹配到了key，速度足夠快。然后模擬了1w個不存在于布隆過濾器中的key，匹配錯誤率為318/10000，也就是說，出錯率大概為3%，跟蹤下BloomFilter的源碼發(fā)現(xiàn)默認的容錯率就是0.03：

public static <T> BloomFilter<T> create(Funnel<T> funnel, int expectedInsertions /* n */) {
? return create(funnel, expectedInsertions, 0.03); // FYI, for 3%, we always get 5 hash functions
}
我們可調(diào)用BloomFilter的這個方法顯式的指定誤判率：

private static BloomFilter<Integer> bloomFilter = BloomFilter.create(Funnels.integerFunnel(), capacity,0.01);
我們斷點跟蹤下，誤判率為0.02和默認的0.03時候的區(qū)別:

對比兩個出錯率可以發(fā)現(xiàn)，誤判率為0.02時數(shù)組大小為8142363，0.03時為7298440，誤判率降低了0.01，BloomFilter維護的數(shù)組大小也減少了843923，可見BloomFilter默認的誤判率0.03是設(shè)計者權(quán)衡系統(tǒng)性能后得出的值。要注意的是，布隆過濾器不支持刪除操作。用在這邊解決緩存穿透問題就是：

public String getByKey(String key) {
? ? // 通過key獲取value
? ? String value = redisService.get(key);
? ? if (StringUtil.isEmpty(value)) {
? ? ? ? if (bloomFilter.mightContain(key)) {
? ? ? ? ? ? value = userService.getById(key);
? ? ? ? ? ? redisService.set(key, value);
? ? ? ? ? ? return value;
? ? ? ? } else {
? ? ? ? ? ? return null;
? ? ? ? }
? ? }
? ? return value;
}
(三)緩存雪崩問題
緩存在同一時間內(nèi)大量鍵過期（失效），接著來的一大波請求瞬間都落在了數(shù)據(jù)庫中導(dǎo)致連接異常。

解決方案：

方案1、也是像解決緩存穿透一樣加鎖排隊，實現(xiàn)同上;

方案2、建立備份緩存，緩存A和緩存B，A設(shè)置超時時間，B不設(shè)值超時時間，先從A讀緩存，A沒有讀B，并且更新A緩存和B緩存;

方案3、設(shè)置緩存超時時間的時候加上一個隨機的時間長度，比如這個緩存key的超時時間是固定的5分鐘加上隨機的2分鐘，醬紫可從一定程度上避免雪崩問題；

public String getByKey(String keyA,String keyB) {
? ? String value = redisService.get(keyA);
? ? if (StringUtil.isEmpty(value)) {
? ? ? ? value = redisService.get(keyB);
? ? ? ? String newValue = getFromDbById();
? ? ? ? redisService.set(keyA,newValue,31, TimeUnit.DAYS);
? ? ? ? redisService.set(keyB,newValue);
? ? }
? ? return value;
}
(四)緩存并發(fā)問題
這里的并發(fā)指的是多個redis的client同時set key引起的并發(fā)問題。其實redis自身就是單線程操作，多個client并發(fā)操作，按照先到先執(zhí)行的原則，先到的先執(zhí)行，其余的阻塞。當然，另外的解決方案是把redis.set操作放在隊列中使其串行化，必須的一個一個執(zhí)行，具體的代碼就不上了，當然加鎖也是可以的，至于為什么不用redis中的事務(wù)，留給各位看官自己思考探究。
?

總結(jié)

以上是生活随笔為你收集整理的缓存穿透、缓存雪崩、redis并发的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。