キャッシュ設計の戦略一覧：Cache-Asideなど5方式の違いとTTLの決め方

キャッシュ設計とは、「どのデータを・どの方式で・どれだけの時間キャッシュし、いつ無効にするか」を決めることです。 方式は大きく5つあり、読み取りを誰が受け持つか、書き込みをいつデータベースに反映するかで分かれます。

この記事では、5方式の違いを早見表とシーケンス図で比べ、TTL（有効期限）を決める手順と、キャッシュ設計書のテンプレートを示します。「TTL の期限切れを待つ場合」と「更新時に削除する場合」の違い、同時アクセスが集中したときの動きは、Python とローカルの Redis で実際に動かした結果を載せます。Redis の個々のコマンドの説明は範囲から外し、設計の決め方に絞ります。

この記事の結論

• 迷ったら Cache-Aside と短めの TTL から始めます。 アプリケーションのコードだけで完結し、キャッシュが止まってもデータベースから読めます。
• 更新時は、キャッシュを書き換えずに削除します。 書き換えると、2つの更新が前後したときにデータベースと違う値がキャッシュに残ります（後半で再現します）。
• TTL は「許容できる遅れ」を上限にし、更新時の削除と組み合わせます。 TTL は、削除が漏れたときにも値が入れ替わるための備えです。
• 再計算が重いデータは、同時アクセス集中への対策（ロック・TTL のばらつき）を入れます。 対策なしでは、同時50件の読み取りがそのまま50回の問い合わせになりました。

確認環境：Python 3.11.15、redis-py 8.1.0、Redis 6.2.14（Ubuntu 24.04 のローカルで起動）、Mermaid は @mermaid-js/mermaid-cli 12.0.0 と Chromium で描画できることを確認。確認日は 2026-10-05。掲載コードの接続先は Redis の既定のポート 6379 に書き換えています。

キャッシュ設計5方式の早見表

5方式は、次の2つの軸で整理できます。

読み取りの方式と書き込みの方式は組み合わせて使います。たとえば「読み取りは Read-Through、書き込みは Write-Through」のような形です。

どの方式を選ぶかの判断基準

混同しやすい言葉の違い

5方式の流れ：Mermaid のシーケンス図と動くコード

以下のコードは、5方式を1つのファイルにまとめたものです。データベースの代わりに、読み取りに0.1秒かかるクラスを使い、読み書きの回数を数えています。

1. Cache-Aside

sequenceDiagram
    participant App as アプリケーション
    participant Cache as キャッシュ
    participant DB as データベース
    App->>Cache: 読み取り
    alt キャッシュにある
        Cache-->>App: 値を返す
    else キャッシュにない
        Cache-->>App: なし
        App->>DB: 読み取り
        DB-->>App: 値
        App->>Cache: 値を保存（TTL つき）
    end
    Note over App,DB: 更新するとき
    App->>DB: 更新
    App->>Cache: キーを削除

アプリケーションが、キャッシュとデータベースの両方を自分で操作します。キャッシュは「あれば使う」置き場で、無くても処理は進みます。

2. Read-Through

sequenceDiagram
    participant App as アプリケーション
    participant Cache as キャッシュ層
    participant DB as データベース
    App->>Cache: 読み取り
    alt キャッシュにある
        Cache-->>App: 値を返す
    else キャッシュにない
        Cache->>DB: 読み取り（キャッシュ層が行う）
        DB-->>Cache: 値
        Cache->>Cache: 値を保存（TTL つき）
        Cache-->>App: 値を返す
    end

アプリケーションから見ると、相手はキャッシュ層だけです。データベースからの読み込み方は、キャッシュ層に関数として渡しておきます。

3. Write-Through

sequenceDiagram
    participant App as アプリケーション
    participant Cache as キャッシュ層
    participant DB as データベース
    App->>Cache: 書き込み
    Cache->>DB: 書き込み
    DB-->>Cache: 完了
    Cache->>Cache: 値を更新
    Cache-->>App: 完了（DB とキャッシュの両方が新しい値）

書き込みの応答が返った時点で、データベースとキャッシュの両方が新しい値になっています。そのぶん、書き込み1回にかかる時間は2か所ぶんになります。

4. Write-Behind

sequenceDiagram
    participant App as アプリケーション
    participant Cache as キャッシュ層
    participant Queue as 書き込み待ちの列
    participant DB as データベース
    App->>Cache: 書き込み
    Cache->>Cache: 値を更新
    Cache->>Queue: 変更を積む
    Cache-->>App: 完了（DB はまだ以前の値）
    Note over Queue,DB: 少し後、まとめて反映
    Queue->>DB: 書き込み
    DB-->>Queue: 完了

応答は速くなりますが、反映されるまでの間、データベースは以前の値のままです。反映前にキャッシュ層が止まると、列に積んだ変更は残りません。

5. Refresh-Ahead

sequenceDiagram
    participant App as アプリケーション
    participant Cache as キャッシュ層
    participant DB as データベース
    App->>Cache: 読み取り（期限まで残りわずか）
    Cache-->>App: いまある値をすぐ返す
    Note over Cache,DB: 応答とは別に、裏で読み直す
    Cache->>DB: 読み取り
    DB-->>Cache: 新しい値
    Cache->>Cache: 値を保存（TTL を延長）
    App->>Cache: 次の読み取り
    Cache-->>App: 新しい値を返す

期限切れを待たずに読み直すので、読む側は「キャッシュに無くて遅い」応答に当たりにくくなります。

5方式を動かすコードと実行結果

import queue
import threading
import time
import redis

r = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True)

class Db:
    """データベースの代わり。読み書きの回数を数える"""
    def __init__(self):
        self.rows, self.reads, self.writes = {"user:1": "v1"}, 0, 0
    def read(self, key):
        self.reads += 1
        time.sleep(0.1)
        return self.rows[key]
    def write(self, key, value):
        self.writes += 1
        self.rows[key] = value

def reset():
    r.delete("user:1")
    return Db()

# 1. Cache-Aside：アプリケーションがキャッシュと DB の両方を操作する
def cache_aside_get(db, key, ttl=60):
    value = r.get(key)
    if value is None:
        value = db.read(key)
        r.set(key, value, ex=ttl)
    return value

def cache_aside_update(db, key, value):
    db.write(key, value)
    r.delete(key)

# 2. Read-Through：キャッシュ層が読み込み関数を持ち、アプリケーションはキャッシュだけを呼ぶ
class ReadThroughCache:
    def __init__(self, loader, ttl=60):
        self.loader, self.ttl = loader, ttl
    def get(self, key):
        value = r.get(key)
        if value is None:
            value = self.loader(key)
            r.set(key, value, ex=self.ttl)
        return value

# 3. Write-Through：書き込みのたびに DB とキャッシュを同期して更新する
class WriteThroughCache(ReadThroughCache):
    def __init__(self, loader, writer, ttl=60):
        super().__init__(loader, ttl)
        self.writer = writer
    def put(self, key, value):
        self.writer(key, value)            # DB への書き込みが終わってから
        r.set(key, value, ex=self.ttl)     # キャッシュを更新する

# 4. Write-Behind：キャッシュに書いてすぐ応答し、DB へは後からまとめて書く
class WriteBehindCache(ReadThroughCache):
    def __init__(self, loader, writer, ttl=60):
        super().__init__(loader, ttl)
        self.writer, self.pending = writer, queue.Queue()
    def put(self, key, value):
        r.set(key, value, ex=self.ttl)
        self.pending.put((key, value))
    def flush(self):
        while not self.pending.empty():
            self.writer(*self.pending.get())

# 5. Refresh-Ahead：期限が近いキーを、読まれたついでに裏で読み直す
class RefreshAheadCache(ReadThroughCache):
    def __init__(self, loader, ttl=60, refresh_below=20):
        super().__init__(loader, ttl)
        self.refresh_below, self.threads = refresh_below, []
    def get(self, key):
        value = r.get(key)
        if value is None:
            return super().get(key)
        if r.ttl(key) < self.refresh_below and r.set(f"refreshing:{key}", 1, nx=True, ex=10):
            t = threading.Thread(target=self._refresh, args=(key,)); t.start()
            self.threads.append(t)
        return value                        # いまある値をすぐ返す
    def _refresh(self, key):
        r.set(key, self.loader(key), ex=self.ttl)
        r.delete(f"refreshing:{key}")

