有趣的Redis：快取被我寫滿了，該怎么辦？

介紹

Redis是一個記憶體資料庫，當Redis使用的記憶體超過物理記憶體的限制后，記憶體資料會和磁盤產生頻繁的交換，交換會導致Redis性能急劇下降，所以在生產環境中我們通過配置引數maxmemoey來限制使用的記憶體大小，

當實際使用的記憶體超過maxmemoey后，Redis提供了如下幾種可選策略，

noeviction：寫請求回傳錯誤

volatile-lru：使用lru演算法洗掉設定了過期時間的鍵值對
volatile-lfu：使用lfu演算法洗掉設定了過期時間的鍵值對
volatile-random：在設定了過期時間的鍵值對中隨機進行洗掉
volatile-ttl：根據過期時間的先后進行洗掉，越早過期的越先被洗掉

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

我們來詳細了解一下lru和lfu演算法，這是2個常見的快取淘汰演算法，因為計算機快取的容量是有限的，所以我們要洗掉那些沒用的資料，而這兩種演算法的區別就是判定沒用的緯度不一樣，

LRU演算法

lru（Least recently used，最近最少使用）演算法，即最近訪問的資料，后續很大概率還會被訪問到，即是有用的，而長時間未被訪問的資料，應該被淘汰

lru演算法中資料會被放到一個鏈表中，鏈表的頭節點為最近被訪問的資料，鏈表的尾節點為長時間沒有被訪問的資料

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

至于為什么是雙向鏈表呢？主要是要洗掉元素，所以要獲取前繼節點，資料結構圖示如下

使用雙向鏈表+HashMap

雙向鏈表節點定義如下

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方法，需要注意，這兩種基本的方法都涉及到對兩種資料結構的修改，

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;
        }
        // 先洗掉該節點，再接到尾部
        doubleList.remove(node);
        doubleList.addLast(node);
        return node.value;
    }

    public void put(K key, V value) {
        // 直接呼叫這邊的get方法，如果存在，它會在get內部被移動到尾巴，不用再移動一遍,直接修改值即可
        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);
    }
}

這里我們的實作為最近訪問的放在鏈表的尾節點，不經常訪問的放在鏈表的頭節點

測驗一波，輸出為鏈表的正序輸出（代碼為了簡潔沒有貼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的底層實作就是哈希表加雙向鏈表，所以你可以用LinkedHashMap替換HashMap和DoubleList來改寫一下上面的類，

我來演示一下更騷的操作，只需要重寫一個建構式和removeEldestEntry方法即可，

使用LinkedHashMap實作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;
    }

    /**
     * 當呼叫put或者putAll方法時會呼叫如下方法，是否洗掉最老的資料，默認為false
     */
    @Override
    protected boolean removeEldestEntry(Map.Entry<K, V> eldest) {
        return size() > cacheSize;
    }
}

注意這個快取并不是執行緒安全的，可以呼叫Collections.synchronizedMap方法回傳執行緒安全的map

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

Collections.synchronizedMap實作執行緒安全的方式很簡單，只是回傳一個代理類，代理類對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，最不經常使用）則是按照訪問頻率來判斷key的冷熱程度的，每次洗掉的是一段時間內訪問頻率較低的資料，比LRU演算法更準確，

使用3個hash表實作lfu演算法

那么我們應該如何組織資料呢？

為了時間鍵值的對快速訪問，用一個map來保存鍵值對

private HashMap<K, Integer> keyToFreq;

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

private HashMap<K, Integer> keyToFreq;

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

接下來就是最核心的部分，洗掉訪問頻率最低的資料，

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

下面就是具體的實作，

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;
    }

    // 洗掉出現頻率最低的key
    private void removeMinFreqKey() {
        LinkedHashSet<K> keyList = freqTokeys.get(minFreq);
        K deleteKey = keyList.iterator().next();
        keyList.remove(deleteKey);
        if (keyList.isEmpty()) {
            // 這里洗掉元素后不需要重新設定minFreq
            // 因為put方法執行完會將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"));

參考博客

lru演算法和lfu演算法
[1]https://my.oschina.net/lscherish/blog/4467394
海子大佬的頁面置換演算法
[2]https://www.cnblogs.com/dolphin0520/p/3749259.html
[3]leetcode lru演算法
https://leetcode-cn.com/problems/lru-cache-lcci/
[4]leetcode lfu演算法
https://leetcode-cn.com/problems/lfu-cache
[5]https://labuladong.gitbook.io/algo/shu-ju-jie-gou-xi-lie/shou-ba-shou-she-ji-shu-ju-jie-gou/lru-suan-fa
lfu演算法
[6]https://mp.weixin.qq.com/s/oXv03m1J8TwtHwMJEZ1ApQ