Redis：內(nèi)存被我用完了！該怎么辦？

作者：李立敏 2021-01-26 08:02:04

Redis是一個內(nèi)存數(shù)據(jù)庫，當Redis使用的內(nèi)存超過物理內(nèi)存的限制后，內(nèi)存數(shù)據(jù)會和磁盤產(chǎn)生頻繁的交換，交換會導致Redis性能急劇下降。所以在生產(chǎn)環(huán)境中我們通過配置參數(shù)maxmemoey來限制使用的內(nèi)存大小。

[[378340]]

介紹

當實際使用的內(nèi)存超過maxmemoey后，Redis提供了如下幾種可選策略。

noeviction：寫請求返回錯誤

volatile-lru：使用lru算法刪除設置了過期時間的鍵值對 volatile-lfu：使用lfu算法刪除設置了過期時間的鍵值對 volatile-random：在設置了過期時間的鍵值對中隨機進行刪除 volatile-ttl：根據(jù)過期時間的先后進行刪除，越早過期的越先被刪除

allkeys-lru：在所有鍵值對中，使用lru算法進行刪除 allkeys-lfu：在所有鍵值對中，使用lfu算法進行刪除 allkeys-random：所有鍵值對中隨機刪除

我們來詳細了解一下lru和lfu算法，這是2個常見的緩存淘汰算法。「因為計算機緩存的容量是有限的，所以我們要刪除那些沒用的數(shù)據(jù)，而這兩種算法的區(qū)別就是判定沒用的緯度不一樣。」

LRU算法

「lru(Least recently used，最近最少使用)算法，即最近訪問的數(shù)據(jù)，后續(xù)很大概率還會被訪問到，即是有用的。而長時間未被訪問的數(shù)據(jù)，應該被淘汰」

lru算法中數(shù)據(jù)會被放到一個鏈表中，鏈表的頭節(jié)點為最近被訪問的數(shù)據(jù)，鏈表的尾節(jié)點為長時間沒有被訪問的數(shù)據(jù)

「lru算法的核心實現(xiàn)就是哈希表加雙向鏈表」。鏈表可以用來維護訪問元素的順序，而hash表可以幫我們在O(1)時間復雜度下訪問到元素。

「至于為什么是雙向鏈表呢」?主要是要刪除元素，所以要獲取前繼節(jié)點。數(shù)據(jù)結構圖示如下

使用雙向鏈表+HashMap

雙向鏈表節(jié)點定義如下

public class ListNode<K, V> { 
    K key; 
    V value; 
    ListNode pre; 
    ListNode next; 
 
    public ListNode() {} 
 
    public ListNode(K key, V value) { 
        this.key = key; 
        this.value = value; 
    } 
}

封裝雙向鏈表的常用操作

public class DoubleList { 
 
    private ListNode head; 
    private ListNode tail; 
 
    public DoubleList() { 
        head = new ListNode(); 
        tail = new ListNode(); 
        head.next = tail; 
        tail.pre = head; 
    } 
 
    public void remove(ListNode node) { 
        node.pre.next = node.next; 
        node.next.pre = node.pre; 
    } 
 
    public void addLast(ListNode node) { 
        node.pre = tail.pre; 
        tail.pre = node; 
        node.pre.next = node; 
        node.next = tail; 
    } 
 
    public ListNode removeFirst() { 
        if (head.next == tail) { 
            return null; 
        } 
        ListNode first = head.next; 
        remove(first); 
        return first; 
    } 
}

封裝一個緩存類，提供最基本的get和put方法。「需要注意，這兩種基本的方法都涉及到對兩種數(shù)據(jù)結構的修改」。

public class MyLruCache<K, V> { 
 
    private int capacity; 
    private DoubleList doubleList; 
    private Map<K, ListNode> map; 
 
    public MyLruCache(int capacity) { 
        this.capacity = capacity; 
        map = new HashMap<>(); 
        doubleList = new DoubleList(); 
    } 
 
    public V get(Object key) { 
        ListNode<K, V> node = map.get(key); 
        if (node == null) { 
            return null; 
        } 
        // 先刪除該節(jié)點，再接到尾部 
        doubleList.remove(node); 
        doubleList.addLast(node); 
        return node.value; 
    } 
 