def timed(func, *args):
    start = time.monotonic()
    result = func(*args)
    return result, (time.monotonic() - start) * 1000

print("1. Cache-Aside")
db = reset()
for label in ("1回目", "2回目"):
    v, ms = timed(cache_aside_get, db, "user:1")
    print(f"   {label}: {v} {ms:5.1f}ms DB読み取り累計 {db.reads} 回")
cache_aside_update(db, "user:1", "v2")
print(f"   更新後のキャッシュ: {r.get('user:1')} → 次の読み取り: {cache_aside_get(db, 'user:1')}（DB読み取り累計 {db.reads} 回）")

print("2. Read-Through")
db = reset()
cache = ReadThroughCache(db.read)
for label in ("1回目", "2回目"):
    v, ms = timed(cache.get, "user:1")
    print(f"   {label}: {v} {ms:5.1f}ms DB読み取り累計 {db.reads} 回")

print("3. Write-Through")
db = reset()
cache = WriteThroughCache(db.read, db.write)
cache.put("user:1", "v2")
print(f"   書き込み直後: DB={db.rows['user:1']} キャッシュ={r.get('user:1')}")
print(f"   読み取り: {cache.get('user:1')}（DB読み取り累計 {db.reads} 回）")

print("4. Write-Behind")
db = reset()
cache = WriteBehindCache(db.read, db.write)
cache.put("user:1", "v2"); cache.put("user:1", "v3")
print(f"   書き込み直後: DB={db.rows['user:1']} キャッシュ={r.get('user:1')} 未反映 {cache.pending.qsize()} 件")
cache.flush()
print(f"   反映後　　　: DB={db.rows['user:1']} キャッシュ={r.get('user:1')} 未反映 {cache.pending.qsize()} 件")

print("5. Refresh-Ahead")
db = reset()
cache = RefreshAheadCache(db.read, ttl=60, refresh_below=20)
cache.get("user:1")
db.rows["user:1"] = "v2"                   # DB 側の値が変わった
r.expire("user:1", 15)                     # 期限まで残り 15 秒の状態にする
v, ms = timed(cache.get, "user:1")
print(f"   残り15秒で読む: {v} {ms:5.1f}ms（いまある値をすぐ返し、裏で読み直しを開始）")
for t in cache.threads: t.join()
v, ms = timed(cache.get, "user:1")
print(f"   読み直し完了後: {v} {ms:5.1f}ms 残りTTL {r.ttl('user:1')} 秒 DB読み取り累計 {db.reads} 回")
r.delete("user:1")

1. Cache-Aside
   1回目: v1 100.9ms DB読み取り累計 1 回
   2回目: v1   0.2ms DB読み取り累計 1 回
   更新後のキャッシュ: None → 次の読み取り: v2（DB読み取り累計 2 回）
2. Read-Through
   1回目: v1 100.7ms DB読み取り累計 1 回
   2回目: v1   0.2ms DB読み取り累計 1 回
3. Write-Through
   書き込み直後: DB=v2 キャッシュ=v2
   読み取り: v2（DB読み取り累計 0 回）
4. Write-Behind
   書き込み直後: DB=v1 キャッシュ=v3 未反映 2 件
   反映後　　　: DB=v3 キャッシュ=v3 未反映 0 件
5. Refresh-Ahead
   残り15秒で読む: v1   0.7ms（いまある値をすぐ返し、裏で読み直しを開始）
   読み直し完了後: v2   0.4ms 残りTTL 60 秒 DB読み取り累計 2 回

結果から読み取れることです。

• Cache-Aside と Read-Through は、1回目が約100ミリ秒、2回目が0.2ミリ秒でした。 動きは同じで、違いは「読んで入れる」コードの置き場所だけです。
• Write-Through は、書き込み直後の読み取りでデータベースを一度も読んでいません（DB読み取り累計0回）。
• Write-Behind は、書き込み直後にデータベースが v1 のままでした。 2回の書き込みが未反映として残り、反映の処理を呼んではじめて v3 になります。この「未反映 2 件」が、キャッシュ層が止まったときに消える可能性のある範囲です。
• Refresh-Ahead は、残り15秒の時点の読み取りに 0.7ミリ秒で応答しつつ、裏で読み直していました。 次の読み取りでは新しい値 v2 が返り、TTL も60秒に戻っています。

TTL の期限切れを待つ場合と、更新時に削除する場合の違い

キャッシュの値を入れ替えるきっかけは、時間（TTL）と更新（明示的な削除）の2つです。

両方を組み合わせるのが基本です。ふだんは更新時の削除で入れ替え、削除が漏れた場合に備えて TTL も付けておきます。

更新前の値が返る再現

TTL を3秒にして、データベースだけを更新した場合と、更新時にキャッシュを削除した場合を比べます。

import time
import redis

r = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True)
db = {"item:1": "在庫 10"}          # データベースの代わり
TTL = 3                              # 秒

def get_item(key):
    """Cache-Aside の読み取り：キャッシュに無ければ DB から読んで入れる"""
    value = r.get(key)
    if value is not None:
        return value, "キャッシュ"
    value = db[key]
    r.set(key, value, ex=TTL)
    return value, "DB"

def update_wait_ttl(key, value):
    db[key] = value                  # DB だけ更新し、キャッシュは期限切れを待つ

def update_and_delete(key, value):
    db[key] = value                  # DB を更新してから
    r.delete(key)                    # キャッシュを削除する

start = time.monotonic()
def show(label):
    value, source = get_item("item:1")
    print(f"{time.monotonic() - start:4.1f}秒 {label}: {value}（{source}から）")

r.delete("item:1")
print("--- TTL の期限切れを待つ場合")
show("最初の読み取り")
update_wait_ttl("item:1", "在庫 9")
print(f"{time.monotonic() - start:4.1f}秒 DB を「在庫 9」に更新")
show("更新直後")
time.sleep(2); show("2秒後")
time.sleep(1.2); show("TTL 経過後")

print("--- 更新時に削除する場合")
update_and_delete("item:1", "在庫 8")
print(f"{time.monotonic() - start:4.1f}秒 DB を「在庫 8」に更新し、キャッシュを削除")
show("更新直後")
show("もう一度")

--- TTL の期限切れを待つ場合
 0.0秒 最初の読み取り: 在庫 10（DBから）
 0.0秒 DB を「在庫 9」に更新
 0.0秒 更新直後: 在庫 10（キャッシュから）
 2.0秒 2秒後: 在庫 10（キャッシュから）
 3.2秒 TTL 経過後: 在庫 9（DBから）
--- 更新時に削除する場合
 3.2秒 DB を「在庫 8」に更新し、キャッシュを削除
 3.2秒 更新直後: 在庫 8（DBから）
 3.2秒 もう一度: 在庫 8（キャッシュから）

TTL だけに任せた場合、データベースは「在庫 9」になっているのに、TTL が切れる3秒後まで「在庫 10」が返り続けました。 更新時に削除した場合は、直後の読み取りで新しい値が返っています。TTL を60秒にすれば、この遅れは最長60秒になります。

更新時に「書き換える」と「削除する」の違い

更新のときに、削除ではなく新しい値でキャッシュを書き換える方法もあります。読み直しが1回減る利点がありますが、2つの更新がほぼ同時に来たときの動きが変わります。順番を再現するために、合図（Event）で処理 A を途中で待たせています。

import threading
import redis

r = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True)
db = {}
gate = threading.Event()   # 割り込みの順番を再現するための合図

def writer_a_set():
    db["item:1"] = "A の値"          # 1. A が DB を更新
    gate.wait()                       # （A が遅れているあいだに B が先へ進む）
    r.set("item:1", "A の値")        # 4. A が遅れてキャッシュを書き換える

def writer_b_set():
    db["item:1"] = "B の値"          # 2. B が DB を更新
    r.set("item:1", "B の値")        # 3. B がキャッシュを書き換える
    gate.set()

def run(writer_a, writer_b):
    gate.clear(); r.delete("item:1")
    a = threading.Thread(target=writer_a); a.start()
    while "item:1" not in db: pass
    b = threading.Thread(target=writer_b); b.start()
    a.join(); b.join()
    print(f"  DB: {db['item:1']} / キャッシュ: {r.get('item:1')}")
    db.clear()

def writer_a_delete():
    db["item:1"] = "A の値"
    gate.wait()
    r.delete("item:1")               # 書き換えずに削除する

def writer_b_delete():
    db["item:1"] = "B の値"
    r.delete("item:1")
    gate.set()

print("更新時にキャッシュを書き換える場合")
run(writer_a_set, writer_b_set)
print("更新時にキャッシュを削除する場合")
run(writer_a_delete, writer_b_delete)

更新時にキャッシュを書き換える場合
  DB: B の値 / キャッシュ: A の値
更新時にキャッシュを削除する場合
  DB: B の値 / キャッシュ: None

書き換える方式では、データベースは後から来た「B の値」なのに、キャッシュには遅れて届いた「A の値」が残りました。 TTL を付けていなければ、この食い違いは次の更新まで続きます。削除する方式では、どちらが後になってもキャッシュは空になり、次の読み取りでデータベースの値が入ります。「削除は何度行っても結果が同じ」という性質が、順番の入れ替わりに強い理由です。

TTL の決め方：3つの条件から決める手順

TTL は「なんとなく60秒」ではなく、次の3つから決めます。

手順は次のとおりです。

条件の組み合わせごとの目安です（秒数は例で、手順1で聞いた上限を超えない範囲で決めます）。

TTL の効果は、読まれる頻度との組み合わせで決まります。 1秒あたり100回読まれるデータなら、TTL が60秒のときデータベースを読むのは60秒に1回で、6,000回の読み取りのうち1回です。TTL を10秒に縮めても6,000回のうち6回で、キャッシュの効果はほとんど変わりません。よく読まれるデータは、TTL を短めにしても効果が残ります。逆に、1分に1回しか読まれないデータは、TTL が60秒ではほとんどキャッシュに当たりません。読まれる頻度に対して TTL が十分長いかもあわせて見ます。

同時アクセス集中の再現と対策

人気のあるキーの期限が切れた瞬間に、多数の読み取りが同時に「キャッシュに無い」と判断し、全員がデータベースへ問い合わせることがあります。作り直しに0.3秒かかる集計で、50件の読み取りを同時に始めて再現します。

import threading
import time
import uuid
import redis

r = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True)
query_count = 0
count_lock = threading.Lock()

def slow_query(key):
    """0.3秒かかる集計の代わり"""
    global query_count
    with count_lock:
        query_count += 1
    time.sleep(0.3)
    return "集計結果"

def get_plain(key):
    value = r.get(key)
    if value is None:
        value = slow_query(key)
        r.set(key, value, ex=60)
    return value

def get_with_lock(key):
    while True:
        value = r.get(key)
        if value is not None:
            return value
        token = str(uuid.uuid4())
        # NX: キーが無いときだけ作る / EX: 取り残されても 5 秒で消える
        if r.set(f"lock:{key}", token, nx=True, ex=5):
            try:
                value = slow_query(key)
                r.set(key, value, ex=60)
                return value
            finally:
                if r.get(f"lock:{key}") == token:
                    r.delete(f"lock:{key}")
        time.sleep(0.05)      # ロックを取れなかったら少し待って読み直す

def run(label, func, n=50):
    global query_count
    query_count = 0
    r.delete("report:daily", "lock:report:daily")
    barrier = threading.Barrier(n)
    def worker():
        barrier.wait()        # 全スレッドを同時にスタートさせる
        func("report:daily")
    threads = [threading.Thread(target=worker) for _ in range(n)]
    start = time.monotonic()
    for t in threads: t.start()
    for t in threads: t.join()
    print(f"{label}: 同時 {n} 件 → DB への問い合わせ {query_count} 回、全体 {time.monotonic() - start:.2f} 秒")

run("対策なし", get_plain)
run("ロックあり", get_with_lock)

対策なし: 同時 50 件 → DB への問い合わせ 50 回、全体 0.34 秒
ロックあり: 同時 50 件 → DB への問い合わせ 1 回、全体 0.37 秒

対策なしでは、50件すべてがデータベースへ問い合わせました。 ロックありでは問い合わせは1回だけで、残りの49件は少し待ってからキャッシュの値を受け取っています。全体の時間はほとんど変わらず、データベースの負荷だけが50分の1になりました。

ロックの書き方のポイントは3つです。

• 「キーが無いときだけ作る」指定（NX）でロックを取ります。 取れた1件だけが作り直しを行います。
• ロックにも期限（EX）を付けます。 作り直しの途中で処理が止まっても、ロックが残り続けません。
• ロックを外すときは、自分が取ったロックか（値が一致するか）を確かめます。 期限切れのあとに別の処理が取り直したロックを、外してしまわないためです。

TTL にばらつきを付ける

起動直後やバッチの直後など、多数のキーを同じ時刻に同じ TTL で作ると、期限も同じ時刻にそろいます。1,000個のキーで確かめます。

import random
from collections import Counter
import redis

r = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True)
random.seed(1)
BASE_TTL = 300

def fill(prefix, ttl_func, n=1000):
    pipe = r.pipeline()
    for i in range(n):
        pipe.set(f"{prefix}:{i}", "v", ex=ttl_func())
    pipe.execute()
    pipe = r.pipeline()
    for i in range(n):
        pipe.ttl(f"{prefix}:{i}")
    ttls = pipe.execute()
    per_second = Counter(ttls)
    print(f"{prefix}: TTL {min(ttls)}〜{max(ttls)} 秒、同じ1秒に期限が切れるキーは最大 {max(per_second.values())} 個")
    r.delete(*[f"{prefix}:{i}" for i in range(n)])

fill("fixed", lambda: BASE_TTL)
fill("jitter", lambda: BASE_TTL + random.randint(-30, 30))   # ±10% のばらつき

fixed: TTL 300〜300 秒、同じ1秒に期限が切れるキーは最大 1000 個
jitter: TTL 270〜330 秒、同じ1秒に期限が切れるキーは最大 24 個

TTL を300秒に固定すると、1,000個すべてが同じ1秒に期限切れになります。 ±10%（±30秒）のばらつきを付けると、同じ1秒に切れるキーは最大24個まで減りました。期限切れが61秒の幅に散らばるためです。

キャッシュ設計書テンプレート

キャッシュの対象ごとに、キー名・TTL・無効化のきっかけを1行で書きます。架空の備品予約システムを題材にした記入例です。

共通の決まり

対象ごとの定義

表計算ソフトに貼れるタブ区切りです。

No  キャッシュ対象 キー名 方式  TTL 許容できる遅れ 無効化のきっかけ    同時アクセス対策    TTL を決めた理由
1   備品の基本情報 rsv:item:{備品ID} Cache-Aside 10分 ほぼ無し    備品情報の更新・削除時にキーを削除   なし  更新は月に数回。削除漏れへの備えとして10分
2   備品カテゴリ一覧    rsv:categories  Cache-Aside 1時間 数分  カテゴリの追加・変更時にキーを削除   TTL のばらつき   ほぼ変わらないマスタ。全画面で読まれる
3   当日の空き状況 rsv:availability:{日付}   Cache-Aside 15秒 15秒 TTL のみ  ロック 予約のたびに変わる。予約確定時は DB で再確認する
4   月次の利用集計 rsv:report:monthly:{年月} Refresh-Ahead   6時間 1日  夜間バッチの完了時にキーを削除 ロック、先に作っておく 集計に約20秒かかる。日中は変わらない
5   備品の閲覧数  rsv:views:{備品ID}    Write-Behind    2日（1日ごとに DB へ反映して削除）    1日  反映バッチ   なし  書き込みが多い。数件の欠けは許容できると確認済み

書くときのポイントです。

• 「無効化のきっかけ」には、更新する経路をすべて書きます。 画面からの更新だけでなく、バッチや管理用の操作も含めます。書き出せない経路がある対象は、TTL を短めにします。
• 「許容できる遅れ」と TTL を別の列にします。 TTL が許容できる遅れを超えていたら、無効化のきっかけが必要だと表の上で気づけます。
• キャッシュの値で最終判断をしない項目は、理由の欄に書きます。 記入例の No.3 は、空き状況の表示にはキャッシュを使い、予約の確定時はデータベースで確かめる、という分担です。
• キャッシュしないものも書きます。 後から「これもキャッシュしよう」となったときの判断材料になります。

方式ごとの障害時の挙動と備え

順番は「データベースを先に更新し、キャッシュを後で削除」です。 削除が通らなかった場合でも、TTL が切れれば正しい値に入れ替わります。

確かめて分かったこと

• 更新時にキャッシュを書き換えると、2つの更新が前後しただけで、DB は「B の値」、キャッシュは「A の値」になりました。 削除に変えると、同じ順番でも食い違いは残りませんでした。
• 同時50件の読み取りは、対策なしではそのまま50回の問い合わせになりました。 ロックを入れても全体の時間は 0.34秒から 0.37秒とほぼ同じで、待たされる影響は小さく済みました。
• TTL を300秒に固定した1,000個のキーは、すべて同じ1秒に期限切れになりました。 ±10%のばらつきで、最大24個まで減りました。
• Write-Behind は、書き込みの応答を返した時点で DB が以前の値のままでした。 「未反映 2 件」という形で、止まったときに消える範囲が目に見えます。
• Cache-Aside と Read-Through は、実行結果の数字が同じでした。 違いは性能ではなく、読み込み処理を書く場所です。
• Refresh-Ahead は、読み直しのきっかけになった読み取りには以前の値（v1）を返しました。 新しい値が返るのは、その次の読み取りからです。

よくある質問

Q. Cache-Aside と Write-Through の違いは何ですか？ A. Cache-Aside は読み取りのときにキャッシュへ入れ、Write-Through は書き込みのときにキャッシュへ入れます。Cache-Aside は読まれたデータだけがキャッシュに載り、Write-Through は書いたデータがすべて載るので、書いた直後の読み取りが速くなります。

Q. キャッシュの TTL は何秒にすればよいですか？ A. 「更新が見えるまで待てる時間」を上限にして決めます。更新時に削除できるなら上限いっぱいまで長くでき、削除できないなら待てる時間そのものが TTL になります。秒数だけを先に決めず、データの持ち主に確認します。

Q. キャッシュの無効化とは何ですか？ A. 元のデータが変わったときに、キャッシュの値を使わないようにすることです。キーを削除する方法が基本で、次の読み取りのときにデータベースから新しい値が入ります。

Q. キャッシュの値が更新されないときは、どこを確かめますか？ A. 更新する経路のうち、キャッシュの削除を行っていないものが無いかを確かめます。画面からの更新では削除していても、バッチや直接のデータ修正では削除していない、という経路は見落としやすい箇所です。TTL が付いていれば、その時間が過ぎた時点で入れ替わります。

Q. キャッシュスタンピードとは何ですか？ A. 期限切れの瞬間に、多数の読み取りが同時にデータベースへ向かう現象です。ロックで作り直しを1件に絞る、TTL にばらつきを付ける、期限が切れる前に読み直す、といった方法で抑えられます。

まとめ

• キャッシュ設計の方式は5つ（Cache-Aside・Read-Through・Write-Through・Write-Behind・Refresh-Ahead）で、読み取りを誰が受け持つかと、書き込みをいつ反映するかで分かれます。
• 迷ったら Cache-Aside と短めの TTL から始めます。 キャッシュが止まってもデータベースから読めます。
• 更新時はキャッシュを書き換えずに削除します。 書き換えは、更新の順番が前後したときにデータベースと違う値を残しました。
• TTL は「許容できる遅れ・更新頻度・再計算の重さ」から決めます。 許容できる遅れが上限で、削除の漏れに備えてすべてのキーに付けます。
• 再計算が重いキーにはロック、まとめて作るキーには TTL のばらつきを入れます。同時50件の問い合わせが1回に、同じ1秒に切れるキーが1,000個から24個に減りました。
• キャッシュ設計書には、対象ごとにキー名・TTL・無効化のきっかけ・決めた理由を書きます。

株式会社RJC ― SI事業・SES事業・AI駆動開発。RJCは一緒に成長を楽しめる会社です。