Posted inPython

Hướng dẫn sử dụng DiskCache trong Python: Caching trên đĩa an toàn đa tiến trình, không cần Redis

Python tutorial - IT technology blog
Python tutorial - IT technology blog

Vấn đề thực tế: Bài toán nghẽn RAM trên VPS 1GB

Vài năm trước, mình nhận một batch job xử lý báo cáo tài chính trên con VPS giá rẻ vỏn vẹn 1GB RAM. Nhiệm vụ của script là cào tỷ giá từ 4 API đối tác, đối soát dữ liệu và tính toán chỉ số thống kê. Mỗi chu kỳ mất khoảng 8 đến 10 giây. Phần lớn độ trễ đến từ network latency và các phép gom nhóm dữ liệu lặp lại.

Phương án tức thời lúc đó: dùng cache. Mình bọc ngay @functools.lru_cache vào các hàm xử lý. Kết quả ban đầu rất hứa hẹn. Response time giảm từ 8 giây xuống dưới 20ms đối với dữ liệu đã nằm trong RAM. Thế nhưng khi dữ liệu tăng dần, hệ thống bộc lộ ba vấn đề chí mạng:

  • Bay sạch dữ liệu khi restart: Mỗi lần deploy bản vá hoặc tiến trình khởi động lại, cache trong RAM mất trắng. Hệ thống lại phải gánh đợt tính toán nặng từ đầu.
  • Bị OOM Killer ‘trảm’: Tập dữ liệu tăng từ vài chục MB lên 400MB. VPS cạn sạch bộ nhớ. Linux OOM Killer lập tức hạ sát tiến trình Python mà không kịp ghi log cảnh báo.
  • Phân mảnh giữa các worker: Khi scale sang mô hình đa tiến trình với Gunicorn và Celery (4 worker), mỗi worker ôm một vùng RAM riêng. Worker 1 vừa fetch xong dữ liệu, Worker 2 vẫn cặm cụi gọi API lần nữa. CPU bị lãng phí đến 3-4 lần.

Vì sao cache in-memory thất bại trong kiến trúc thực tế?

Vấn đề không nằm ở code tồi. Nó nằm ở giới hạn kiến trúc của bộ nhớ trong:

  1. Process isolation: Bộ nhớ của tiến trình Python bị cô lập theo mặc định. Trừ khi bạn thiết lập cấu trúc multiprocessing.shared_memory phức tạp, các tiến trình không thể đọc cache của nhau.
  2. Thiếu tính bền vững (Persistence): RAM chỉ sống theo tiến trình. Những tác vụ batch chạy định kỳ 2 tiếng một lần bắt buộc phải tải lại mọi thứ nếu cache chỉ lưu trên RAM.
  3. Chi phí RAM đắt hơn SSD: Ổ cứng NVMe hiện nay đọc tuần tự tới vài GB/s, giá rẻ hơn RAM hàng chục lần. Nhồi nhét hàng gigabyte dữ liệu tạm vào RAM trên máy chủ cấu hình thấp là nước đi rủi ro.

Ba lựa chọn phổ biến và cái giá phải trả

Trước khi tìm ra giải pháp tối ưu, hầu hết anh em backend đều thử qua ba cách này:

1. Tự dump ra file JSON hoặc Pickle

Cách làm này rất quen thuộc: băm tham số thành tên file, rồi dùng pickle.dump() lưu vào thư mục /tmp.

import pickle
import os

def get_cached_report(key):
    path = f"/tmp/cache_{key}.pkl"
    if os.path.exists(path):
        with open(path, "rb") as f:
            return pickle.load(f)
    # Tính toán lại nếu miss...

Rủi ro: Race condition. Hai worker cùng ghi một file vào cùng một mili-giây sẽ khiến file pkl bị corrupt ngay. Bạn cũng phải tự code cơ chế hết hạn (TTL) và tự quét dọn khi ổ đĩa chạm ngưỡng 100%.

2. Dựng Redis hoặc Memcached

Redis là chuẩn mực ngành cho caching. Tốc độ cao, đa tiến trình tốt, hỗ trợ TTL chuẩn chỉnh.

Cái giá phải trả: Chi phí vận hành. Với script nội bộ, tool cào dữ liệu nhỏ hoặc cronjob, việc cài Redis daemon, mở port mạng, lo bảo mật firewall và duy trì kết nối TCP là một gánh nặng không cần thiết. Bản thân Redis daemon cũng chiếm sẵn 40-70MB RAM tĩnh.

3. Tự viết cache wrapper trên SQLite

SQLite hỗ trợ transaction ACID rất tốt. Tuy nhiên, việc tự viết schema, xử lý serialize/deserialize object phức tạp và duy trì giải thuật LRU dọn dẹp dung lượng tốn nhiều công sức bảo trì.

Giải pháp tối ưu: DiskCache

DiskCache là thư viện Python thuần túy giải quyết trọn vẹn bài toán này. Bên dưới, DiskCache kết hợp SQLite ở chế độ Write-Ahead Logging (WAL) để quản lý metadata và ghi file trực tiếp lên đĩa cho các object lớn. Trên ổ NVMe hiện đại, DiskCache đạt từ 25.000 đến 40.000 lượt đọc/giây. Tốc độ này thậm chí nhanh hơn Redis kết nối qua localhost vì không tốn overhead của giao thức mạng TCP socket.

