Python 實現LRU Cache

本文轉載自查看原文 2018-03-10 17:21 2036 Python相關/ web學習/ 網絡編程

LRU: 最近最少使用算法。使用場景：在有限的空間存儲對象時，當空間滿時，按照一定的原則刪除原有對象。常用的算法有LRU，FIFO，LFU。如memcached緩存系統即使用的LRU。

LRU的算法是比較簡單的，當對key進行訪問時（一般有查詢，更新，增加，在get()和set()兩個方法中實現即可）時，將該key放到隊列的最前端（或最后端）就行了，這樣就實現了對key按其最后一次訪問的時間降序（或升序）排列，當向空間中增加新對象時，如果空間滿了，刪除隊尾（或隊首）的對象。

在Python中，可以使用collections.OrderedDict很方便的實現LRU算法。

# coding: utf-8
import collections


class LRUCache(collections.OrderedDict):
    def __init__(self, size=5):
        self.size = size,
        self.cache = collections.OrderedDict()

    def get(self, key):
        if self.cache.has_key(key):
            val = self.cache.pop(key)
            self.cache[key] = val
        else:
            val = None

        return val

    def set(self, key, val):
        if self.cache.has_key(key):
            val = self.cache.pop(key)
            self.cache[key] = val
        else:
            if len(self.cache) == self.size:
                self.cache.popitem(last=False)
                self.cache[key] = val
            else:
                self.cache[key] = val


if __name__ == '__main__':
    """ test """
    cache = LRUCache(6)

    for i in range(10):
        cache.set(i, i)

    for i in range(10):
        import random

        i = random.randint(1, 20)
        print 'cache', cache.cache.keys()
        if cache.get(i):
            print 'hit, %s\n' % i
        else:
            print 'not hit, %s\n' % i

　　這個實現存在的問題：
1 . cache的value是能是不可變對象。
2. 在並發時，多個線程對緩存進行讀寫，那么必須對set()操作加鎖。TODO 實現一個支持並發訪問的LRU cache
3. value不能設置過期時間，而常用的redis和memcached都支持給value設置expire time。TODO 實現一個支持expire的LRU cache

Python 標准庫之 LRU 緩存實現學習

https://www.jianshu.com/p/f7258e266cc6

引言

LRU (Least Recently Used) 是緩存置換策略中的一種常用的算法。當緩存隊列已滿時，新的元素加入隊列時，需要從現有隊列中移除一個元素，LRU 策略就是將最近最少被訪問的元素移除，從而騰出空間給新的元素。

研讀 Python 3.6 中 functools.lru_cache 源碼可以發現，它是通過一個雙向鏈表加字典實現 LRU 緩存的。下面就來學習一下這個工具函數的實現。

應用

在深入學習該函數之前，我們可以看看它的常規用法。合理使用緩存，可以有效地減少一些長耗時函數調用的次數，從而大大提高整體效率。

看一個經典的例子，即斐波那契函數的遞歸實現：

def fibonacci(n): if n == 0: return 0 if n == 1: return 1 return fibonacci(n - 1) + fibonacci(n - 2)

眾所周知，當需要計算的 N 比較大時，上述函數計算會非常緩慢。我們先來分析下為何上述函數在計算較大 N 時會耗時很久，以便了解為何可以使用緩存機制來提高效率。以下是 N 為 5 時上述函數遞歸調用樹狀圖：

Snip20170915_96

顯然，在調用過程中，有多次重復計算。於是，我們可以添加 lru_cache 裝飾器緩存已經計算過的數據，從而改善遞歸版的斐波那契函數：

@lru_cache() def fibonacci(n): if n == 0: return 0 if n == 1: return 1 return fibonacci(n - 1) + fibonacci(n - 2)

當 N = 32 時，可以對比下兩個版本計算耗時，可以看到計算效率的提升是驚人的：

fibonacci(32) = 2178309

# 沒有加緩存的遞歸版本 Elapsed time: 1497.54ms # 添加緩存的遞歸版本 Elapsed time: 0.16ms

當然啦，事實上我們還有更好的方法來實現斐波那契函數（時間復雜度 O(n)），示例如下：

def fibonacci_fast(n): a, b = 0, 1 for _ in range(n): a, b = a + b, a return a

貌似跑偏了，接下來趕緊進入正題，窺探下 lru_cache 是如何實現 LRU 緩存的。

LRU 緩存實現

查看源碼，可以看到 LRU 緩存是在函數 _lru_cache_wrapper 中實現的。本節只研究 LRU 是如何在其中實現的，所以，下面的源碼中移除了無關的代碼。

def _lru_cache_wrapper(user_function, maxsize, typed, _CacheInfo): # 所有 LRU 緩存元素共享的常量: sentinel = object() # 特殊標記，用來表示緩存未命中 make_key = _make_key # 根據函數參數生成緩存 key # # --------------------------------- # | PREV | DATA(KEY+RESULT) | NEXT| # --------------------------------- # PREV, NEXT, KEY, RESULT = 0, 1, 2, 3 # 鏈表各個域 # 存放 key 到 node 的映射 cache = {} full = False cache_get = cache.get lock = RLock() # 鏈表更新不是線程安全的，所以需要加鎖 root = [] # 關鍵：環形雙向鏈表 # 根節點兩側分別是訪問頻率較高和較低的節點 root[:] = [root, root, None, None] # 初始根節點（相當於一個空的頭節點） def wrapper(*args, **kwds): nonlocal root, full key = make_key(args, kwds, typed) with lock: link = cache_get(key) if link is not None: # 緩存命中 # 將被訪問的節點移動到環形鏈表的前面（即 root 的前邊） link_prev, link_next, _key, result = link link_prev[NEXT] = link_next link_next[PREV] = link_prev last = root[PREV] last[NEXT] = root[PREV] = link link[PREV] = last link[NEXT] = root return result # 緩存未命中，調用用戶函數生成 RESULT result = user_function(*args, **kwds) with lock: if key in cache: # 考慮到此時鎖已經釋放，而且 key 已經被緩存了，就意味着上面的 # 節點移動已經做了，緩存也更新了，所以此時什么都不用做。 pass elif full: # 新增緩存結果，移除訪問頻率低的節點 # 下面的操作是使用 root 當前指向的節點存儲 KEY 和 RESULT oldroot = root oldroot[KEY] = key oldroot[RESULT] = result # 接下來將原 root 指向的下一個節點作為新的 root， # 同時將新 root 節點的 KEY 和 RESULT 清空，這樣 # 使用頻率最低的節點結果就從緩存中移除了。 root = oldroot[NEXT] oldkey = root[KEY] oldresult = root[RESULT] root[KEY] = root[RESULT] = None del cache[oldkey] cache[key] = oldroot else: # 僅僅新增緩存結果 # 新增節點插入到 root 節點的前面 last = root[PREV] link = [last, root, key, result] last[NEXT] = root[PREV] = cache[key] = link full = (len(cache) >= maxsize) return result return wrapper