    public void put(K key, V value) { 
        // 直接調(diào)用這邊的get方法，如果存在，它會在get內(nèi)部被移動到尾巴，不用再移動一遍,直接修改值即可 
        if ((get(key)) != null) { 
            map.get(key).value = value; 
            return; 
        } 
 
        // 如果超出容量，把頭去掉 
        if (map.size() == capacity) { 
            ListNode listNode = doubleList.removeFirst(); 
            map.remove(listNode.key); 
        } 
 
        // 若不存在，new一個出來 
        ListNode node = new ListNode(key, value); 
        map.put(key, node); 
        doubleList.addLast(node); 
    } 
}

這里我們的實現(xiàn)為最近訪問的放在鏈表的尾節(jié)點，不經(jīng)常訪問的放在鏈表的頭節(jié)點

測試一波，輸出為鏈表的正序輸出(代碼為了簡潔沒有貼toString方法)

MyLruCache<String, String> myLruCache = new MyLruCache<>(3); 
// {5 : 5} 
myLruCache.put("5", "5"); 
// {5 : 5}{3 : 3} 
myLruCache.put("3", "3"); 
// {5 : 5}{3 : 3}{4 : 4} 
myLruCache.put("4", "4"); 
// {3 : 3}{4 : 4}{2 : 2} 
myLruCache.put("2", "2"); 
// {4 : 4}{2 : 2}{3 : 3} 
myLruCache.get("3");

「因為LinkedHashMap的底層實現(xiàn)就是哈希表加雙向鏈表，所以你可以用LinkedHashMap替換HashMap和DoubleList來改寫一下上面的類」。

我來演示一下更騷的操作，只需要重寫一個構造函數(shù)和removeEldestEntry方法即可。

使用LinkedHashMap實現(xiàn)LRU

public class LruCache<K, V> extends LinkedHashMap<K, V> { 
 
    private int cacheSize; 
 
 
    public LruCache(int cacheSize) { 
        /** 
         * initialCapacity: 初始容量大小 
         * loadFactor: 負載因子 
         * accessOrder: false基于插入排序（默認），true基于訪問排序 
         */ 
        super(cacheSize, 0.75f, true); 
        this.cacheSize = cacheSize; 
    } 
 
    /** 
     * 當調(diào)用put或者putAll方法時會調(diào)用如下方法，是否刪除最老的數(shù)據(jù)，默認為false 
     */ 
    @Override 
    protected boolean removeEldestEntry(Map.Entry<K, V> eldest) { 
        return size() > cacheSize; 
    } 
}

注意這個緩存并不是線程安全的，可以調(diào)用Collections.synchronizedMap方法返回線程安全的map

LruCache<String, String> lruCache = new LruCache(3); 
Map<String, String> safeMap = Collections.synchronizedMap(lruCache);

Collections.synchronizedMap實現(xiàn)線程安全的方式很簡單，只是返回一個代理類。代理類對Map接口的所有方法加鎖

public static <K,V> Map<K,V> synchronizedMap(Map<K,V> m) { 
    return new SynchronizedMap<>(m); 
}

LFU算法

「LRU算法有一個問題，當一個長時間不被訪問的key，偶爾被訪問一下后，可能會造成一個比這個key訪問更頻繁的key被淘汰。」

即LRU算法對key的冷熱程度的判斷可能不準確。而LFU算法(Least Frequently Used，最不經(jīng)常使用)則是按照訪問頻率來判斷key的冷熱程度的，每次刪除的是一段時間內(nèi)訪問頻率較低的數(shù)據(jù)，比LRU算法更準確。

使用3個hash表實現(xiàn)lfu算法

那么我們應該如何組織數(shù)據(jù)呢?

為了實現(xiàn)鍵值的對快速訪問，用一個map來保存鍵值對

private HashMap<K, Integer> keyToFreq;

還需要用一個map來保存鍵的訪問頻率

private HashMap<K, Integer> keyToFreq;

「當然你也可以把值和訪問頻率封裝到一個類中，用一個map來替代上述的2個map」

接下來就是最核心的部分，刪除訪問頻率最低的數(shù)據(jù)。

為了能在O(1)時間復雜度內(nèi)找到訪問頻率最低的數(shù)據(jù)，我們需要一個變量minFreq記錄訪問最低的頻率
每個訪問頻率有可能對應多個鍵。當空間不夠用時，我們要刪除最早被訪問的數(shù)據(jù)，所以需要如下數(shù)據(jù)結構，Map<頻率, 有序集合>。每次內(nèi)存不夠用時，刪除有序集合的第一個元素即可。并且這個有序集合要能快速刪除某個key，因為某個key被訪問后，需要從這個集合中刪除，加入freq+1對應的集合中
有序集合很多，但是能滿足快速刪除某個key的只有set，但是set插入數(shù)據(jù)是無序的。「幸虧Java有LinkedHashSet這個類，鏈表和集合的結合體，鏈表不能快速刪除元素，但是能保證插入順序。集合內(nèi)部元素無序，但是能快速刪除元素，完美」

下面就是具體的實現(xiàn)。

public class LfuCache<K, V> { 
 
    private HashMap<K, V> keyToVal; 
    private HashMap<K, Integer> keyToFreq; 
    private HashMap<Integer, LinkedHashSet<K>> freqTokeys; 
 
    private int minFreq; 
    private int capacity; 
 
    public LfuCache(int capacity) { 
        keyToVal = new HashMap<>(); 
        keyToFreq = new HashMap<>(); 
        freqTokeys = new HashMap<>(); 
        this.capacity = capacity; 
        this.minFreq = 0; 
    } 
 
    public V get(K key) { 
        V v = keyToVal.get(key); 
        if (v == null) { 
            return null; 
        } 
        increaseFrey(key); 
        return v; 
    } 
 
    public void put(K key, V value) { 
        // get方法里面會增加頻次 
        if (get(key) != null) { 
            // 重新設置值 
            keyToVal.put(key, value); 
            return; 
        } 
 
        // 超出容量，刪除頻率最低的key 
        if (keyToVal.size() >= capacity) { 
            removeMinFreqKey(); 
        } 
 
        keyToVal.put(key, value); 
        keyToFreq.put(key, 1); 
        // key對應的value存在，返回存在的key 
        // key對應的value不存在，添加key和value 
        freqTokeys.putIfAbsent(1, new LinkedHashSet<>()); 
        freqTokeys.get(1).add(key); 
        this.minFreq = 1; 
    } 
 
    // 刪除出現(xiàn)頻率最低的key 
    private void removeMinFreqKey() { 
        LinkedHashSet<K> keyList = freqTokeys.get(minFreq); 
        K deleteKey = keyList.iterator().next(); 
        keyList.remove(deleteKey); 
        if (keyList.isEmpty()) { 
            // 這里刪除元素后不需要重新設置minFreq 
            // 因為put方法執(zhí)行完會將minFreq設置為1 
            freqTokeys.remove(keyList); 
        } 
        keyToVal.remove(deleteKey); 
        keyToFreq.remove(deleteKey); 
    } 
 
    // 增加頻率 
    private void increaseFrey(K key) { 
        int freq = keyToFreq.get(key); 
        keyToFreq.put(key, freq + 1); 
        freqTokeys.get(freq).remove(key); 
        freqTokeys.putIfAbsent(freq + 1, new LinkedHashSet<>()); 
        freqTokeys.get(freq + 1).add(key); 
        if (freqTokeys.get(freq).isEmpty()) { 
            freqTokeys.remove(freq); 
            // 最小頻率的set為空，key被移動到minFreq+1對應的set了 
            // 所以minFreq也要加1 
            if (freq == this.minFreq) { 
                this.minFreq++; 
            } 
        } 
    } 
}

測試一下

LfuCache<String, String> lfuCache = new LfuCache(2); 
lfuCache.put("1", "1"); 
lfuCache.put("2", "2"); 
// 1 
System.out.println(lfuCache.get("1")); 
lfuCache.put("3", "3"); 
// 1的頻率為2，2和3的頻率為1，但2更早之前被訪問，所以被清除 
// 結果為null 
System.out.println(lfuCache.get("2"));

責任編輯：武曉燕來源： Java識堂

Redis 內(nèi)存數(shù)據(jù)庫