Khi xử lý pipeline bóc tách dữ liệu từ log trước lúc đẩy vào cache, bạn sẽ cần chuẩn hóa chuỗi bằng regex khá nhiều. Nếu cần kiểm tra nhanh các biểu thức chính quy phức tạp mà không muốn mở script test riêng, bạn có thể ghé công cụ regex tester tại toolcraft.app/vi/tools/developer/regex-tester để kiểm tra pattern trực tiếp trên trình duyệt.

Bước 1: Cài đặt

pip install diskcache

Bước 2: Sử dụng Cache cơ bản với Key-Value và TTL

Object Cache hoạt động quen thuộc như một dictionary, nhưng mọi thứ đều được đồng bộ xuống đĩa an toàn:

from diskcache import Cache
import time

# Khởi tạo cache tại thư mục mong muốn
cache = Cache('/tmp/my_disk_cache')

# Lưu key kèm thời gian sống (expire = 60 giây)
cache.set('user_101', {'name': 'Nguyen Van A', 'role': 'admin'}, expire=60)

# Đọc dữ liệu ra
user = cache.get('user_101')
print("Dữ liệu từ cache:", user)

# Kiểm tra sự tồn tại của key
if 'user_101' in cache:
    print("Key vẫn còn hiệu lực")

# Kiểm tra sau khi hết hạn
time.sleep(61)
print("Sau 60s:", cache.get('user_101'))  # Trả về None

Bước 3: Tối ưu hàm nặng với decorator @memoize

Thay vì viết code get/set thủ công, bạn chỉ cần gắn @cache.memoize. Thư viện sẽ tự băm tham số của hàm thành key:

from diskcache import Cache
import time

cache = Cache('./api_cache_dir')

@cache.memoize(expire=3600)  # Lưu kết quả trong 1 giờ
def fetch_partner_rates(date_str: str):
    print(f"[API CALL] Đang tải tỷ giá cho ngày {date_str}...")
    time.sleep(2.5)  # Giả lập network delay
    return {"USD": 25450, "EUR": 27300, "date": date_str}

# Lần 1: Hàm thực thi thật, tốn 2.5s
t0 = time.time()
print(fetch_partner_rates('2026-03-30'))
print(f"Lần 1 tốn: {time.time() - t0:.3f}s")

# Lần 2: Đọc thẳng từ ổ đĩa, tốn dưới 2ms
t0 = time.time()
print(fetch_partner_rates('2026-03-30'))
print(f"Lần 2 tốn: {time.time() - t0:.3f}s")

Bước 4: Đảm bảo an toàn khi chạy đa tiến trình (Process-safe)

Nhờ SQLite WAL mode và cơ chế file locking tự động ở cấp OS, nhiều tiến trình con có thể ghi đọc đồng thời mà không lo race condition:

from multiprocessing import Pool
from diskcache import Cache
import os

cache = Cache('./shared_cache')

def worker_task(item_id):
    # Các worker độc lập truy cập chung thư mục cache an toàn
    val = cache.get(item_id)
    if val is None:
        val = f"Computed_item_{item_id}_by_pid_{os.getpid()}"
        cache.set(item_id, val, expire=300)
    return val

if __name__ == '__main__':
    # 4 tiến trình con cùng chạy
    with Pool(processes=4) as pool:
        results = pool.map(worker_task, [10, 20, 30, 10, 20, 30])
        print("Kết quả:", results)

Bước 5: Khống chế dung lượng với size_limit (Eviction tự động)

Để ổ cứng không bao giờ bị đầy, bạn hãy đặt hạn mức bằng tham số size_limit (đơn vị byte). Khi vượt ngưỡng, DiskCache áp dụng giải thuật Least Recently Used (LRU) để dọn các bản ghi cũ nhất:

from diskcache import Cache

# Giới hạn cache tối đa 500MB (500 * 1024 * 1024 bytes)
cache = Cache('./bounded_cache', size_limit=524288000)

# DiskCache tự động giải phóng bản ghi cũ khi dung lượng chạm mốc 500MB

Khi nào nên dùng DiskCache, khi nào chọn Redis?

Chọn đúng công cụ dựa trên kiến trúc hạ tầng của bạn:

  • Chọn DiskCache khi: Ứng dụng chạy trên một node đơn, VPS ít RAM (1-2GB), công cụ dòng lệnh (CLI), app desktop, hoặc cronjob cần cache bền bỉ qua các lần restart mà không muốn gánh thêm một Redis server.
  • Chọn Redis khi: Hệ thống microservices phân tán gồm nhiều server cần chia sẻ chung một kho cache, hoặc bạn cần các tính năng nâng cao như Pub/Sub, Sorted Sets và Redis Streams.

Nếu hệ thống chưa cần scale nhiều node, DiskCache là cách tiếp cận thực dụng: cài đặt gọn gàng, chạy ổn định và tiết kiệm tối đa RAM cho máy chủ.

Share:
Tags: