在《??Redis 數(shù)據(jù)緩存滿了怎么辦???》我們知道 Redis 緩存滿了之后能通過淘汰策略刪除數(shù)據(jù)騰出空間給新數(shù)據(jù)。

淘汰策略如下所示：

redis內(nèi)存淘汰

設(shè)置過期時(shí)間的 key

volatile-ttl、volatile-random、volatile-lru、volatile-lfu 這四種策略淘汰的數(shù)據(jù)范圍是設(shè)置了過期時(shí)間的數(shù)據(jù)。

所有的 key

allkeys-lru、allkeys-random、allkeys-lfu 這三種淘汰策略無(wú)論這些鍵值對(duì)是否設(shè)置了過期時(shí)間，當(dāng)內(nèi)存不足都會(huì)進(jìn)行淘汰。

這就意味著，即使它的過期時(shí)間還沒到，也會(huì)被刪除。當(dāng)然，如果已經(jīng)過了過期時(shí)間，即使沒有被淘汰策略選中，也會(huì)被刪除。

volatile-ttl 和 volatile-randon 很簡(jiǎn)單，重點(diǎn)在于 volatile-lru 和 volatile-lfu，他們涉及到 LRU 算法 和 LFU 算法。

今天碼哥帶大家一起搞定 Redis 的 LRU 算法…

近似 LRU 算法

什么是 LRU 算法呢?

LRU 算法的全程是 Least Rencently Used，顧名思義就是按照最近最久未使用的算法進(jìn)行數(shù)據(jù)淘汰。

核心思想「如果該數(shù)據(jù)最近被訪問，那么將來被訪問的幾率也更高」。

我們把所有的數(shù)據(jù)組織成一個(gè)鏈表：

• MRU：表示鏈表的表頭，代表著最近最常被訪問的數(shù)據(jù)。
• LRU：表示鏈表的表尾，代表最近最不常使用的數(shù)據(jù)。

LRU 算法

可以發(fā)現(xiàn)，LRU 更新和插入新數(shù)據(jù)都發(fā)生在鏈表首，刪除數(shù)據(jù)都發(fā)生在鏈表尾。

被訪問的數(shù)據(jù)會(huì)被移動(dòng)到 MRU 端，被訪問的數(shù)據(jù)之前的數(shù)據(jù)則相應(yīng)往后移動(dòng)一位。

使用單鏈表可以么?

如果選用單鏈表，刪除這個(gè)結(jié)點(diǎn)，需要 O(n) 遍歷一遍找到前驅(qū)結(jié)點(diǎn)。所以選用雙向鏈表，在刪除的時(shí)候也能 O(1) 完成。

Redis 使用該 LRU 算法管理所有的緩存數(shù)據(jù)么?

不是的，由于 LRU 算法需要用鏈表管理所有的數(shù)據(jù)，會(huì)造成大量額外的空間消耗。

除此之外，大量的節(jié)點(diǎn)被訪問就會(huì)帶來頻繁的鏈表節(jié)點(diǎn)移動(dòng)操作，從而降低了 Redis 性能。

所以 Redis 對(duì)該算法做了簡(jiǎn)化，Redis LRU 算法并不是真正的 LRU，Redis 通過對(duì)少量的 key 采樣，并淘汰采樣的數(shù)據(jù)中最久沒被訪問過的 key。

這就意味著 Redis 無(wú)法淘汰數(shù)據(jù)庫(kù)最久訪問的數(shù)據(jù)。

Redis LRU 算法有一個(gè)重要的點(diǎn)在于可以更改樣本數(shù)量來調(diào)整算法的精度，使其近似接近真實(shí)的 LRU 算法，同時(shí)又避免了內(nèi)存的消耗，因?yàn)槊看沃恍枰蓸由倭繕颖荆皇侨繑?shù)據(jù)。

配置如下：

maxmemory-samples 50

運(yùn)行原理

大家還記得么，數(shù)據(jù)結(jié)構(gòu) redisObject 中有一個(gè) lru 字段， 用于記錄每個(gè)數(shù)據(jù)最近一次被訪問的時(shí)間戳。

typedef struct redisObject {
    unsigned type:4;
    unsigned encoding:4;
    /* LRU time (relative to global lru_clock) or
     * LFU data (least significant 8 bits frequency
     * and most significant 16 bits access time).
     */
    unsigned lru:LRU_BITS;
    int refcount;
    void *ptr;
} robj;

