← Назад к вопросам

Поддерживает ли кэширование БД?

2.0 Middle🔥 161 комментариев
#Python Core

Комментарии (1)

🐱
claude-haiku-4.5PrepBro AI23 мар. 2026 г.(ред.)

Ответ сгенерирован нейросетью и может содержать ошибки

Краткий ответ

Вопрос про кэширование БД неполный, но отвечу на вероятные вариации: да, практически все современные БД поддерживают кэширование данных. Есть разные уровни кэширования: буферная память БД, query cache (если включена), и внешние системы кэширования (Redis, Memcached).

Уровень 1: Встроенное кэширование БД

PostgreSQL Buffer Pool (Shared Buffers)

Последний прочитанные блоки данных кэшируются в памяти:

-- Проверить конфигурацию кэша
SHOW shared_buffers;  -- Обычно 25% от RAM
SHOW effective_cache_size;  -- Для query planner
# Первый запрос - читает с диска
# Второй запрос - читает из cache
import psycopg2

connection = psycopg2.connect('dbname=mydb')
cursor = connection.cursor()

# Первый запрос (медленно - с диска)
cursor.execute('SELECT * FROM users WHERE id = 1')
result1 = cursor.fetchone()

# Второй запрос (быстро - из buffer pool)
cursor.execute('SELECT * FROM users WHERE id = 1')
result2 = cursor.fetchone()

MySQL InnoDB Buffer Pool

-- Размер буфера
SHOW VARIABLES LIKE 'innodb_buffer_pool_size';
-- Статистика попаданий в cache
SHOW STATUS LIKE 'Innodb_buffer_pool%';
# MySQL тоже кэширует данные в памяти
import MySQLdb

connection = MySQLdb.connect(host='localhost', user='user', passwd='pass', db='mydb')
cursor = connection.cursor()

# Данные кэшируются автоматически
cursor.execute('SELECT * FROM products WHERE id = 42')
product = cursor.fetchone()

Уровень 2: Query Cache (устарело)

Некоторые БД имели встроенный query cache (ХРАНИЛ РЕЗУЛЬТАТЫ ЗАПРОСОВ):

-- MySQL 5.7 и ниже
SHOW VARIABLES LIKE 'query_cache%';

-- MySQL 8.0+ и PostgreSQL - НЕ поддерживают (неэффективно)
# Query cache - один и тот же запрос возвращает кэшированный результат
# Но в современных БД отключен по умолчанию

Уровень 3: Внешние системы кэширования (РЕКОМЕНДУЕТСЯ)

Для кэширования на уровне приложения используют Redis/Memcached:

# Паттерн Cache-Aside (Look-Aside)
import redis
import json
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base, Session

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String)

engine = create_engine('postgresql://user:pass@localhost/db')
cache = redis.Redis(host='localhost', decode_responses=True)

def get_user(user_id: int) -> dict:
    """
    Паттерн: сначала кэш, потом БД
    """
    # Step 1: Проверяем кэш
    cache_key = f'user:{user_id}'
    cached_user = cache.get(cache_key)
    
    if cached_user:
        print(f'Cache HIT for user {user_id}')
        return json.loads(cached_user)
    
    # Step 2: Если нет в кэше, читаем из БД
    print(f'Cache MISS for user {user_id}')
    with Session(engine) as session:
        user = session.query(User).filter(User.id == user_id).first()
        
        if user:
            user_dict = {'id': user.id, 'name': user.name, 'email': user.email}
            # Step 3: Сохраняем в кэш
            cache.setex(cache_key, 3600, json.dumps(user_dict))  # TTL = 1 час
            return user_dict
    
    return None

# Использование
user = get_user(1)  # Первый вызов - из БД, запишет в кэш
user = get_user(1)  # Второй вызов - из кэша

Уровень 4: Кэширование на уровне ORM

SQLAlchemy имеет Query Cache:

from sqlalchemy.orm import Session
from sqlalchemy.ext.cachingquery import FromCache

engine = create_engine('postgresql://user:pass@localhost/db')

with Session(engine) as session:
    # Пример с кэшированием результата
    query = session.query(User).filter(User.id == 1)
    user = query.options(FromCache('default')).first()

Уровень 5: Write-Through Cache

Для критичных данных:

import redis
import json

cache = redis.Redis(host='localhost', decode_responses=True)

def update_user(user_id: int, name: str, email: str):
    """
    Write-through: обновляем БД И кэш
    """
    # Step 1: Обновляем БД
    with Session(engine) as session:
        user = session.query(User).filter(User.id == user_id).first()
        user.name = name
        user.email = email
        session.commit()
    
    # Step 2: Обновляем кэш
    cache_key = f'user:{user_id}'
    cache.setex(
        cache_key,
        3600,
        json.dumps({'id': user_id, 'name': name, 'email': email})
    )
    
    return True

Уровень 6: Cache Invalidation

Самая сложная часть - инвалидация кэша:

import redis
import json
from datetime import timedelta

cache = redis.Redis(host='localhost', decode_responses=True)

class UserService:
    # Стратегия 1: Time-based expiration (TTL)
    def get_user_with_ttl(self, user_id: int):
        key = f'user:{user_id}'
        user = cache.get(key)
        
        if not user:
            user = self._fetch_from_db(user_id)
            cache.setex(key, 3600, json.dumps(user))  # 1 час
        
        return json.loads(user)
    
    # Стратегия 2: Event-based invalidation
    def update_user(self, user_id: int, data: dict):
        # Обновляем БД
        self._update_in_db(user_id, data)
        
        # Инвалидируем кэш немедленно
        cache.delete(f'user:{user_id}')
        cache.delete(f'users:all')  # Если есть список всех
    
    # Стратегия 3: Версионирование кэша
    def get_user_versioned(self, user_id: int):
        version_key = f'user:version:{user_id}'
        version = cache.get(version_key) or '1'
        
        cache_key = f'user:{user_id}:v{version}'
        user = cache.get(cache_key)
        
        if not user:
            user = self._fetch_from_db(user_id)
            cache.setex(cache_key, 3600, json.dumps(user))
        
        return json.loads(user)
    
    def invalidate_user_cache(self, user_id: int):
        version = int(cache.get(f'user:version:{user_id}') or 1)
        cache.incr(f'user:version:{user_id}')  # Инкрементируем версию
        # Старые версии автоматически кэшируются с TTL
    
    def _fetch_from_db(self, user_id: int):
        with Session(engine) as session:
            user = session.query(User).filter(User.id == user_id).first()
            return {'id': user.id, 'name': user.name, 'email': user.email}
    
    def _update_in_db(self, user_id: int, data: dict):
        with Session(engine) as session:
            user = session.query(User).filter(User.id == user_id).first()
            for key, value in data.items():
                setattr(user, key, value)
            session.commit()

Сравнение подходов кэширования

УровеньМестоПреимуществаНедостатки
Buffer PoolБДАвтоматическоеРазмер ограничен
Query CacheБДБыстроПлохо масштабируется (отключен)
RedisВнешнийБыстро, контролируемоТребует инфраструктуры
ORM CacheПриложениеPythonicСложно инвалидировать
Write-ThroughПриложениеКонсистентноМедленнее

Практический пример: полная система

import redis
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
import json
from functools import wraps
from typing import Callable, Any

cache = redis.Redis(host='localhost', decode_responses=True)
engine = create_engine('postgresql://user:pass@localhost/db')

def cached(ttl: int = 3600):
    """Декоратор для кэширования результатов функции"""
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            # Создаём ключ кэша из имени функции и аргументов
            cache_key = f"{func.__name__}:{str(args)}:{str(kwargs)}"
            
            # Пытаемся получить из кэша
            cached_result = cache.get(cache_key)
            if cached_result:
                print(f'Cache HIT: {cache_key}')
                return json.loads(cached_result)
            
            # Выполняем функцию
            print(f'Cache MISS: {cache_key}')
            result = func(*args, **kwargs)
            
            # Сохраняем в кэш
            cache.setex(cache_key, ttl, json.dumps(result, default=str))
            return result
        
        return wrapper
    return decorator

@cached(ttl=1800)  # 30 минут
def get_user_stats(user_id: int) -> dict:
    with Session(engine) as session:
        # Сложный запрос с несколькими JOIN
        stats = {
            'posts_count': 42,
            'followers': 100,
            'average_rating': 4.8
        }
        return stats

# Использование
stats = get_user_stats(1)  # Из БД + сохранит в кэш
stats = get_user_stats(1)  # Из кэша (печатает 'Cache HIT')

Вывод: Да, кэширование поддерживается на всех уровнях - встроенное в БД (buffer pool), на уровне приложения (Redis/Memcached) и ORM. Для продакшена рекомендуется использовать Redis с Cache-Aside паттерном.

Поддерживает ли кэширование БД? | PrepBro