Posted inPython

PythonのDiskCache入門：Redis不要、マルチプロセス対応の安全なディスクキャッシュ活用法

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

実環境での課題：1GB RAMのVPSで直面したメモリ枯渇問題

数年前、わずか1GBのRAMしかない格安VPS上で財務レポートを処理するバッチジョブを担当したことがあります。スクリプトの役割は、連携先4社のAPIから為替レートをスクレイピングし、データの突合を行って統計指標を算出することでした。1サイクルあたりの所要時間は約8〜10秒。遅延の大部分はネットワークレイテンシと、繰り返されるデータの集計処理によるものでした。

その時すぐに思いついた解決策はキャッシュの導入でした。処理関数にすぐさま @functools.lru_cache を適用したところ、初期の結果は非常に良好でした。すでにRAM上にあるデータに対しては、レスポンスタイムが8秒から20ms以下へと劇的に短縮されたのです。しかし、データ量が増加するにつれて、システムは3つの致命的な問題に直面しました。

  • 再起動によるデータの全消失： パッチのデプロイやプロセスの再起動のたびに、RAM内のキャッシュが完全に失われます。その結果、システムは再び重い計算処理をゼロから実行せざるを得なくなりました。
  • OOM Killerによる強制終了： データセットが数十MBから400MBへと膨らむと、VPSのメモリが完全に枯渇しました。LinuxのOOM Killerが即座にPythonプロセスを強制終了（Kill）し、警告ログすら残らない状態に陥りました。
  • ワーカー間でのキャッシュの断片化： GunicornやCeleryによるマルチプロセス構成（4ワーカー）へスケールさせた際、各ワーカーが独立したRAM領域を保持していました。ワーカー1がデータを取得し終えたばかりでも、ワーカー2は律儀に再度APIを叩きにいきます。結果としてCPUリソースが3〜4倍も無駄遣いされていました。

なぜインメモリキャッシュは本番アーキテクチャで破綻するのか？

問題はコードの良し悪しではなく、インメモリというアーキテクチャそのものの制約にあります。

  1. プロセスの分離（Process Isolation）： Pythonプロセスのメモリはデフォルトで隔離されています。複雑な multiprocessing.shared_memory の仕組みを構築しない限り、各プロセスがお互いのキャッシュを直接読み取ることはできません。
  2. 永続性（Persistence）の欠如： RAMのデータはプロセスの生存期間に依存します。2時間ごとに定期実行されるバッチ処理などでは、キャッシュがRAM上にしかない場合、毎回すべてを再取得・再計算しなければなりません。
  3. SSDに比べて高価なRAMコスト： 現代のNVMe SSDは数GB/sのシーケンシャルリード速度を誇り、RAMの何十分の一ものコストで利用できます。スペックの限られたサーバーでギガバイト単位の一時データを無理やりRAMに詰め込むのは、極めてリスクの高いアプローチです。

よくある3つのアプローチと、その代償

最適な解決策にたどり着く前に、多くのバックエンドエンジニアが試みる3つのアプローチがあります。

1. JSONやPickleファイルへの自前ダンプ

非常によく見られる手法です。パラメータをハッシュ化してファイル名にし、pickle.dump() を使って /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)
    # キャッシュミスの場合は再計算...

リスク： レースコンディション（競合状態）が発生します。2つのワーカーが全く同じミリ秒単位のタイミングで同一ファイルに書き込もうとすると、pklファイルは即座に破損します。また、有効期限（TTL）の仕組みや、ディスク使用率が100%に達した際のクリーンアップ処理もすべて自前で実装しなければなりません。

2. RedisまたはMemcachedの導入

Redisはキャッシングにおけるデファクトスタンダードです。高速でマルチプロセス対応に優れ、TTLサポートも申し分ありません。

代償： 運用コストです。社内向けスクリプト、小規模なスクレイピングツール、cronジョブなどの用途において、Redisデーモンのインストール、ネットワークポートの開放、ファイアウォールの設定、TCP接続の維持管理は不必要なオーバーヘッドになります。また、Redisデーモン自体も起動しているだけで40〜70MBのRAMを常時消費します。

3. SQLiteを使った自作キャッシュラッパー

SQLiteはACIDトランザクションを強力にサポートしています。しかし、テーブルスキーマの設計、複雑なオブジェクトのシリアライズ／デシリアライズ処理、容量制限のためのLRUアルゴリズムの実装・維持には、多大なメンテナンス工数がかかります。

最適なソリューション：DiskCache

