Кеширование: паттерны, стратегии и реализации

О чём эта статья

Что такое кеширование

Кеш (cache) — это быстрое промежуточное хранилище для данных, к которым часто обращаются. Вместо того чтобы каждый раз идти в медленный источник (база данных, диск, внешний API), программа сначала проверяет кеш. Если данные там есть — возвращает мгновенно (это называется cache hit, попадание в кеш). Если нет (cache miss, промах) — загружает из источника и сохраняет в кеш для будущих запросов.

Простая аналогия: рабочий стол, на котором лежат часто используемые документы. Не надо каждый раз идти в шкаф с архивом.

Ключевые понятия

Существует знаменитая фраза Фила Карлтона: «В Computer Science есть только две по-настоящему сложные вещи: инвалидация кеша и придумывание названий». Инвалидация сложна потому, что нужно понять: какие именно записи затронуты изменением, когда их удалять, и как синхронизировать несколько кешей в распределённой системе.

Что вы узнаете

Кеширование — это не «добавим Redis, и всё ускорится». Это архитектурное решение с trade-off между свежестью данных, задержкой и сложностью. Неправильно настроенный кеш создаёт больше проблем, чем решает: устаревшие данные, stampede при прогреве, деградация при падении Redis.

В этой статье — полный инженерный разбор: от выбора паттерна до кода, который можно скопировать и запустить. Разбираем 5 паттернов кеширования с примерами на Python, работу с Redis, кеширование в Go, HTTP-заголовки и архитектурные решения.

Для кого: разработчики и архитекторы, которым нужно внедрить кеширование в проект и не наступить на типичные грабли.

Зачем кешировать

Три причины, подтверждённые цифрами:

Закон Паркинсона для кеша: данные заполняют весь доступный объём. Если вы выделили 1 ГБ под Redis — через неделю он будет занят. Планируйте политику вытеснения заранее.

Паттерны кеширования

Пять паттернов. У каждого — код на Python, плюсы, минусы и когда применять.

1. Cache-Aside (Lazy Loading)

Самый популярный паттерн. Приложение сначала спрашивает кеш. Если данных нет — идёт в БД, сохраняет в кеш и возвращает.

import redis
import json

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

def get_user(user_id: int) -> dict:
    """Cache-Aside: проверяем кеш, при промахе — грузим из БД."""
    cache_key = f"user:{user_id}"

    # 1. Пытаемся прочитать из кеша
    cached = r.get(cache_key)
    if cached:
        return json.loads(cached)

    # 2. Промах — идём в БД
    user = db_get_user(user_id)  # ваша функция чтения из БД
    if user is None:
        return None

    # 3. Сохраняем в кеш с TTL
    r.setex(cache_key, 300, json.dumps(user))  # 5 минут
    return user

def update_user(user_id: int, data: dict) -> None:
    """При обновлении — инвалидируем кеш."""
    db_update_user(user_id, data)
    r.delete(f"user:{user_id}")  # удаляем, а не обновляем

Плюсы: простой, кеш не знает о БД, легко внедрить в существующий код.

Минусы: первый запрос всегда медленный (cold start), инвалидация — вручную.

Когда применять: 90% случаев. Если сомневаетесь — берите Cache-Aside.

2. Read-Through

Кеш сам ходит в БД при промахе. Приложение вообще не знает о существовании базы — оно всегда читает из кеша.

class ReadThroughCache:
    """Кеш, который сам загружает данные из БД при промахе."""

    def __init__(self, redis_client, db_loader):
        self.redis = redis_client
        self.loader = db_loader  # функция, которая умеет читать из БД

    def get(self, key: str, ttl: int = 300) -> dict | None:
        cached = self.redis.get(key)
        if cached:
            return json.loads(cached)

        # Кеш сам идёт в БД
        data = self.loader(key)
        if data is None:
            return None

        self.redis.setex(key, ttl, json.dumps(data))
        return data


# Использование:
cache = ReadThroughCache(r, lambda key: db_get_user(int(key.split(":")[1])))
user = cache.get("user:42")

Плюсы: приложение не знает о БД, простая модель.

Минусы: нужен loader для каждого типа данных, холодный старт первого запроса.

Когда применять: когда у вас много операций чтения и вы хотите абстрагировать слой доступа к данным. Хорошо ложится на Repository-паттерн.

3. Write-Through

Синхронная запись: данные пишутся в кеш и в БД одновременно.

def create_user_write_through(data: dict) -> int:
    """Write-Through: пишем в БД и в кеш в одной транзакции."""
    user_id = db_insert_user(data)

    # Сразу кладём в кеш — следующий read получит свежие данные
    r.setex(f"user:{user_id}", 300, json.dumps({**data, "id": user_id}))

    return user_id

Плюсы: данные в кеше всегда актуальны, нет окна несогласованности.

Минусы: каждая запись — 2 операции, latency записи растёт (БД + Redis). Если Redis упал — запись не проходит (если не обрабатывать ошибку).

Когда применять: когда свежесть данных критична, а latency записи не является узким местом. Например, админка с редкими изменениями.

Важно: не путайте с двойной записью «сначала БД, потом кеш» — это всё ещё Cache-Aside с eager warm. Write-Through подразумевает, что именно кеш-слой управляет записью.

4. Write-Behind (Write-Back)

Асинхронная запись: данные пишутся в кеш мгновенно, а в БД — с задержкой, пачками.

import threading
import time
from collections import deque

class WriteBehindBuffer:
    """Буферизует записи и сбрасывает их в БД пачками."""

    def __init__(self, redis_client, db_writer, flush_interval: float = 1.0):
        self.redis = redis_client
        self.db_writer = db_writer  # функция записи в БД
        self.queue = deque()
        self._stop = threading.Event()
        self._thread = threading.Thread(target=self._flush_loop, daemon=True)
        self._thread.start()

    def write(self, key: str, data: dict, ttl: int = 300) -> None:
        """Пишем в кеш мгновенно, в БД — отложенно."""
        self.redis.setex(key, ttl, json.dumps(data))
        self.queue.append((key, data))

    def _flush_loop(self) -> None:
        """Поток, который периодически сбрасывает очередь в БД."""
        while not self._stop.is_set():
            time.sleep(1.0)
            batch = []
            while self.queue:
                batch.append(self.queue.popleft())
            if batch:
                self.db_writer(batch)  # пакетная вставка

    def stop(self) -> None:
        self._stop.set()
        self._thread.join(timeout=5)


# Использование:
buffer = WriteBehindBuffer(r, db_batch_insert_users)
buffer.write("user:42", {"name": "Иван"})  # мгновенно в кеш
# ... через 1 секунду данные уйдут в БД пачкой

Плюсы: минимальная latency записи, давление на БД резко снижено (batch insert).

Минусы: риск потери данных при падении кеша до flush, сложность, eventual consistency.

Когда применять: высоконагруженные системы, где latency записи критична (логи, счётчики, аналитика). Никогда для финансовых транзакций.

5. Refresh-Ahead

Кеш проактивно обновляет данные до истечения TTL. Если ключ востребован и до истечения осталось менее 20% TTL — запускается асинхронное обновление.

import time
import random

class RefreshAheadCache:
    """Обновляет часто используемые ключи до истечения TTL."""

    def __init__(self, redis_client, loader, ttl: int = 300, threshold: float = 0.2):
        self.redis = redis_client
        self.loader = loader
        self.ttl = ttl
        self.threshold = threshold  # доля TTL, при которой начинаем обновление

    def get(self, key: str) -> dict | None:
        ttl = self.redis.ttl(key)  # -1 = нет TTL, -2 = ключа нет

        if ttl == -2:  # ключа нет — холодный старт
            return self._load_and_cache(key)

        cached = self.redis.get(key)
        if cached is None:
            return None

        # Если осталось меньше threshold * ttl — обновляем фоном
        if 0 < ttl < self.ttl * self.threshold:
            # Probabilistic early recompute: обновляем с вероятностью, чтобы
            # избежать stampede при одинаковых TTL у тысяч ключей
            if random.random() < 0.3:  # 30% запросов инициируют обновление
                threading.Thread(target=self._load_and_cache, args=(key,), daemon=True).start()

        return json.loads(cached)

    def _load_and_cache(self, key: str) -> dict | None:
        data = self.loader(key)
        if data:
            # Добавляем jitter к TTL, чтобы ключи не истекали одновременно
            jitter = random.randint(0, int(self.ttl * 0.1))
            self.redis.setex(key, self.ttl + jitter, json.dumps(data))
        return data

Плюсы: нет холодных промахов для горячих ключей, снижается tail latency.

Минусы: сложность, лишние чтения из БД для редко используемых ключей.

Когда применять: данные, которые запрашиваются постоянно и дорого считаются (например, конфигурация, рекомендации).

Таблица сравнения

Стратегии инвалидации

Инвалидация — самая сложная часть кеширования. Фил Карлтон сказал: «Есть только две сложные проблемы в программировании: инвалидация кеша и именование переменных». Разбираемся с первой.

TTL (Time-To-Live)

Самый простой механизм. Вы задаёте время жизни ключа — и Redis удаляет его автоматически.

# Установка TTL при записи
r.setex("config:app", 3600, json.dumps(config))  # 1 час

# Проверить оставшееся время
remaining = r.ttl("config:app")  # 3547 (секунд)

Как выбрать TTL:

• Данные меняются раз в день? TTL = 1 час.
• Данные меняются раз в час? TTL = 5 минут.
• Данные почти статичны? TTL = 24 часа.
• Никогда не ставьте TTL = 0 (вечное хранение) — кеш заполнится, и Redis начнёт вытеснять ключи непредсказуемо.

Джиттер (jitter): добавляйте случайное смещение ±10% к TTL. Если 10 000 ключей истекают одновременно — вы получите stampede. Jitter размазывает истечения по времени.

import random

base_ttl = 300
jitter = random.randint(0, int(base_ttl * 0.1))
r.setex(key, base_ttl + jitter, value)

LRU и LFU — политики вытеснения в Redis

Когда память Redis заполнена, он удаляет ключи по одной из политик:

Настройка в redis.conf:

maxmemory 1gb
maxmemory-policy allkeys-lru

LRU vs LFU на практике:

• LRU (Least Recently Used): удаляет то, что давно не читали. Хорошо для новостной ленты — старые новости перестают читать.
• LFU (Least Frequently Used): удаляет то, что редко читали. Хорошо для справочника — редко запрашиваемые страны должны вытесняться, даже если их недавно кто-то случайно запросил.

Write-invalidate vs write-update

При обновлении данных у вас есть два варианта:

# Вариант 1: Write-invalidate (рекомендуется)
def update_user(user_id: int, data: dict) -> None:
    db_update_user(user_id, data)
    r.delete(f"user:{user_id}")  # удаляем — следующий read загрузит свежее

# Вариант 2: Write-update (рискованно)
def update_user(user_id: int, data: dict) -> None:
    db_update_user(user_id, data)
    r.setex(f"user:{user_id}", 300, json.dumps(data))  # обновляем кеш

Почему write-invalidate лучше:

1. Если между delete и следующим read данные меняются ещё раз — вы теряете одно обновление, но оно подтянется при следующем чтении.
2. При write-update: если запись в кеш падает с ошибкой — кеш содержит устаревшие данные, и вы об этом не знаете.
3. Write-invalidate проще: меньше кода, меньше race conditions.

Исключение: Write-Through и Write-Behind паттерны требуют write-update по определению.

Cache Stampede

Проблема: 1000 клиентов одновременно запрашивают ключ, у которого только что истёк TTL. Все 1000 идут в БД. БД падает.

Решение 1 — Locking (блокировка):

import hashlib

def get_with_lock(key: str, loader, ttl: int = 300) -> dict | None:
    """Только первый запрос идёт в БД, остальные ждут на блокировке."""
    cached = r.get(key)
    if cached:
        return json.loads(cached)

    lock_key = f"lock:{key}"
    # Пытаемся захватить блокировку (NX = только если не существует)
    if r.set(lock_key, "1", nx=True, ex=10):  # блокировка на 10 секунд
        try:
            # Перепроверяем — другой процесс мог уже заполнить кеш
            cached = r.get(key)
            if cached:
                return json.loads(cached)
            data = loader(key)
            if data:
                r.setex(key, ttl, json.dumps(data))
            return data
        finally:
            r.delete(lock_key)
    else:
        # Ждём и перечитываем
        for _ in range(10):
            time.sleep(0.1)
            cached = r.get(key)
            if cached:
                return json.loads(cached)
        # Таймаут — идём в БД сами
        return loader(key)

Решение 2 — Probabilistic early recompute (вероятностное обновление):

Показано выше в Refresh-Ahead: когда TTL ниже порога, с вероятностью 30% запускаем фоновое обновление. 1000 запросов → ~300 попыток обновления → только 1 выигрывает блокировку → остальные 299 видят, что ключ уже обновлён.

Thundering Herd

Похоже на stampede, но причина другая: не истечение TTL, а массовый перезапуск/деплой. Все инстансы приложения стартуют одновременно и все ломятся в БД за одними и теми же данными.

Решение: при старте приложения добавляйте случайную задержку перед прогревом кеша:

import random

def warm_cache_on_startup():
    """Прогреваем кеш с рандомизированной задержкой."""
    delay = random.randint(0, 30)  # от 0 до 30 секунд
    time.sleep(delay)

    for user_id in get_active_user_ids():
        get_user(user_id)  # Cache-Aside сам заполнит кеш

Redis: быстрый старт

5 типов данных

import redis
r = redis.Redis(host='localhost', port=6379, decode_responses=True)

# 1. String — самый частый (ключ → значение)
r.set("visits", 1)
r.incr("visits")          # атомарный инкремент → 2
r.get("visits")           # "2"

# 2. Hash — объект с полями (экономит память на маленьких объектах)
r.hset("user:42", mapping={"name": "Иван", "email": "ivan@example.com"})
r.hget("user:42", "name")  # "Иван"
r.hgetall("user:42")       # {"name": "Иван", "email": "ivan@example.com"}

# 3. List — очередь/стек
r.lpush("queue:tasks", "task1", "task2")   # слева
r.brpop("queue:tasks", timeout=5)           # блокирующее чтение справа → "task1"

# 4. Set — уникальные значения (тэги, подписчики)
r.sadd("tags:article:1", "python", "redis", "cache")
r.smembers("tags:article:1")  # {"python", "redis", "cache"}
r.sinter("tags:article:1", "tags:article:2")  # пересечение множеств

# 5. Sorted Set — рейтинг (топ N)
r.zadd("leaderboard", {"player:1": 100, "player:2": 250})
r.zrevrange("leaderboard", 0, 2, withscores=True)  # топ-3 → [("player:2", 250.0), ...]

Правило выбора типа:

• Простое значение (сериализованный JSON, число, строка) → String
• Объект, поля которого вы читаете по отдельности → Hash (экономия памяти)
• Очередь задач → List
• Уникальные идентификаторы, связи → Set
• Рейтинг, топ, выборка по score → Sorted Set

Pipelines — batch-операции

Каждый вызов r.get() — это round-trip по сети. Pipeline собирает команды в пакет и отправляет за один round-trip:

# Медленно: 100 round-trips
for i in range(100):
    r.get(f"key:{i}")

# Быстро: 1 round-trip
pipe = r.pipeline()
for i in range(100):
    pipe.get(f"key:{i}")
results = pipe.execute()  # список из 100 значений

Разница в цифрах: 100 одиночных GET по localhost ≈ 25 мс. 100 GET через pipeline ≈ 0.5 мс. В 50 раз быстрее.

Pub/Sub для инвалидации

Когда данные меняются в одном сервисе, другие сервисы должны узнать об этом. Pub/Sub — простой способ:

# Сервис A: публикует инвалидацию
def update_user(user_id: int, data: dict) -> None:
    db_update_user(user_id, data)
    r.publish("cache:invalidate", f"user:{user_id}")

# Сервис B: подписывается на инвалидацию
pubsub = r.pubsub()
pubsub.subscribe("cache:invalidate")

for message in pubsub.listen():
    if message['type'] == 'message':
        r.delete(message['data'])  # инвалидируем локальный кеш

Ограничение: Pub/Sub не гарантирует доставку. Если подписчик был отключен — сообщение потеряно. Для гарантированной доставки используйте Redis Streams.

Lua-скрипты: атомарный check-and-set

Когда нужно атомарно проверить значение и изменить его — Lua-скрипт выполняется на стороне Redis без гонок:

# Атомарный check-and-set: обновить только если значение совпадает с ожидаемым
CHECK_AND_SET = """
local current = redis.call('GET', KEYS[1])
if current == ARGV[1] then
    redis.call('SET', KEYS[1], ARGV[2], 'EX', ARGV[3])
    return 1
else
    return 0
end
"""

# Компилируем скрипт один раз
cas_script = r.register_script(CHECK_AND_SET)

# Используем
result = cas_script(keys=["user:42"], args=["old_value", "new_value", 300])
if result == 1:
    print("Обновлено")
else:
    print("Данные изменились — повторяем логику")

Когда нужны Lua-скрипты: атомарный инкремент с проверкой лимита, условное удаление, перенос данных между структурами.

Redis Cluster: шардирование и consistent hashing

Один Redis-инстанс ограничен памятью машины (~256 GB на мощном сервере) и CPU одного ядра. Когда данных больше — данные разбиваются по шардам через Redis Cluster.

Как работает:

• 16 384 хеш-слота делятся между master-нодами (по умолчанию 3+ мастера с репликами)
• Каждый ключ попадает на ноду по формуле CRC16(key) mod 16384
• При добавлении ноды часть слотов мигрирует между нодами (resharding)

Хеш-теги для гарантии совместного хранения:

# Без тегов: user:42 и user:42:profile могут попасть на разные ноды
cache.set('user:42', {...})              # нода A
cache.set('user:42:profile', {...})      # нода B (другой слот!)

# С тегами {user:42} — оба ключа на одной ноде
cache.set('{user:42}:main', {...})        # нода A
cache.set('{user:42}:profile', {...})     # нода A (тот же слот)

Используйте хеш-теги, когда нужны атомарные операции над группой ключей (multi-key транзакции, MGET, Lua-скрипты с несколькими ключами).

Hot key — главная проблема шардирования:

Если 1% ключей принимает 80% трафика (например, кеш популярного товара или сессия известного пользователя), одна нода становится узким местом. Решения:

• Репликация hot key на несколько нод с разными именами (key:1, key:2), чтение случайным
• Клиентский шардинг — приложение само знает о hot keys и распределяет нагрузку
• Локальный in-process кеш поверх Redis (двухуровневый кеш) для сверхпопулярных данных

# Двухуровневый кеш: L1 = functools.lru_cache (in-process), L2 = Redis
import functools
import redis

redis_client = redis.Redis()

@functools.lru_cache(maxsize=1000)
def get_hot_product(product_id: int):
    """L1: 1000 самых горячих товаров в памяти процесса"""
    cached = redis_client.get(f'product:{product_id}')
    if cached:
        return cached
    data = fetch_from_db(product_id)
    redis_client.setex(f'product:{product_id}', 300, data)
    return data

Memcached vs Redis: Memcached изначально многопоточный и лучше работает с pure-cache нагрузкой (короткие TTL, без персистентности). Redis — single-threaded, но поддерживает персистентность, Lua, структуры данных, паб-саб. Для нового проекта обычно выбирают Redis; Memcached — legacy или специализированные сценарии с pure-cache нагрузкой.

Python: кеширование на практике

@functools.lru_cache — бенчмарк

Встроенный декоратор для кеширования результатов функций в памяти процесса:

import functools
import time

# Без кеша
def fibonacci_no_cache(n: int) -> int:
    if n < 2:
        return n
    return fibonacci_no_cache(n - 1) + fibonacci_no_cache(n - 2)

# С кешем
@functools.lru_cache(maxsize=128)
def fibonacci_cached(n: int) -> int:
    if n < 2:
        return n
    return fibonacci_cached(n - 1) + fibonacci_cached(n - 2)

# Бенчмарк
start = time.perf_counter()
fibonacci_no_cache(35)
no_cache_time = time.perf_counter() - start

start = time.perf_counter()
fibonacci_cached(35)
cached_time = time.perf_counter() - start

print(f"Без кеша: {no_cache_time:.2f}с")
print(f"С lru_cache: {cached_time:.6f}с")
print(f"Ускорение: {no_cache_time / cached_time:.0f}x")

# Вывод (типичный):
# Без кеша: 2.15с
# С lru_cache: 0.000032с
# Ускорение: 67187x

Важно: lru_cache хранит данные в оперативной памяти процесса. При перезапуске приложения кеш очищается. Не подходит для распределённых систем — каждый инстанс хранит свой кеш.

Django Cache Framework

Django предоставляет единый интерфейс для разных бекендов кеширования:

# settings.py — настройка Redis как бекенда
CACHES = {
    "default": {
        "BACKEND": "django.core.cache.backends.redis.RedisCache",
        "LOCATION": "redis://127.0.0.1:6379/1",
        "TIMEOUT": 300,  # TTL по умолчанию (5 минут)
        "OPTIONS": {
            "parser_class": "redis.rust.RedisParser",  # быстрее
        },
    }
}

# views.py — кеширование целой view
from django.views.decorators.cache import cache_page

@cache_page(60 * 15)  # 15 минут
def article_detail(request, slug: str):
    article = Article.objects.get(slug=slug)
    return render(request, "article.html", {"article": article})

# Ручное управление кешем
from django.core.cache import cache

def get_expensive_data(key: str):
    data = cache.get(key)
    if data is None:
        data = compute_expensive_result()
        cache.set(key, data, timeout=3600)
    return data

# Инвалидация
def update_article(article_id: int, data: dict) -> None:
    Article.objects.filter(id=article_id).update(**data)
    cache.delete(f"article:{article_id}")              # точечная
    cache.delete_pattern("articles:page:*")             # по шаблону (осторожно в Redis!)

redis-py — основные операции

import redis
from redis.exceptions import ConnectionError

r = redis.Redis(
    host='localhost',
    port=6379,
    db=0,
    decode_responses=True,
    socket_connect_timeout=5,
    socket_keepalive=True,          # держим соединение открытым
    retry_on_timeout=True,          # переподключаемся при таймауте
    health_check_interval=30,       # проверка здоровья соединения
)

# Базовые операции с обработкой ошибок
def safe_get(key: str) -> str | None:
    try:
        return r.get(key)
    except ConnectionError:
        # Redis недоступен — деградируем gracefully, идём в БД
        return None

cachetools — in-memory кеши

Библиотека cachetools предоставляет разные стратегии вытеснения для in-memory кешей:

from cachetools import TTLCache, LRUCache, cached

# TTL-кеш: автоматическое удаление по времени
ttl_cache = TTLCache(maxsize=1000, ttl=300)

# LRU-кеш: вытеснение давно неиспользуемых
lru_cache = LRUCache(maxsize=1000)

# Декоратор с TTLCache
@cached(cache=TTLCache(maxsize=128, ttl=60))
def get_config(key: str) -> dict:
    return load_config_from_db(key)

Go: кеширование (кратко)

go-redis

import (
    "context"
    "encoding/json"
    "time"
    "github.com/redis/go-redis/v9"
)

var ctx = context.Background()

func NewRedisClient() *redis.Client {
    return redis.NewClient(&redis.Options{
        Addr:     "localhost:6379",
        Password: "",
        DB:       0,
    })
}

func GetUser(rdb *redis.Client, userID string) (*User, error) {
    key := "user:" + userID

    // Cache-Aside
    cached, err := rdb.Get(ctx, key).Result()
    if err == nil {
        var user User
        json.Unmarshal([]byte(cached), &user)
        return &user, nil
    }

    // Промах — идём в БД
    user, err := dbGetUser(userID)
    if err != nil {
        return nil, err
    }

    data, _ := json.Marshal(user)
    rdb.Set(ctx, key, data, 5*time.Minute)
    return user, nil
}

singleflight — защита от stampede в стандартной библиотеке

Go предоставляет singleflight в пакете golang.org/x/sync/singleflight. Он гарантирует, что только один вызов функции выполняется одновременно для одного ключа:

import "golang.org/x/sync/singleflight"

var sf singleflight.Group

func GetUserSafe(rdb *redis.Client, userID string) (*User, error) {
    key := "user:" + userID

    cached, err := rdb.Get(ctx, key).Result()
    if err == nil {
        var user User
        json.Unmarshal([]byte(cached), &user)
        return &user, nil
    }

    // Только ОДИН горутин идёт в БД, остальные ждут результат
    result, err, _ := sf.Do(key, func() (interface{}, error) {
        user, err := dbGetUser(userID)
        if err != nil {
            return nil, err
        }
        data, _ := json.Marshal(user)
        rdb.Set(ctx, key, data, 5*time.Minute)
        return user, nil
    })

    if err != nil {
        return nil, err
    }
    return result.(*User), nil
}

Почему singleflight важен: 1000 одновременных запросов к одному ключу с истёкшим TTL → только 1 идёт в БД. Это простейшая защита от stampede без блокировок в Redis.

bigcache — in-memory кеш для Go

import "github.com/allegro/bigcache/v3"

func NewBigCache() *bigcache.BigCache {
    config := bigcache.DefaultConfig(10 * time.Minute) // TTL
    config.Shards = 1024                               // количество шардов
    config.MaxEntriesInWindow = 1000 * 10 * 60         // оптимизация GC

    cache, _ := bigcache.NewBigCache(config)
    return cache
}

// Использование:
cache := NewBigCache()
cache.Set("key", []byte("value"))
data, _ := cache.Get("key")

Когда bigcache, а не Redis: однопроцессное приложение, не хотите поднимать внешнюю зависимость, нужна минимальная latency (память процесса быстрее, чем сеть до Redis).

HTTP-кеширование

HTTP-кеширование — это кеш между браузером и сервером. Работает на уровне протокола, не требует кода.

Cache-Control

Заголовок, который говорит браузеру и CDN, как кешировать ответ:

Cache-Control: public, max-age=3600, immutable

Практические правила:

• Картинки, шрифты, CSS/JS с хэшем в имени → max-age=31536000, immutable (год)
• HTML-страницы → max-age=0, must-revalidate (всегда проверять)
• API-ответы → max-age=60, private (минута — если данные не критсвежие)
• Личные данные → private, no-cache, no-store

ETag + If-None-Match

ETag — это хэш содержимого. Клиент отправляет обратно ETag в заголовке If-None-Match, а сервер отвечает 304 Not Modified без тела, если данные не изменились:

# Flask/FastAPI: валидация ETag
import hashlib
import json

def etag_from_data(data: dict) -> str:
    return hashlib.md5(json.dumps(data, sort_keys=True).encode()).hexdigest()

@app.get("/api/article/{slug}")
def get_article(slug: str, request: Request, response: Response):
    article = load_article(slug)
    etag = etag_from_data(article)

    if request.headers.get("If-None-Match") == etag:
        response.status_code = 304
        return b""  # тело не отправляем — клиент использует свой кеш

    response.headers["ETag"] = etag
    response.headers["Cache-Control"] = "public, max-age=60"
    return article

Экономия: для API, возвращающего JSON на 50 КБ, ответ 200 отдаёт 50 КБ, ответ 304 отдаёт ~200 байт (заголовки). Экономия: 250x по трафику, 10-100x по времени.

CDN-кеширование

CloudFlare, Fastly и другие CDN кешируют ответы на своих edge-серверах по всему миру:

# Nginx: настройка для CDN
location /static/ {
    expires 1y;
    add_header Cache-Control "public, immutable";
}

location /api/ {
    # CDN кеширует, но браузер перепроверяет
    add_header Cache-Control "public, max-age=60, must-revalidate";
}

# Варьировать кеш по языку
add_header Vary "Accept-Language";

Как это работает:

1. Пользователь из Токио запрашивает /api/articles.
2. Запрос уходит на edge-сервер CDN в Токио (5 мс), а не на origin-сервер в Европе (200 мс).
3. Если данные есть в кеше CDN — ответ за 5 мс.
4. Если нет — CDN идёт на origin, сохраняет и отдаёт клиенту.

Важно: CDN-кеш общий для всех пользователей. Не кешируйте через CDN персонализированные данные (Cache-Control: private).

Архитектурный уровень

Куда вставить кеш

Уровни, на которых можно применить кеширование:

Клиент (браузер/мобильное приложение)
   ↓ Cache-Control, ETag, Service Worker
CDN (CloudFlare/Fastly)
   ↓ edge-кеширование, 200+ точек присутствия
API Gateway (Nginx/Kong)
   ↓ кеширование ответов микросервисов
Приложение (in-memory)
   ↓ lru_cache, cachetools, Caffeine (Java)
Распределённый кеш (Redis/Memcached)
   ↓ общий кеш для всех инстансов приложения
База данных (PostgreSQL shared_buffers, MySQL buffer pool)
   ↓ кеш на уровне БД — не игнорируйте его

Правило: кешируйте на самом близком к клиенту уровне, где это возможно. Чем ближе к клиенту — тем больше latency вы отрезаете.

Cache warming при деплое

При деплое новой версии приложения старый кеш может стать невалидным (изменилась схема данных). Два подхода:

1. Холодный старт: просто удаляем кеш и позволяем Cache-Aside заполнить его. Минус: первые запросы медленные.
2. Прогрев: при деплое асинхронно загружаем горячие ключи:

async def warm_cache_on_deploy():
    """Прогреваем кеш после деплоя — асинхронно, не задерживая старт."""
    # Ждём, пока приложение начнёт отвечать
    await asyncio.sleep(5)

    hot_keys = await db_get_hot_keys()  # топ-1000 самых запрашиваемых сущностей
    for key in hot_keys:
        await cache_set(key, await db_get_by_key(key))

Мониторинг

Без мониторинга кеш — чёрный ящик. Отслеживайте:

Прометеус-экспортер для Redis:

docker run -d --name redis-exporter \
  -p 9121:9121 \
  oliver006/redis_exporter \
  --redis.addr redis://localhost:6379

Когда НЕ надо кешировать

Частые ошибки

Как быстро подключить: чек-лист

7 шагов для внедрения кеширования в существующий проект:

Шаг 1. Измерьте baseline

# Средняя latency самого частого запроса (без кеша)
ab -n 1000 -c 10 http://localhost:8080/api/articles/
# Запишите p50, p95, p99

Шаг 2. Выберите паттерн

Для 90% проектов — Cache-Aside. Код выше в разделе «Паттерны».

Шаг 3. Выберите TTL

Начните с 60 секунд. Поднимите до 300 (5 минут) если hit rate < 70%. Опустите до 30 секунд если пользователи жалуются на устаревшие данные.

Шаг 4. Подключите Redis

docker run -d --name redis-cache -p 6379:6379 \
  redis:7-alpine redis-server \
  --maxmemory 256mb \
  --maxmemory-policy allkeys-lru \
  --save ""  # отключаем RDB — это кеш, не БД

Шаг 5. Оберните в декоратор

import functools
import json
import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

def redis_cache(ttl: int = 60):
    """Декоратор Cache-Aside для произвольной функции."""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # Формируем ключ из имени функции и аргументов
            key_parts = [func.__name__] + [str(a) for a in args]
            key_parts += [f"{k}={v}" for k, v in sorted(kwargs.items())]
            cache_key = ":".join(key_parts)

            cached = r.get(cache_key)
            if cached:
                return json.loads(cached)

            result = func(*args, **kwargs)
            r.setex(cache_key, ttl, json.dumps(result))
            return result
        return wrapper
    return decorator

# Использование:
@redis_cache(ttl=300)
def get_article(slug: str) -> dict:
    return db_get_article(slug)  # тяжёлый запрос

Шаг 6. Добавьте инвалидацию

В код, который изменяет данные, добавьте удаление из кеша:

def update_article(slug: str, data: dict) -> None:
    db_update_article(slug, data)
    r.delete(f"get_article:{slug}")  # ключ должен совпадать с тем, что в декораторе

Шаг 7. Настройте мониторинг

# Быстрый взгляд на hit rate (обновлять раз в 5 секунд)
watch -n 5 'redis-cli INFO stats | grep keyspace'

Когда keyspace_hits / (keyspace_hits + keyspace_misses) > 0.8 — вы сделали всё правильно.

Ресурсы

• Redis official documentation — все типы данных, настройка, кластеризация
• AWS Caching Best Practices — отличный обзор паттернов от Amazon
• Scaling Memcache at Facebook — классическая статья про распределённый кеш
• golang.org/x/sync/singleflight — документация пакета singleflight

Что дальше

• Справочник архитектурных паттернов — 80+ паттернов, включая CQRS, Event Sourcing и другие паттерны, в которых кеш играет ключевую роль
• Шпаргалка по Python — от основ до продвинутых техник, включая декораторы и async/await
• Шпаргалка по Django — полный разбор Django ORM, views, middleware и cache framework
• Шпаргалка по FastAPI — REST API, Pydantic, фоновые задачи, WebSocket
• SQL для начинающих и продвинутых — индексы, оптимизация запросов, explain analyze

Связанные статьи: Микросервисы vs монолит, 97 принципов архитектора ПО, Архитектурные роли

• System Design: как проектировать масштабируемые системы — фреймворк проектирования с учётом кэширования
• Балансировщик нагрузки — L4 vs L7, Nginx, HAProxy