Redis - LRU原理 + Redis的LRU實現

總結

1.Redis的LRU

• 實現思路
  • 最初思路：隨機選三個Key，把idle time（距離最后一次被命令程序訪問的時間）最大的那個Key移除。后來，把3改成可配置的一個參數，默認為N=5：maxmemory-samples 5。該方法雖簡單有效，但它還是有缺點的：每次隨機選擇的時候，並沒有利用歷史信息。其實每一輪在移除時，其實是知道了N個Key的idle time的情況的。如果有效利用這些信息，可以改進其准確性。
  • 優化思路：采用緩沖池(pooling)，把一個全局排序問題 轉化成為了 局部的比較問題。當每一輪移除Key時，拿到了這個N個Key的idle time，如果它的idle time比 pool 里面的 Key的idle time還要大，就把它添加到pool里面去。這樣一來，每次移除的Key並不僅僅是隨機選擇的N個Key里面最大的，而且還是pool里面idle time最大的，並且：pool 里面的Key是經過多輪比較篩選的，它的idle time 在概率上比隨機獲取的Key的idle time要大，可以這么理解：pool 里面的Key 保留了"歷史經驗信息"。
• 數據結構
  • Redis的LRU沒有使用雙向鏈表數據結構，它嫌LinkedList占用的空間太大了。
  • Redis並不是直接基於字符串、鏈表、字典等數據結構來實現KV數據庫，而是在這些數據結構上創建了一個對象系統Redis Object。在redisObject結構體中定義了一個長度24bit的unsigned類型的字段，用來存儲對象最后一次被命令程序訪問的時間。

 

2.非Redis的普通情況下，使用 HashMap + 雙向鏈表 實現LRU的原因：

• 使用 HashMap [存儲 key]，這樣可以做到 save 和 get key的時間都是 O(1)
• 雙向鏈表可以讓 [移除 key] 的時間也是 O(1)

使用 HashMap 的 Value 指向雙向鏈表實現的 LRU 的 Node 節點

 

 

一、LRU 原理（Least Recently Used）

在一般標准的操作系統教材里，會用下面的方式來演示 LRU 原理，假設內存只能容納3個頁大小，按照 7 0 1 2 0 3 0 4 的次序訪問頁。假設內存按照棧的方式來描述訪問時間，在上面的，是最近訪問的，在下面的是，最遠時間訪問的，LRU就是這樣工作的。

但是如果讓我們自己設計一個基於 LRU 的緩存，這樣設計可能問題很多，這段內存按照訪問時間進行了排序，會有大量的內存拷貝操作，所以性能肯定是不能接受的。

那么如何設計一個LRU緩存，使得放入和移除都是 O(1) 的，我們需要把訪問次序維護起來，但是不能通過內存中的真實排序來反應，有一種方案就是使用雙向鏈表。

 

二、基於 HashMap 和 雙向鏈表 實現 LRU

Java中的LinkedHashmap對哈希鏈表已經有了很好實現了，需要注意的是，這段不是線程安全的，要想做到線程安全，需要加上synchronized修飾符。

整體的設計思路是，可以使用 HashMap 存儲 key，這樣可以做到 save 和 get key的時間都是 O(1)，而 HashMap 的 Value 指向雙向鏈表實現的 LRU 的 Node 節點，如圖所示。

RU 存儲是基於雙向鏈表實現的，下面的圖演示了它的原理。其中 h 代表雙向鏈表的表頭，t 代表尾部。首先預先設置 LRU 的容量，如果存儲滿了，可以通過 O(1) 的時間淘汰掉雙向鏈表的尾部，每次新增和訪問數據，都可以通過 O(1)的效率把新的節點增加到對頭，或者把已經存在的節點移動到隊頭。

下面展示了，預設大小是 3 的，LRU存儲的在存儲和訪問過程中的變化。為了簡化圖復雜度，圖中沒有展示 HashMap部分的變化，僅僅演示了上圖 LRU 雙向鏈表的變化。我們對這個LRU緩存的操作序列如下：

save(“key1”, 7)

save(“key2”, 0)

save(“key3”, 1)

save(“key4”, 2)

get(“key2”)

save(“key5”, 3)

get(“key2”)

save(“key6”, 4)

相應的 LRU 雙向鏈表部分變化如下：

 

 

 

總結一下核心操作的步驟:

save(key, value)，首先在 HashMap 找到 Key 對應的節點，如果節點存在，更新節點的值，並把這個節點移動隊頭。如果不存在，需要構造新的節點，並且嘗試把節點塞到隊頭，如果LRU空間不足，則通過 tail 淘汰掉隊尾的節點，同時在 HashMap 中移除 Key。
get(key)，通過 HashMap 找到 LRU 鏈表節點，把節點插入到隊頭，返回緩存的值。

 

完整基於 Java 的代碼參考如下：

class DLinkedNode {
    String key;
    int value;
    DLinkedNode pre;
    DLinkedNode post;
}

 

LRU Cache

public class LRUCache {
   
    private Hashtable<Integer, DLinkedNode>
            cache = new Hashtable<Integer, DLinkedNode>();
    private int count;
    private int capacity;
    private DLinkedNode head, tail;
 
    public LRUCache(int capacity) {
        this.count = 0;
        this.capacity = capacity;
 
        head = new DLinkedNode();
        head.pre = null;
 
        tail = new DLinkedNode();
        tail.post = null;
 
        head.post = tail;
        tail.pre = head;
    }
 
    public int get(String key) {
 
        DLinkedNode node = cache.get(key);
        if(node == null){
            return -1; // should raise exception here.
        }
 
        // move the accessed node to the head;
        this.moveToHead(node);
 
        return node.value;
    }
 
 
    public void set(String key, int value) {
        DLinkedNode node = cache.get(key);
 
        if(node == null){
 
            DLinkedNode newNode = new DLinkedNode();
            newNode.key = key;
            newNode.value = value;
 
            this.cache.put(key, newNode);
            this.addNode(newNode);
 
            ++count;
 
            if(count > capacity){
                // pop the tail
                DLinkedNode tail = this.popTail();
                this.cache.remove(tail.key);
                --count;
            }
        }else{
            // update the value.
            node.value = value;
            this.moveToHead(node);
        }
    }
    /**
     * Always add the new node right after head;
     */
    private void addNode(DLinkedNode node){
        node.pre = head;
        node.post = head.post;
 
        head.post.pre = node;
        head.post = node;
    }
 
    /**
     * Remove an existing node from the linked list.
     */
    private void removeNode(DLinkedNode node){
        DLinkedNode pre = node.pre;
        DLinkedNode post = node.post;
 
        pre.post = post;
        post.pre = pre;
    }
 
    /**
     * Move certain node in between to the head.
     */
    private void moveToHead(DLinkedNode node){
        this.removeNode(node);
        this.addNode(node);
    }
 
    // pop the current tail.
    private DLinkedNode popTail(){
        DLinkedNode res = tail.pre;
        this.removeNode(res);
        return res;
    }
}

 

三、Redis 中如何實現 LRU

最直觀的想法：LRU啊，記錄下每個key 最近一次的訪問時間（比如unix timestamp），unix timestamp最小的Key，就是最近未使用的，把這個Key移除。看下來一個HashMap就能搞定啊。是的，但是首先需要存儲每個Key和它的timestamp。其次，還要比較timestamp得出最小值。代價很大，不現實啊。

第二種方法：換個角度，不記錄具體的訪問時間點(unix timestamp)，而是記錄idle time：idle time越小，意味着是最近被訪問的。

The LRU algorithm evicts the Least Recently Used key, which means the one with the greatest idle time.

