Python結(jié)合diskcache實現(xiàn)磁盤緩存工具

更新時間：2026年03月27日 09:38:05 作者：Looooking

之前寫了?cachetools 的緩存工具,那個是純內(nèi)存的,性能上確實有優(yōu)勢,但重啟后緩存數(shù)據(jù)會丟失,下面小編就和大家詳細介紹一下Python如何結(jié)合diskcache實現(xiàn)磁盤緩存工具吧

之前寫了 cachetools 的緩存工具，那個是純內(nèi)存的，性能上確實有優(yōu)勢，但重啟后緩存數(shù)據(jù)會丟失。diskcache 則利用輕量級的 sqlite 數(shù)據(jù)庫，該數(shù)據(jù)庫不需要單獨的服務(wù)器進程，并可以持久化數(shù)據(jù)結(jié)構(gòu)，且可以突破內(nèi)存的限制，針對大量數(shù)據(jù)的緩存時，不會因為內(nèi)存溢出而丟失數(shù)據(jù)。

特性	diskcache	cachetools
存儲位置	磁盤為主（內(nèi)存為輔）	純內(nèi)存
持久化	? 支持（重啟后數(shù)據(jù)還在）	? 不支持
數(shù)據(jù)大小	適合大數(shù)據(jù)（受磁盤限制）	適合小數(shù)據(jù)（受內(nèi)存限制）
速度	磁盤I/O較慢	純內(nèi)存很快
使用場景	長期緩存、大數(shù)據(jù)	短期緩存、小數(shù)據(jù)

安裝

pip install diskcache

淘汰策略

從源碼的 EVICTION_POLICY 值可以看出，淘汰策略主要有以下幾種。

'least-recently-stored' - 默認，按存儲時間淘汰
'least-recently-used' - 按訪問時間淘汰（每次訪問都寫數(shù)據(jù)庫）
'least-frequently-used' - 按訪問頻率淘汰（每次訪問都寫數(shù)據(jù)庫）
'none' - 禁用自動淘汰

默認則是按照 LRS 按緩存存儲的先后時間進行淘汰的淘汰策略。

簡單存取

from diskcache import Cache

# 1. 實例一個緩存對象
# 需要傳入目錄路徑。如果目錄路徑不存在，將創(chuàng)建該路徑，并且會在指定位置創(chuàng)建一個cache.db的文件。
# 如果未指定，則會自動創(chuàng)建一個臨時目錄，并且會在指定位置創(chuàng)建一個cache.db的文件。
cache = Cache("cache")

# 2. 保存緩存
cache.set('name', '張三', expire=60, read=True, tag='姓名', retry=True)

# 3. 獲取緩存 expire_time為真，返回有效期時間；tag為真，返回緩存設(shè)置的tag值；
name = cache.get('name', default=False, expire_time=True, tag=True)
print(name)
# ('張三', 1770617370.6258903, '姓名')

上面代碼執(zhí)行之后，我們可以在當(dāng)前位置的下發(fā)現(xiàn)有個 cache 目錄，其中有個 cache.db 文件。因為這個 cache.db 是個 sqlite 數(shù)據(jù)庫文件，我們可以嘗試使用 pandas 讀取一下。

import pandas as pd

from sqlalchemy import create_engine

engine = create_engine('sqlite:///cache/cache.db')
pd.set_option('display.max_columns', None)  # 不限制列數(shù)
pd.set_option('display.width', None)  # 不限制列寬

if __name__ == '__main__':
    res = pd.read_sql('SELECT * FROM cache;', con=engine)
    print(res)

rowid key raw store_time expire_time access_time access_count tag size mode filename value
0 1 name 1 1.770605e+09 1.770605e+09 1.770605e+09 0 姓名 0 1 None 張三

從查詢結(jié)果中可以看出，數(shù)據(jù)庫除了常規(guī)的 key 和 value，還有過期時間戳 expire_time，tag, access_count 等參數(shù)。

緩存獲取

獲取指定

正常使用 get() 方法獲取緩存，但這種方法如果緩存不存或者緩存已過期的話，就只是返回 None 而不會報錯。

from diskcache import Cache

cache = Cache("cache")
print(cache.get('key_not_exist'))
# None

還有就是使用 read() 方法獲取緩存，Key 不存在或已過期則直接報 KeyError 的錯。具體使用哪種看自己的使用場景。

from diskcache import Cache

cache = Cache("cache")
print(cache.read('key_not_exist'))
# Traceback (most recent call last):
#   File "E:\lky_project\test_project\test4.py", line 4, in <module>
#     print(cache.read('key_not_exist'))
#           ^^^^^^^^^^^^^^^^^^^^^^^^^^^
#   File "E:\myenv\py11_base_env\Lib\site-packages\diskcache\core.py", line 1252, in read
#     raise KeyError(key)
# KeyError: 'key_not_exist'

獲取最近

有時候，我們想知道最近（最遠）存儲的緩存是哪個，則可以使用 peekitem() 方法獲取最近（最遠）存儲的緩存數(shù)據(jù)。

注意：該操作會自動清空數(shù)據(jù)庫中已過期的緩存。

from diskcache import Cache

cache = Cache("cache")
print(cache.peekitem())
# ('key5', 'value5')
print(cache.peekitem(last=False))
# ('key1', 'value1')

緩存過期

設(shè)置緩存的時候（無論是使用 set() 還是 add() 方法），都有一個過期時長參數(shù) expire 指定該緩存多久后過期。

from diskcache import Cache

cache = Cache("cache")
result = cache.add("key1", "value1", expire=60)
print(result)
# True
result = cache.add("key2", "value2", expire=30)
print(result)
# True

如果緩存已過期，則返回的對應(yīng)的 value 為 None（當(dāng)然，如果指定了默認值 default，則返回默認值，比如下面的 False）。

from diskcache import Cache

cache = Cache("cache")

name = cache.get('name', default=False, expire_time=True, tag=True)
print(name)
# (False, None, None)

但其實這個時候，數(shù)據(jù)仍然在數(shù)據(jù)庫里的，只不過因為當(dāng)前時間超過了過期時間 expire_time，所以不會將這個數(shù)據(jù)取回來。

import pandas as pd

from sqlalchemy import create_engine

engine = create_engine('sqlite:///cache/cache.db')
pd.set_option('display.max_columns', None)  # 不限制列數(shù)
pd.set_option('display.width', None)  # 不限制列寬

if __name__ == '__main__':
    res = pd.read_sql('SELECT * FROM cache;', con=engine)
    print(res)
    #    rowid   key  raw    store_time   expire_time   access_time  access_count tag  size  mode filename value
    # 0      1  name    1  1.770617e+09  1.770617e+09  1.770617e+09             0  姓名     0     1     None    張三

緩存清理

清空過期

使用 expire() 方法可以清空過期緩存，并返回清空的緩存數(shù)。

import time
from diskcache import Cache

cache = Cache("cache")

cache.set('name', '張三', expire=60, read=True, tag='姓名', retry=True)
cache.set('age', 30, expire=30, read=True, tag='年齡', retry=True)
time.sleep(35)
count = cache.expire()
print(count)
# 1

清空所有

使用 clear() 方法可以清空所有緩存，并返回清空的緩存數(shù)。

from diskcache import Cache

cache = Cache("cache")

cache.set('name', '張三', expire=60, read=True, tag='姓名', retry=True)
cache.set('age', 30, expire=60, read=True, tag='年齡', retry=True)
count = cache.clear()
print(count)
# 2

強制清理

使用 cull() 方法會先清理過期緩存，再按照給定的緩存淘汰策略，清理緩存直到磁盤緩存容量小于size_limit 的大小。

from diskcache import Cache

cache = Cache("cache")

cache.set('name', '張三', expire=60, read=True, tag='姓名', retry=True)
cache.set('age', 30, expire=60, read=True, tag='年齡', retry=True)
count = cache.cull()
print(count)

按 tag 清理

使用 evict() 方法可以手動按照 tag 的名稱清理緩存，并返回清空的緩存數(shù)。

from diskcache import Cache

cache = Cache("cache")

cache.set('name', '張三', expire=60, read=True, tag='姓名', retry=True)
cache.set('age', 30, expire=60, read=True, tag='年齡', retry=True)
count = cache.evict("年齡")
print(count)
# 1

按 key 清理

delete() 刪除并返回是否刪除成功。

from diskcache import Cache

cache = Cache("cache")

cache.set('name', '張三', expire=60, read=True, tag='姓名', retry=True)
cache.set('age', 30, expire=60, read=True, tag='年齡', retry=True)
result = cache.delete("name")
print(result)
# True

pop() 刪除并返回緩存的 value。

from diskcache import Cache

cache = Cache("cache")

cache.set('name', '張三', expire=60, read=True, tag='姓名', retry=True)
cache.set('age', 30, expire=60, read=True, tag='年齡', retry=True)
result = cache.pop("name")
print(result)
# 張三

緩存添加

可以使用 add() 方法僅添加緩存。

只有鍵不存在時才會存儲
如果鍵已存在，不會覆蓋，返回 False
相當(dāng)于 setdefault() 的行為

from diskcache import Cache

cache = Cache("cache")
result = cache.add("key1", "value1")
print(result)
# True

當(dāng)然，set() 方法也可以設(shè)置緩存。與 add() 的區(qū)別如下。

總是會存儲值，無論鍵是否已存在
如果鍵已存在，會覆蓋舊值
相當(dāng)于字典的 dict[key] = value

緩存刷新

針對未過期的 key，重新刷新 key 的過期時間。如果 key 已過期，則會刷新失敗。

from diskcache import Cache

cache = Cache("cache")
result = cache.touch("key1", 60)
print(result)
# True

緩存判斷

Cache 中定義了 __contains__ 魔法方法，可以用 in 的方式判斷對應(yīng)的 key 是否在緩存中。

from diskcache import Cache

cache = Cache("cache")
cache.set("key1", "value1")
print("key1" in cache)
# True

也可以使用 get() 方法，看返回的結(jié)果是不是 None 來進行判斷（但這種判斷并不精確，如果 value 值本身是 None 的話，則會造成誤判）。

當(dāng)然，還可以使用 read() 方法，Key 不存在的話直接報 KeyError，這種需要自己處理這種異常。

緩存修改

針對整型的緩存數(shù)據(jù)類型，可以使用 incr() 和 decr() 對 value 進行加減指定 delta 的數(shù)值。

from diskcache import Cache

cache = Cache("cache")

cache.set('age', 30, expire=60, read=True, tag='年齡', retry=True)
cache.incr("age", delta=5)
print(cache.get("age"))  # 35
cache.decr("age", delta=3)
print(cache.get("age"))  # 32

緩存檢查

check() 方法會對數(shù)據(jù)庫和文件系統(tǒng)的一致性進行檢查，如果有告警，則返回告警信息。

from diskcache import Cache

cache = Cache("cache")
warnings = cache.check()
print(warnings)
# []

自動緩存

像之前的 cachetools 或標(biāo)準(zhǔn)庫的 lru_cache，都是可以直接裝飾在函數(shù)上，實現(xiàn)對函數(shù)調(diào)用自動緩存的，diskcache 也可以使用緩存實例的 memoize() 方法來裝飾函數(shù)，實現(xiàn)調(diào)用結(jié)果自動緩存。

from diskcache import Cache

cache = Cache('cache')


# 使用cache.memoize()裝飾器自動緩存函數(shù)結(jié)果
@cache.memoize(expire=60)
def compute_expensive_operation(x):
    # 模擬耗時計算
    print("call function")
    return x * x * x


# 第一次調(diào)用，計算結(jié)果并緩存
result = compute_expensive_operation(3)
print(result)  # 輸出: call function 和 27

# 第二次調(diào)用，直接從緩存獲取結(jié)果，不會再真正執(zhí)行函數(shù)
result = compute_expensive_operation(3)
print(result)  # 輸出: 27

看源碼可知，它會自動將輸入?yún)?shù)等按照指定流程轉(zhuǎn)換成元組（這樣能確保相同輸入轉(zhuǎn)換后的 key 始終一致）作為緩存的 key，每次函數(shù)調(diào)用時，如果已經(jīng)有獲取到緩存結(jié)果，則直接返回；如果還沒有緩存結(jié)果，則調(diào)用函數(shù)，并將結(jié)果緩存并返回。

            def wrapper(*args, **kwargs):
                """Wrapper for callable to cache arguments and return values."""
                key = wrapper.__cache_key__(*args, **kwargs)
                result = self.get(key, default=ENOVAL, retry=True)

                if result is ENOVAL:
                    result = func(*args, **kwargs)
                    if expire is None or expire > 0:
                        self.set(key, result, expire, tag=tag, retry=True)

                return result

隊列操作

我們可以使用相同前綴的 key 來約定一個隊列（簡單理解就是相同前綴的 key 同屬于一個隊列），然后通過 push() 和 pull() 方法來實現(xiàn)入列和出列。

隊列推送

可以使用 push() 方法往指定前綴的隊列推送 value（key 則由方法根據(jù)前綴及緩存中已有的相同前綴的 key 來自動生成）。

from diskcache import Cache

cache = Cache("cache")
result = cache.push("first", prefix="test")
print(result)  # test-500000000000000
result = cache.push("second", prefix="test")
print(result)  # test-500000000000001
result = cache.push("third", prefix="test", side="front")  # 從隊列前面插入
print(result)  # test-499999999999999

隊列拉取

出隊的順序與入隊順序相同（遵從 FIFO 先進先出的規(guī)則）。

from diskcache import Cache

cache = Cache("cache")
result = cache.pull("test")
print(result)  # ('test-499999999999999', 'third')
result = cache.pull("test")
print(result)  # ('test-500000000000000', 'first')
result = cache.pull("test")
print(result)  # ('test-500000000000001', 'second')

當(dāng)然，我們也可以通過參數(shù) side 指定從頭部出列還是尾部出列。

from diskcache import Cache

cache = Cache("cache")
result = cache.pull("test")
print(result)  # ('test-499999999999999', 'third')
result = cache.pull("test", side="back")
print(result)  # ('test-500000000000001', 'second')
result = cache.pull("test")
print(result)  # ('test-500000000000000', 'first')

隊列查看

我們可以使用 peek() 方法，查看指定前綴的隊列的隊首和隊尾的數(shù)據(jù)。

注意：不同于 pull() 方法，peek() 只是查看而不會刪除未過期的緩存（過期緩存仍然會自動清理），畢竟從 peek 這個單詞的意思本身就是窺視，偷看的意思。

from diskcache import Cache

cache = Cache("cache")
print(cache.peek(prefix="test"))
# ('test-499999999999999', 'third')
print(cache.peek(prefix="test", side="back"))
# ('test-500000000000001', 'second')

連接關(guān)閉

使用 close() 方法可以關(guān)閉與底層 sqlite 數(shù)據(jù)庫的連接。

from diskcache import Cache

cache = Cache("cache")
cache.close()

配置重置

緩存的配置存儲在 Settings 表中。

SELECT key, value FROM Settings;

key value
0 count 0
1 size 0
2 hits 0
3 misses 0
4 statistics 0
5 tag_index 0
6 eviction_policy least-recently-stored
7 size_limit 1073741824
8 cull_limit 10
9 sqlite_auto_vacuum 1
10 sqlite_cache_size 8192
11 sqlite_journal_mode wal
12 sqlite_mmap_size 67108864
13 sqlite_synchronous 1
14 disk_min_file_size 32768
15 disk_pickle_protocol 5

我們可以使用 reset() 方法重置 Settings 中對應(yīng) key 的 value。比如我們設(shè)置 cull_limit 值為 30。

from diskcache import Cache

cache = Cache("cache")

value = cache.reset("cull_limit", 30)
print(value)
# 30

再次查看 Settings 表時，可以看到 cull_limit 的值已被修改。

key value
0 count 2
1 size 0
2 hits 0
3 misses 0
4 statistics 0
5 tag_index 0
6 eviction_policy least-recently-stored
7 size_limit 1073741824
8 cull_limit 30
...

配置查看

如果我么僅僅是想查看配置的值，而不需要修改的話，則在使用 reset() 方法時，不需要傳入對應(yīng)的值，且 update 置為 False 即可。

from diskcache import Cache

cache = Cache("cache")

value = cache.reset("cull_limit", update=False)
print(value)
# 10

事務(wù)操作

可以使用事務(wù)的方式批量進行操作。

from diskcache import Cache

cache = Cache("cache")
items = {"key1": "value1", "key2": "value2", "key3": "value3"}

with cache.transact():  # 性能提升2-5倍
    for key, value in items.items():
        cache[key] = value

命中統(tǒng)計

from diskcache import Cache

cache = Cache("cache")
# 啟用統(tǒng)計
cache.stats(enable=True)

# 獲取命中率
hits, misses = cache.stats()
print(f"hits: {hits}; misses: {misses}")  # hits: 5; misses: 2

# 檢查緩存一致性
warnings = cache.check()
print(warnings)  # []

# 獲取緩存體積（字節(jié)）
size = cache.volume()
print(size)  # 32768

Deque 隊列緩存

估計是怕我們直接用 Cache 原生的 push() 和 pull() 方法來操作隊列太不方便，diskcache 還幫我們封裝實現(xiàn)了一個 Deque 類（其中大部分操作也是基于 Cache 的 push() 和 pull() 方法來實現(xiàn)）。

from diskcache import Deque

cache = Deque([0, 1, 2, 3, 4], "cache")
cache.append(5)
print(list(cache))
# [0, 1, 2, 3, 4, 5]

其 key 值則是無前綴的 key。其他操作則與標(biāo)準(zhǔn)庫 collections 中的 deque 基本類似，就不贅述了。不同之處在于每個修改操作都會持久化到數(shù)據(jù)庫，感興趣的可以看 Deque 中方法的源碼實現(xiàn)。

rowid key raw store_time expire_time access_time access_count tag size mode filename value
0 1 500000000000000 1 1.770714e+09 None 1.770714e+09 0 None 0 1 None 0
1 2 500000000000001 1 1.770714e+09 None 1.770714e+09 0 None 0 1 None 1
2 3 500000000000002 1 1.770714e+09 None 1.770714e+09 0 None 0 1 None 2
3 4 500000000000003 1 1.770714e+09 None 1.770714e+09 0 None 0 1 None 3
4 5 500000000000004 1 1.770714e+09 None 1.770714e+09 0 None 0 1 None 4
5 6 500000000000005 1 1.770714e+09 None 1.770714e+09 0 None 0 1 None 5

Index 索引緩存

Index 組件提供了類似字典的持久化存儲方案（從源碼中也可以看出，它默認沒有設(shè)置緩存淘汰策略，也即永不過期）。它支持事務(wù)操作，確保數(shù)據(jù)一致性，同時保持高效的讀寫性能。

self._cache = Cache(directory, eviction_policy='none')

from diskcache import Index

# 持久化字典，永不淘汰
index = Index("cache", {"a": 1, "b": 2}, c=3)

index.update([('d', 4), ('e', 5)])
print(list(index))
# ['a', 'b', 'c', 'd', 'e']

