Как кэширование ускоряет код?

Как кэширование ускоряет код

Кэширование — это сохранение результатов дорогостоящих операций для их повторного использования. Это один из самых эффективных способов оптимизации производительности.

1. Принцип работы кэширования

Кэширование избегает повторных вычислений:

# БЕЗ кэширования — выполняется каждый раз
def fibonacci(n):
    if n <= 1:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

import time
start = time.time()
result = fibonacci(40)  # Очень медленно (~40 сек)
print(f"Время: {time.time() - start:.2f}s")  # 40+ секунд

# С кэшированием — вычисляется один раз
from functools import lru_cache

@lru_cache(maxsize=None)
def fibonacci_cached(n):
    if n <= 1:
        return n
    return fibonacci_cached(n-1) + fibonacci_cached(n-2)

start = time.time()
result = fibonacci_cached(40)  # Мгновенно
print(f"Время: {time.time() - start:.2f}s")  # <0.001 сек

Улучшение на миллионы раз! Вместо 10 миллиардов рекурсивных вызовов — только 41.

2. Виды кэширования

2.1 Функциональное кэширование (memoization)

Сохранение результатов функции по входным параметрам:

from functools import lru_cache

@lru_cache(maxsize=128)  # Максимум 128 результатов
def expensive_calculation(x, y):
    """Дорогостоящая операция"""
    import time
    time.sleep(1)  # Эмуляция медленного кода
    return x ** y

# Первый вызов — медленно (1 сек)
result1 = expensive_calculation(2, 10)  # 1024

# Второй вызов с теми же аргументами — мгновенно
result2 = expensive_calculation(2, 10)  # Из кэша

print(expensive_calculation.cache_info())
# CacheInfo(hits=1, misses=1, maxsize=128, currsize=1)

2.2 Кэширование запросов к БД

from functools import lru_cache
import time
from datetime import datetime, timedelta

class UserRepository:
    def __init__(self):
        self._cache = {}
        self._cache_time = {}
    
    def get_user_by_id(self, user_id: int):
        """Кэш с истечением"""
        now = datetime.now()
        
        # Проверить, есть ли в кэше и не истёк ли
        if user_id in self._cache:
            if now - self._cache_time[user_id] < timedelta(minutes=5):
                print(f"Из кэша: {user_id}")
                return self._cache[user_id]
        
        # Если кэша нет — запрос в БД
        print(f"Из БД: {user_id}")
        time.sleep(0.5)  # Эмуляция запроса
        user = {"id": user_id, "name": f"User {user_id}"}
        
        self._cache[user_id] = user
        self._cache_time[user_id] = now
        return user

repo = UserRepository()

# Первый вызов — из БД (500ms)
user1 = repo.get_user_by_id(1)

# Второй вызов — из кэша (мгновенно)
user2 = repo.get_user_by_id(1)

3. Кэширование HTTP запросов

import requests
from functools import lru_cache
import time

@lru_cache(maxsize=32)
def fetch_github_user(username: str):
    """Кэширование API ответов"""
    print(f"Запрос к API: {username}")
    time.sleep(1)  # Эмуляция сетевой задержки
    
    response = requests.get(f"https://api.github.com/users/{username}")
    return response.json()

# Первый вызов — реальный запрос (1 сек)
data1 = fetch_github_user("torvalds")

# Второй вызов — из кэша (мгновенно)
data2 = fetch_github_user("torvalds")

print(data1 == data2)  # True

4. Кэш с использованием Redis (для распределённых систем)

import redis
import json
from typing import Any

class RedisCache:
    def __init__(self, host="localhost", port=6379):
        self.redis_client = redis.Redis(host=host, port=port, decode_responses=True)
    
    def get(self, key: str) -> Any:
        """Получить значение из кэша"""
        value = self.redis_client.get(key)
        if value:
            return json.loads(value)
        return None
    
    def set(self, key: str, value: Any, ttl: int = 3600):
        """Сохранить в кэш с TTL (время жизни)"""
        self.redis_client.setex(
            key,
            ttl,  # Секунды
            json.dumps(value)
        )

# Использование
cache = RedisCache()

# Первый вызов — вычисляем и кэшируем
key = "user:123"
user_data = cache.get(key)
if not user_data:
    print("Запрос в БД")
    user_data = {"id": 123, "name": "John"}
    cache.set(key, user_data, ttl=300)  # Кэш на 5 минут

# Второй вызов — из кэша
user_data = cache.get(key)  # Мгновенно

5. Кэширование на разных уровнях

# Level 1: Локальная память (в процессе)
from functools import lru_cache

@lru_cache(maxsize=100)
def get_config(key: str):
    # Для часто читаемых настроек
    pass

# Level 2: Redis (между микросервисами)
redis_cache = RedisCache()

# Level 3: CDN кэш (для статического контента)
# Настраивается на уровне HTTP заголовков

# Пример полного стека кэширования
def get_user(user_id: int):
    # Уровень 1: Локальная память
    cached = get_user.cache.get(user_id)
    if cached:
        return cached
    
    # Уровень 2: Redis
    key = f"user:{user_id}"
    cached = redis_cache.get(key)
    if cached:
        get_user.cache[user_id] = cached
        return cached
    
    # Уровень 3: БД
    print(f"Запрос в БД: {user_id}")
    user = db.session.query(User).get(user_id)
    
    # Сохраняем на оба уровня кэша
    get_user.cache[user_id] = user
    redis_cache.set(key, user.to_dict(), ttl=300)
    
    return user

get_user.cache = {}  # Локальный кэш

6. Кэширование в FastAPI

from fastapi import FastAPI, HTTPException
from functools import lru_cache
from datetime import datetime, timedelta

app = FastAPI()

@lru_cache(maxsize=32)
def get_expensive_data(limit: int):
    """Кэшируется как на уровне Python, так и HTTP"""
    import time
    time.sleep(2)
    return [{"id": i, "data": f"item {i}"} for i in range(limit)]

@app.get("/data")
async def get_data(limit: int = 10):
    """HTTP кэширование"""
    return {
        "data": get_expensive_data(limit),
        "cache_info": get_expensive_data.cache_info()._asdict()
    }

# Клиент автоматически кэширует благодаря HTTP заголовкам
# Cache-Control: max-age=300

7. Проблемы с кэшированием

# ПРОБЛЕМА 1: Невалидный кэш (stale cache)
# Решение: TTL (время жизни кэша)
redis_cache.set(key, value, ttl=300)  # Автоматический expiry

# ПРОБЛЕМА 2: Слишком большой кэш (memory leak)
# Решение: maxsize
@lru_cache(maxsize=128)  # Макс 128 записей
def my_function(x):
    pass

# ПРОБЛЕМА 3: Кэш затирается (cache invalidation)
# Решение: инвалидировать при изменении данных
def update_user(user_id: int, data):
    db.update_user(user_id, data)
    
    # Инвалидировать кэш
    redis_cache.delete(f"user:{user_id}")
    get_user.cache_clear()  # Для lru_cache

Реальное сравнение производительности

import timeit
import time

def slow_function(n):
    time.sleep(0.1)
    return n * 2

from functools import lru_cache

@lru_cache(maxsize=128)
def cached_function(n):
    time.sleep(0.1)
    return n * 2

# БЕЗ кэша — 10 вызовов
start = time.time()
for i in range(10):
    slow_function(5)
print(f"Без кэша: {time.time() - start:.2f}s")  # 1.0 сек

# С кэшем — 10 вызовов
start = time.time()
for i in range(10):
    cached_function(5)
print(f"С кэшем: {time.time() - start:.2f}s")  # 0.1 сек (только первый)

print(f"Ускорение: 10x")  # Минимум!

Лучшие практики

• Кэшируй то, что часто читается: конфиги, результаты API, данные БД
• Не кэшируй часто меняющиеся данные: балансы счётов, статусы заказов
• Установи TTL: автоматическое истечение кэша
• Инвалидируй при обновлении: очисти кэш при изменении данных
• Монитори попадания кэша: отслеживай эффективность

Итог

Кэширование ускоряет код на порядки, сохраняя результаты дорогостоящих операций. От простого до распределённых кэшей Redis — выбирай инструмент по задаче. Но помни: проблема кэширования — его инвалидация.