高效能快取解決方案——深入解析 AioCache 套件

Published February 12, 2025 by 徐培鈞
Python

在現代應用程式開發中,快取(Cache) 是提升系統效能、減少伺服器負擔的重要技術之一。

透過快取機制,我們可以在短時間內存取已計算或查詢過的資料,避免重複計算或頻繁存取資料庫,提高應用程式的運行速度。

在 Python 生態系統中,有許多不同的快取解決方案,而 AioCache 是專為 異步(Asynchronous) 應用設計的一款強大快取套件。

它不僅支援多種快取後端(如 Redis、Memcached、內存等),還能與 asyncio 無縫整合,適用於高併發、非同步的應用場景。

本文將詳細介紹 AioCache 的特性、安裝方法、核心功能與應用場景,幫助開發者快速掌握這款套件的使用方式。


什麼是 AioCache?

AioCache 是一款基於 Python asyncio 設計的快取庫,主要用於提高應用程式的執行效率。其主要特點包括:

  • 支援多種快取後端:內建支援 Memory、Redis、Memcached、Disk 等不同類型的快取。
  • 非同步支持:完全基於 asyncio,適用於需要高效能的異步應用,如 FastAPI、Sanic 等。
  • 函式與類別快取裝飾器:提供直覺的裝飾器,讓開發者能輕鬆對函式或類別方法進行快取處理。
  • TTL(存活時間)與自動過期:可以設定快取的有效時間,避免過期數據影響業務邏輯。
  • 內建序列化支援:支援 JSON、Pickle 等序列化格式,使得快取數據的存取更加靈活。

由於這些特性,AioCache 成為異步應用開發者實現快取機制的理想選擇。

延伸閱讀:什麼是 asyncio?——Python 的非同步編程核心


快取與非同步的關係是什麼?

快取(Cache)是一種用來 加速數據存取減少重複計算降低資料庫查詢負擔 的技術。

非同步(Asynchronous) 則是一種讓應用程式 同時處理多個 I/O 任務,而不會彼此阻塞 的編程方式。

高併發應用(如 API 服務、即時數據處理、爬蟲等) 中,非同步與快取的結合能夠發揮 最大效能優勢

快取的角色:減少 I/O 負擔

假設一個 Web API 需要查詢資料庫,每次查詢都會有一定延遲(例如 100ms)。

如果某個請求的數據可以快取,那麼應用程式就可以 直接從記憶體或快取後端(如 Redis)中讀取資料,而不必等待資料庫的回應

📍 問題:如果快取存取本身也是同步操作,會發生什麼?

  • 如果使用同步快取,當快取讀取或寫入時,應用程式 仍然會阻塞,等待快取完成後才繼續執行其他任務。
  • 在高併發環境中,這種阻塞可能會降低整體吞吐量(throughput),影響應用效能。

非同步快取的優勢

非同步快取(如 AioCache)能夠與 asyncio 兼容,使應用程式在存取快取時不會被阻塞。

這樣可以讓應用程式 同時處理多個請求,即使快取 I/O 需要時間,也不會影響其他任務的執行

🔹 同步 vs. 非同步 快取的對比

同步快取(blocking)應用程式會等待快取讀取完成後才繼續執行其他任務
非同步快取(async / non-blocking)讀取快取時,應用程式可以繼續執行其他協程,不會被阻塞
同步快取(blocking)當快取更新時,應用程式無法執行其他請求
非同步快取(async / non-blocking)寫入快取的同時,應用程式可以處理其他請求,提高並發能力
同步快取(blocking)低併發、小型應用
非同步快取(async / non-blocking)高併發、大型應用(API、爬蟲、即時數據處理)

AioCache 如何結合 asyncio 提供非同步快取?

🔴 使用同步快取(阻塞)

如果使用 傳統的同步 Redis 或記憶體快取,存取時會發生 等待

import time

cache = {}  # 簡單的字典作為同步快取

def get_from_cache(key):
    time.sleep(1)  # 模擬快取讀取延遲
    return cache.get(key, "未找到快取")

def main():
    print(get_from_cache("user:123"))  # 這裡會等待 1 秒,影響應用效率
    print("後續邏輯執行")  # 這行代碼會被阻塞,直到快取讀取完成

main()

問題:每次讀取快取時,應用程式都會卡住,無法執行其他任務!

🟢 使用 AioCache(非同步快取,不阻塞)