我們也可以從通用的 Cache() 實例來創(chuàng)建 index。

from diskcache import Cache, Index

cache = Cache("cache")
index = Index.fromcache(cache)

Index 緩存的設(shè)置和獲取和字典一樣（Index 中定義了 __getitem__，__setitem__，__delitem__ 等字典類型的魔法方法）。封裝的其他方法也基本上字典一致。

from diskcache import Index

index = Index("cache")
index['f'] = 6
print(index.get('c'))  # 3

還可以像字典一樣 setdefault，對應(yīng)的 key 有值時，返回緩存的值；沒有值時，使用給定的值設(shè)置緩存并返回。

from diskcache import Index

index = Index("cache")
result = index.setdefault('h', 9)
print(result)  # 8

Fanout 分片緩存

FanoutCache提供了橫向擴展能力，它的一致性哈希算法確保了數(shù)據(jù)均勻分布，將數(shù)據(jù)自動分片到多個子緩存，同時支持動態(tài)擴容和多進程操作。

from diskcache import FanoutCache

# 初始化分布式緩存，8個分片，每個分片1GB大小限制
distributed_cache = FanoutCache(
    directory='./cache',
    shards=8,
    size_limit=1024 ** 3  # 1GB per shard
)

# 在多進程環(huán)境中共享使用
from multiprocessing import Pool


def fetch_page(url):
    return f"content of {url}"


def spider_worker(url):
    # 自動路由到對應(yīng)分片
    if distributed_cache.get(url, default=None) is None:
        data = fetch_page(url)
        distributed_cache.set(url, data, expire=86400)  # 緩存24小時
    return url


url_list = [
    "https://example1.com",
    "https://example2.com",
    "https://example3.com",
    "https://example4.com",
]

if __name__ == '__main__':
    with Pool(processes=8) as pool:
        # 并行處理URL列表
        results = pool.map(spider_worker, url_list)

此時，diskcache 會按照分片數(shù)目，在指定目錄下建立分片個數(shù)目的目錄，并將緩存均勻分散存儲在這些目錄中的 cache.db 數(shù)據(jù)庫中。源碼中的 self._shards 也可以看出他新建分片緩存目錄的邏輯。

        self._shards = tuple(
            Cache(
                directory=op.join(directory, '%03d' % num),
                timeout=timeout,
                disk=disk,
                size_limit=size_limit,
                **settings,
            )
            for num in range(shards)
        )

注意：上面多進程示例，進程池的 with 塊需要放到 __main__ 塊中執(zhí)行，不然會因為安全因素導(dǎo)致執(zhí)行失敗。

下面是其中一個 001 分片的 cache.db 中存儲的緩存數(shù)據(jù)。

import pandas as pd

from sqlalchemy import create_engine

engine = create_engine('sqlite:///cache/001/cache.db')
pd.set_option('display.max_columns', None)  # 不限制列數(shù)
pd.set_option('display.width', None)  # 不限制列寬

if __name__ == '__main__':
    res = pd.read_sql('SELECT * FROM cache;', con=engine)
    # res = pd.read_sql('SELECT key, value FROM Settings;', con=engine)
    print(res)

# rowid key raw store_time expire_time access_time access_count tag size mode filename value
# 0 1 https://example4.com 1 1.770632e+09 1.770718e+09 1.770632e+09 0 None 0 1 None content of https://example4.com

其他的緩存操作方法與 Cache 類似，只是針對每個 key，會通過相同的 Hash 算法先找到其所屬的正確的分片，然后再進行操作。

以上就是Python結(jié)合diskcache實現(xiàn)磁盤緩存工具的詳細內(nèi)容，更多關(guān)于Python磁盤緩存的資料請關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章:

最新国产好看的视频,伊人天堂AV在线,国产Aaaaaa视频,蜜臀视频在线观看一区,人妻av色图,密臀久久久精品影片,青青视频免费观看毛片,久草在线观看视,国产三级精品色情在线