比如A、B、C、D四個Key，A每5s訪問一次，B每2s訪問一次，C和D每10s訪問一次。（一個波浪號代表1s），從上圖中可看出：A的空閑時間是2s，B的idle time是1s，C的idle time是5s，D剛剛訪問了所以idle time是0s

這里，用一個雙向鏈表(linkedlist)把所有的Key鏈表起來，如果一個Key被訪問了，將就這個Key移到鏈表的表頭，而要移除Key時，直接從表尾移除。

It is simple to implement because all we need to do is to track the last time a given key was accessed, or sometimes this is not even needed: we may just have all the objects we want to evict linked in a linked list.

但是在redis中，並沒有采用這種方式實現，它嫌LinkedList占用的空間太大了。Redis並不是直接基於字符串、鏈表、字典等數據結構來實現KV數據庫，而是在這些數據結構上創建了一個對象系統Redis Object。在redisObject結構體中定義了一個長度24bit的unsigned類型的字段，用來存儲對象最后一次被命令程序訪問的時間：

By modifying a bit the Redis Object structure I was able to make 24 bits of space. There was no room for linking the objects in a linked list (fat pointers!)

畢竟，並不需要一個完全准確的LRU算法，就算移除了一個最近訪問過的Key，影響也不太。

To add another data structure to take this metadata was not an option, however since LRU is itself an approximation of what we want to achieve, what about approximating LRU itself?

最初Redis是這樣實現的：

隨機選三個Key，把idle time最大的那個Key移除。后來，把3改成可配置的一個參數，默認為N=5：maxmemory-samples 5

when there is to evict a key, select 3 random keys, and evict the one with the highest idle time

就是這么簡單，簡單得讓人不敢相信了，而且十分有效。但它還是有缺點的：每次隨機選擇的時候，並沒有利用歷史信息。在每一輪移除(evict)一個Key時，隨機從N個里面選一個Key，移除idle time最大的那個Key；下一輪又是隨機從N個里面選一個Key...有沒有想過：在上一輪移除Key的過程中，其實是知道了N個Key的idle time的情況的，那我能不能在下一輪移除Key時，利用好上一輪知曉的一些信息？

However if you think at this algorithm across its executions, you can see how we are trashing a lot of interesting data. Maybe when sampling the N keys, we encounter a lot of good candidates, but we then just evict the best, and start from scratch again the next cycle.

start from scratch太傻了。於是Redis又做出了改進：采用緩沖池(pooling)

當每一輪移除Key時，拿到了這個N個Key的idle time，如果它的idle time比 pool 里面的 Key的idle time還要大，就把它添加到pool里面去。這樣一來，每次移除的Key並不僅僅是隨機選擇的N個Key里面最大的，而且還是pool里面idle time最大的，並且：pool 里面的Key是經過多輪比較篩選的，它的idle time 在概率上比隨機獲取的Key的idle time要大，可以這么理解：pool 里面的Key 保留了"歷史經驗信息"。

Basically when the N keys sampling was performed, it was used to populate a larger pool of keys (just 16 keys by default). This pool has the keys sorted by idle time, so new keys only enter the pool when they have an idle time greater than one key in the pool or when there is empty space in the pool.

采用"pool"，把一個全局排序問題 轉化成為了 局部的比較問題。(盡管排序本質上也是比較，囧)。要想知道idle time 最大的key，精確的LRU需要對全局的key的idle time排序，然后就能找出idle time最大的key了。但是可以采用一種近似的思想，即隨機采樣(samping)若干個key，這若干個key就代表着全局的key，把samping得到的key放到pool里面，每次采樣之后更新pool，使得pool里面總是保存着隨機選擇過的key的idle time最大的那些key。需要evict key時，直接從pool里面取出idle time最大的key，將之evict掉。這種思想是很值得借鑒的。

參考文獻

 

全面講解LRU算法 https://blog.csdn.net/belongtocode/article/details/102989685

Redis的LRU實現 https://www.cnblogs.com/hapjin/archive/2019/06/07/10933405.html