如果使用 AioCache(基於 asyncio,應用程式在快取讀取時不會阻塞:

import asyncio
from aiocache import Cache

cache = Cache(Cache.MEMORY)  # 設定為記憶體快取

async def get_from_cache(key):
    await asyncio.sleep(1)  # 模擬快取讀取延遲
    return await cache.get(key, default="未找到快取")

async def main():
    task1 = asyncio.create_task(get_from_cache("user:123"))
    task2 = asyncio.create_task(get_from_cache("user:456"))
    
    print(await task1)  # 這兩個讀取可以同時進行,不會互相阻塞
    print(await task2)
    print("後續邏輯執行")  # 這行代碼可以在等待快取時執行,提高效率

asyncio.run(main())

🚀 好處:即使快取讀取需要時間,應用程式也可以繼續執行其他任務,提升並發能力!

非同步快取的適用場景

使用 AioCache(非同步快取) 特別適合以下場景:

高併發 API 服務(如 FastAPI、Sanic)
減少資料庫查詢,同時處理多個快取存取,不阻塞其他請求。

即時數據處理(如爬蟲)
避免等待快取 I/O,提升爬蟲效率,讓應用程式可以同時處理多個任務。

WebSocket 或即時聊天應用
快取訊息狀態時,不影響其他用戶的請求處理,確保即時性。


如何安裝 AioCache?

AioCache 可以透過 pip 直接安裝:

pip install aiocache

如果需要使用 Redis 或 Memcached 作為後端,則需安裝對應的額外依賴:

pip install aiocache[redis]
pip install aiocache[memcached]

安裝完成後,即可在 Python 應用中使用 AioCache 來管理快取。


AioCache 的核心功能與用法

創建快取實例

AioCache 允許開發者指定不同的快取後端,例如記憶體(Memory)快取或 Redis 快取:

from aiocache import Cache

# 使用 Memory 進行快取
cache = Cache(Cache.MEMORY)

# 使用 Redis 進行快取
cache = Cache(Cache.REDIS, endpoint="127.0.0.1", port=6379)

基本快取操作(設置、獲取、刪除、清空)

AioCache 提供了簡單直覺的 API 來進行快取管理,主要包含以下四種基本操作:

方法cache.set(key, value, ttl=秒數)
功能設定快取值,可指定存活時間(TTL,Time To Live)
方法cache.get(key, default=None)
功能讀取快取值,若鍵不存在則返回 default
方法cache.delete(key)
功能移除特定快取
方法cache.clear()
功能清除所有快取資料

🔹 📌 示範程式:如何進行基本快取操作

import asyncio
from aiocache import Cache

# 初始化快取(使用記憶體 Memory)
cache = Cache(Cache.MEMORY)

async def main():
    # 設定快取:key 為 "key",值為 "value",存活時間 10 秒
    await cache.set("key", "value", ttl=10)
    
    # 獲取快取
    value = await cache.get("key")  
    print(f"快取值: {value}")  # 輸出: 快取值: value

    # 刪除特定快取
    await cache.delete("key")
    
    # 確認是否刪除成功
    deleted_value = await cache.get("key", default="快取已刪除")
    print(deleted_value)  # 輸出: 快取已刪除

    # 清空所有快取
    await cache.clear()
    print("所有快取已清除")

asyncio.run(main())

🔹 🔍 重點解析:

  • TTL 設定快取過期時間,當時間到了後,快取值會自動刪除。
  • get() 獲取的 key 不存在時,可返回 default,避免程式報錯。
  • cache.clear() 可以清除所有快取,適用於快取重置場景

使用裝飾器快取函式結果

在許多情況下,我們希望 減少函式的重複計算或查詢,這時可以使用 @cached 裝飾器 來快取函式結果,讓相同的輸入 在指定時間內不需要重新計算或查詢

🔹 📌 示範程式:使用 @cached 快取函式返回值

import asyncio
from aiocache import cached

@cached(ttl=10)  # 設定快取 10 秒
async def get_data():
    print("📡 正在從資料庫獲取數據...")
    return {"data": "example"}

async def main():
    print(await get_data())  # 第一次調用時會執行函式並快取結果
    print(await get_data())  # 第二次調用時直接從快取獲取,不會執行函式

asyncio.run(main())

🔹 🔍 執行結果:

📡 正在從資料庫獲取數據...
{'data': 'example'}
{'data': 'example'}

🔹 🔍 重點解析:

  • 第一次調用 get_data() 會執行函式,並將結果存入快取(TTL 為 10 秒)。
  • 第二次調用 get_data(),會直接從快取讀取數據,而不會執行函式,避免不必要的計算或資料庫請求。
  • 適用場景
    • 頻繁查詢的 API 端點
    • 高成本計算的函式(如數據分析、AI 預測)
    • 頻繁變更但可短時間快取的結果

使用 Redis 作為快取後端

雖然 Memory 快取(記憶體) 是最快的,但它 無法跨應用共享,並且 重啟應用後快取會遺失

為了讓快取能夠跨多個應用存取,並且具備持久化能力,我們可以 使用 Redis 作為後端快取

📌 安裝 Redis 版本的 AioCache

在安裝 aiocache 時,請確保安裝 Redis 依賴:

pip install aiocache[redis]

📌 示範程式:使用 Redis 進行快取存取

import asyncio
from aiocache import Cache

# 初始化 Redis 快取(假設 Redis 伺服器運行於 localhost:6379)
cache = Cache(Cache.REDIS, endpoint="localhost", port=6379)

async def main():
    # 設定快取:存放用戶數據,TTL 為 30 秒
    await cache.set("user:123", {"name": "Alice", "age": 25}, ttl=30)
    
    # 讀取快取
    user = await cache.get("user:123")
    print(f"快取用戶數據: {user}")  # 輸出: {'name': 'Alice', 'age': 25}

    # 刪除快取
    await cache.delete("user:123")

asyncio.run(main())

🔹 🔍 重點解析:

  • 透過 Cache(Cache.REDIS, endpoint="localhost", port=6379) 設定 Redis 為快取後端
  • 快取的數據會存入 Redis,即使應用程式重新啟動,數據仍然存在
  • 適用於分佈式應用,多個應用程式可以共享相同的快取數據。

AioCache 的應用場景

Web 應用加速

對於高流量的 Web 應用(如 API 服務),可以透過 AioCache 快取頻繁訪問的數據(如用戶資訊、熱門商品等),減少資料庫查詢壓力,提升響應速度。

高併發的非同步任務

在需要大量請求的場景(如爬蟲、批量數據處理),AioCache 能夠快取已處理的結果,避免重複計算,提升執行效率。

分布式快取

結合 Redis 作為後端時,AioCache 可以在多個服務節點間共享快取數據,適用於微服務架構的應用。


結論

AioCache 是一款專為 異步 Python 應用 設計的強大快取工具,提供了多種快取後端支持、非同步操作以及簡單易用的快取管理功能。

如果你的應用需要處理高併發請求,或希望提升資料查詢效能,AioCache 是一個值得考慮的選擇。

透過合理使用快取策略,可以顯著減少資料庫負擔,提升系統效能,讓你的應用更加高效!