DiskCache は、この課題を完璧に解決するPure Pythonライブラリです。内部では、メタデータ管理にWrite-Ahead Logging（WAL）モードのSQLiteを使用し、大きなオブジェクトはディスクへ直接ファイルとして書き込みます。最新のNVMe SSD環境下では、DiskCacheは毎秒25,000〜40,000回の読み取りを達成します。TCPソケットによるネットワークプロトコルのオーバーヘッドがないため、場合によってはlocalhost経由のRedisよりも高速に動作します。

ログからデータを抽出してキャッシュに投入するパイプライン処理では、正規表現を用いた文字列の正規化が頻繁に必要となります。個別のテストスクリプトを立ち上げることなく複雑な正規表現を素早く検証したい場合は、ブラウザ上で直接パターンをテストできる toolcraft.app/ja/tools/developer/regex-tester のRegexテスターツールを活用してみてください。

ステップ1：インストール

pip install diskcache

ステップ2：Key-ValueとTTLを用いた基本的なCacheの使い方

Cache オブジェクトは通常の辞書（dict）のように直感的に扱えますが、すべてのデータはディスク上に安全に同期されます。

from diskcache import Cache
import time

# 指定したディレクトリにキャッシュを初期化
cache = Cache('/tmp/my_disk_cache')

# 有効期限付きでキーを保存（expire = 60秒）
cache.set('user_101', {'name': 'Nguyen Van A', 'role': 'admin'}, expire=60)

# データを取得
user = cache.get('user_101')
print("キャッシュからのデータ:", user)

# キーの存在確認
if 'user_101' in cache:
    print("キーは現在も有効です")

# 有効期限切れ後の確認
time.sleep(61)
print("60秒後:", cache.get('user_101'))  # Noneを返す

ステップ3：デコレータ @memoize による重い関数の最適化

手動でget/setのコードを書く代わりに、@cache.memoize を付与するだけです。ライブラリが関数の引数を自動的にハッシュ化してキーとして管理してくれます。

from diskcache import Cache
import time

cache = Cache('./api_cache_dir')

@cache.memoize(expire=3600)  # 結果を1時間キャッシュ
def fetch_partner_rates(date_str: str):
    print(f"[API CALL] {date_str} の為替レートを取得中...")
    time.sleep(2.5)  # ネットワーク遅延のシミュレーション
    return {"USD": 25450, "EUR": 27300, "date": date_str}

# 1回目：関数が実際に実行され、2.5秒かかる
t0 = time.time()
print(fetch_partner_rates('2026-03-30'))
print(f"1回目の所要時間: {time.time() - t0:.3f}s")

# 2回目：ディスクから直接読み込まれ、2ms未満で完了
t0 = time.time()
print(fetch_partner_rates('2026-03-30'))
print(f"2回目の所要時間: {time.time() - t0:.3f}s")

ステップ4：マルチプロセス実行時の安全性担保（プロセスセーフ）

SQLiteのWALモードとOSレベルでの自動ファイルロック機構により、複数の子プロセスがレースコンディションを心配することなく並行して読み書きを行えます。

from multiprocessing import Pool
from diskcache import Cache
import os

cache = Cache('./shared_cache')

def worker_task(item_id):
    # 独立したワーカーが同一のキャッシュディレクトリへ安全にアクセス
    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つの子プロセスを並行実行
    with Pool(processes=4) as pool:
        results = pool.map(worker_task, [10, 20, 30, 10, 20, 30])
        print("実行結果:", results)

ステップ5：size_limitによる容量制御（自動エビクション）

ディスクの枯渇を防ぐため、size_limit パラメータ（バイト単位）で上限を設定します。上限を超えると、DiskCacheはLRU（Least Recently Used）アルゴリズムを適用し、最も古くアクセスされていないレコードを自動的に削除します。

from diskcache import Cache

# キャッシュの最大容量を500MB（500 * 1024 * 1024バイト）に制限
cache = Cache('./bounded_cache', size_limit=524288000)

# 容量が500MBに達すると、DiskCacheが古いレコードを自動解放

DiskCacheとRedisの使い分け基準

インフラ構成に合わせて適切なツールを選定することが重要です。

  • DiskCacheを選ぶべきケース： 単一ノードで稼働するアプリケーション、低スペックVPS（RAM 1〜2GB）、CLIツール、デスクトップアプリ、あるいはプロセスの再起動を跨いでキャッシュを維持したいがRedisサーバーの運用負荷は避けたいcronジョブなど。
  • Redisを選ぶべきケース： 複数サーバーからなる分散マイクロサービス構成で共通のキャッシュストアを共有する必要がある場合や、Pub/Sub、Sorted Sets、Redis Streamsといった高度な機能を活用したい場合。

マルチノードへのスケールアウトがまだ不要なシステムであれば、DiskCacheはセットアップが容易で安定稼働し、サーバーのRAMリソースを最大限に節約できる極めて現実的で優れた選択肢となります。

Share:
Tags: