把redis做爲緩存使用已是司空見慣,可是使用redis後也可能會碰到一系列的問題,尤爲是數據量很大的時候,經典的幾個問題以下:web
(一)緩存和數據庫間數據一致性問題redis
分佈式環境下(單機就不用說了)很是容易出現緩存和數據庫間的數據一致性問題,針對這一點的話,只能說,若是你的項目對緩存的要求是強一致性的,那麼請不要使用緩存。咱們只能採起合適的策略來下降緩存和數據庫間數據不一致的機率,而沒法保證二者間的強一致性。合適的策略包括 合適的緩存更新策略,更新數據庫後要及時更新緩存、緩存失敗時增長重試機制,例如MQ模式的消息隊列。算法
(二)緩存擊穿問題數據庫
緩存擊穿表示惡意用戶模擬請求不少緩存中不存在的數據,因爲緩存中都沒有,致使這些請求短期內直接落在了數據庫上,致使數據庫異常。這個咱們在實際項目就遇到了,有些搶購活動、秒殺活動的接口API被大量的惡意用戶刷,致使短期內數據庫宕機了,好在數據庫是多主多從的,hold住了。數組
解決方案的話:緩存
一、使用互斥鎖排隊併發
業界比價廣泛的一種作法,即根據key獲取value值爲空時,鎖上,從數據庫中load數據後再釋放鎖。若其它線程獲取鎖失敗,則等待一段時間後重試。這裏要注意,分佈式環境中要使用分佈式鎖,單機的話用普通的鎖(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; }
這樣作思路比較清晰,也從必定程度上減輕數據庫壓力,可是鎖機制使得邏輯的複雜度增長,吞吐量也下降了,有點治標不治本。ide
二、布隆過濾器(推薦)性能
bloomfilter就相似於一個hash set,用於快速判某個元素是否存在於集合中,其典型的應用場景就是快速判斷一個key是否存在於某容器,不存在就直接返回。布隆過濾器的關鍵就在於hash算法和容器大小,下面先來簡單的實現下看看效果,我這裏用guava實現的布隆過濾器:
<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個數據中只消耗了約0.2毫秒就匹配到了key,速度足夠快。而後模擬了1w個不存在於布隆過濾器中的key,匹配錯誤率爲318/10000,也就是說,出錯率大概爲3%,跟蹤下BloomFilter的源碼發現默認的容錯率就是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 }
咱們可調用BloomFilter的這個方法顯式的指定誤判率:
private static BloomFilter<Integer> bloomFilter = BloomFilter.create(Funnels.integerFunnel(), capacity,0.01);
咱們斷點跟蹤下,誤判率爲0.02和默認的0.03時候的區別:
對比兩個出錯率能夠發現,誤判率爲0.02時數組大小爲8142363,0.03時爲7298440,誤判率下降了0.01,BloomFilter維護的數組大小也減小了843923,可見BloomFilter默認的誤判率0.03是設計者權衡系統性能後得出的值。要注意的是,布隆過濾器不支持刪除操做。用在這邊解決緩存穿透問題就是:
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; }
緩存在同一時間內大量鍵過時(失效),接着來的一大波請求瞬間都落在了數據庫中致使鏈接異常。
解決方案:
一、也是像解決緩存穿透同樣加鎖排隊,實現同上;
二、創建備份緩存,緩存A和緩存B,A設置超時時間,B不設值超時時間,先從A讀緩存,A沒有讀B,而且更新A緩存和B緩存;
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; }
(四)緩存併發問題
這裏的併發指的是多個redis的client同時set key引發的併發問題。比較有效的解決方案就是把redis.set操做放在隊列中使其串行化,必須的一個一個執行,具體的代碼就不上了,固然加鎖也是能夠的,至於爲何不用redis中的事務,留給各位看官本身思考探究。