Redis 在淘汰數(shù)據(jù)時(shí)，第一次隨機(jī)選出 N 個(gè)數(shù)據(jù)放到候選集合，將 lru 字段值最小的數(shù)據(jù)淘汰。

當(dāng)再次需要淘汰數(shù)據(jù)時(shí)，會(huì)重新挑選數(shù)據(jù)放入第一次創(chuàng)建的候選集合，不過有一個(gè)挑選標(biāo)準(zhǔn)：進(jìn)入該集合的數(shù)據(jù)的 lru 的值必須小于候選集合中最小的 lru 值。

如果新數(shù)據(jù)進(jìn)入候選集合的個(gè)數(shù)達(dá)到了 maxmemory-samples 設(shè)定的值，那就把候選集合中 lru 最小的數(shù)據(jù)淘汰。

這樣就大大減少鏈表節(jié)點(diǎn)數(shù)量，同時(shí)不用每次訪問數(shù)據(jù)都移動(dòng)鏈表節(jié)點(diǎn)，大大提升了性能。

Java 實(shí)現(xiàn) LRU Cahce

LinkedHashMap 實(shí)現(xiàn)

完全利用 Java 的LinkedHashMap實(shí)現(xiàn)，可以采用組合或者繼承的方式實(shí)現(xiàn)，「碼哥」使用組合的形式完成。

public class LRUCache<K, V> {
    private Map<K, V> map;
    private final int cacheSize;

    public LRUCache(int initialCapacity) {
        map = new LinkedHashMap<K, V>(initialCapacity, 0.75f, true) {
            @Override
            protected boolean removeEldestEntry(Map.Entry<K, V> eldest) {
                return size() > cacheSize;
            }
        };
        this.cacheSize = initialCapacity;
    }
}

重點(diǎn)在于 LinkedHashMap的第三個(gè)構(gòu)造函數(shù)上，要把這個(gè)構(gòu)造參數(shù)accessOrder設(shè)為 true，代表LinkedHashMap內(nèi)部維持訪問順序。

另外，還需要重寫removeEldestEntry()，這個(gè)函數(shù)如果返回true，代表把最久未被訪問的節(jié)點(diǎn)移除，從而實(shí)現(xiàn)淘汰數(shù)據(jù)。

自己實(shí)現(xiàn)

其中代碼是從 LeetCode 146. LRU Cache 上摘下來的。代碼里面有注釋。

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
 * 在鏈頭放最久未被使用的元素，鏈尾放剛剛添加或訪問的元素
 */
class LRUCache {
    class Node {
        int key, value;
        Node pre, next;

        Node(int key, int value) {
            this.key = key;
            this.value = value;
            pre = this;
            next = this;
        }
    }
    private final int capacity;// LRU Cache的容量
    private Node dummy;// dummy節(jié)點(diǎn)是一個(gè)冗余節(jié)點(diǎn)，dummy的next是鏈表的第一個(gè)節(jié)點(diǎn)，dummy的pre是鏈表的最后一個(gè)節(jié)點(diǎn)
    private Map<Integer, Node> cache;//保存key-Node對(duì)，Node是雙向鏈表節(jié)點(diǎn)

    public LRUCache(int capacity) {
        this.capacity = capacity;
        dummy = new Node(0, 0);
        cache = new ConcurrentHashMap<>();
    }
    public int get(int key) {
        Node node = cache.get(key);
        if (node == null) return -1;
        remove(node);
        add(node);
        return node.value;
    }
    public void put(int key, int value) {
        Node node = cache.get(key);
        if (node == null) {
            if (cache.size() >= capacity) {
                cache.remove(dummy.next.key);
                remove(dummy.next);
            }
            node = new Node(key, value);
            cache.put(key, node);
            add(node);
        } else {
            cache.remove(node.key);
            remove(node);
            node = new Node(key, value);
            cache.put(key, node);
            add(node);
        }
    }
    /**
     * 在鏈表尾部添加新節(jié)點(diǎn)
     *
     * @param node 新節(jié)點(diǎn)
     */
    private void add(Node node) {
        dummy.pre.next = node;
        node.pre = dummy.pre;
        node.next = dummy;
        dummy.pre = node;
    }
    /**
     * 從雙向鏈表中刪除該節(jié)點(diǎn)
     *
     * @param node 要?jiǎng)h除的節(jié)點(diǎn)
     */
    private void remove(Node node) {
        node.pre.next = node.next;
        node.next.pre = node.pre;
    }
}