Что используем как ключ в Redis?

Redis ключи: структура, соглашения и best practices

Что такое Redis ключ

Redis ключ — это уникальный идентификатор для значения в Redis. Это просто строка, которая связана с данными (строка, список, хеш, сет и т.д.).

import redis

redis_client = redis.Redis(host='localhost', port=6379)

# Установка значения
redis_client.set("user:123:name", "John")  # ключ: user:123:name
redis_client.set("cache:products:list", json.dumps([...]))  # ключ: cache:products:list

# Получение значения
name = redis_client.get("user:123:name")  # John

Структура и соглашения для ключей

Правило 1: Используй namespace через двоеточие

# ❌ Плохо: неясная иерархия
redis_client.set("user_data_123", value)
redis_client.set("product_cache_456", value)
redis_client.set("session_789", value)

# ✅ Хорошо: иерархическая структура
redis_client.set("user:123:profile", value)
redis_client.set("user:123:preferences", value)
redis_client.set("product:456:cache", value)
redis_client.set("session:789:data", value)
redis_client.set("cache:products:list", value)

Правило 2: Включай контекст и идентификатор

# Формат: namespace:entity_type:entity_id:field

def get_cache_key(entity_type: str, entity_id: int, field: str = None) -> str:
    """Генерирует Redis ключ по стандарту."""
    if field:
        return f"cache:{entity_type}:{entity_id}:{field}"
    return f"cache:{entity_type}:{entity_id}"

# Примеры использования
key = get_cache_key("user", 123, "profile")
# "cache:user:123:profile"

key = get_cache_key("product", 456)
# "cache:product:456"

Правило 3: Используй версионирование для обновлений

from datetime import datetime

def get_versioned_key(entity_type: str, entity_id: int, version: str = "v1") -> str:
    """Добавляем версию для совместимости при обновлениях."""
    return f"cache:{entity_type}:{entity_id}:{version}"

# Когда нужно изменить структуру данных
redis_client.set(get_versioned_key("user", 123, "v1"), old_format)
redis_client.set(get_versioned_key("user", 123, "v2"), new_format)

# Клиент автоматически выбирает v2
# v1 кеш постепенно вытесняется

Типы ключей и их использование

1. Простые строковые ключи (session, auth token)

# Хранение сессии
redis_client.setex(
    f"session:{session_id}",
    ex=86400,  # TTL: 24 часа
    value=json.dumps({
        "user_id": 123,
        "username": "john",
        "logged_in_at": "2024-03-23T10:00:00Z"
    })
)

# Извлечение
session_data = json.loads(redis_client.get(f"session:{session_id}"))

2. Ключи для кеширования (часто используемые данные)

from functools import wraps
import hashlib

def cache_key(prefix: str, *args, **kwargs) -> str:
    """Генерирует ключ на основе аргументов функции."""
    args_hash = hashlib.md5(
        json.dumps([args, kwargs], sort_keys=True).encode()
    ).hexdigest()
    return f"cache:{prefix}:{args_hash}"

def redis_cache(ttl: int = 3600):
    """Декоратор для кеширования результата функции."""
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            key = cache_key(func.__name__, *args, **kwargs)
            
            # Проверяем кеш
            cached = redis_client.get(key)
            if cached:
                return json.loads(cached)
            
            # Вычисляем результат
            result = await func(*args, **kwargs)
            
            # Сохраняем в кеш
            redis_client.setex(
                key,
                ex=ttl,
                value=json.dumps(result)
            )
            
            return result
        return wrapper
    return decorator

# Использование
@redis_cache(ttl=3600)
async def get_user_profile(user_id: int):
    return await db.fetch_user(user_id)

# Ключ будет: cache:get_user_profile:<hash(123)>

3. Ключи для распределённых счётчиков и лимитов

# Rate limiting
def rate_limit_key(user_id: int, endpoint: str, window: str = "minute") -> str:
    return f"rate_limit:{endpoint}:{user_id}:{window}"

async def check_rate_limit(user_id: int, endpoint: str, limit: int = 100) -> bool:
    """Проверяет, не превышен ли лимит запросов."""
    key = rate_limit_key(user_id, endpoint)
    
    # Увеличиваем счётчик
    count = redis_client.incr(key)
    
    # Устанавливаем TTL при первом запросе
    if count == 1:
        redis_client.expire(key, 60)  # 60 секунд
    
    # Проверяем лимит
    if count > limit:
        raise RateLimitExceeded(f"User {user_id} exceeded limit")
    
    return True

# Использование
await check_rate_limit(user_id=123, endpoint="/api/posts", limit=10)

4. Ключи для очередей (job queue, message queue)

from rq import Queue
from rq.job import Job

# Redis используется для RQ (Redis Queue)
q = Queue(connection=redis_client)

# Ключи автоматически генерируются
def send_email(user_email):
    # Отправляет email асинхронно
    pass

job = q.enqueue(send_email, "user@example.com")
# Redis ключ: rq:queue:default
# Redis ключ: rq:job:<job_id>

# Проверка статуса
print(job.get_status())  # queued, started, finished, failed

5. Ключи для Real-time данных (pub/sub, stream)

# Publish-Subscribe
def publish_update(channel: str, message: dict):
    redis_client.publish(channel, json.dumps(message))

# Ключ здесь — название канала
publish_update("user:123:updates", {"status": "online"})

# Подписка
subscriber = redis_client.pubsub()
subscriber.subscribe("user:123:updates")

for message in subscriber.listen():
    if message['type'] == 'message':
        print(f"Received: {message['data']}")

Правильные и неправильные ключи

# ❌ НЕПРАВИЛЬНО
redis_client.set("data", value)  # Слишком общий
redis_client.set("u123", value)  # Неясный смысл
redis_client.set("product_cache_456_name", value)  # Неоднородный формат
redis_client.set("user-123-profile", value)  # Дефис вместо двоеточия

# ✅ ПРАВИЛЬНО
redis_client.set("cache:user:123:profile", value)
redis_client.set("session:abc123xyz", value)
redis_client.set("rate_limit:api:user:123", value)
redis_client.set("queue:email:pending", value)
redis_client.set("lock:order:456", value)  # Распределённая блокировка

Управление ключами

1. Очистка старых ключей (TTL)

import time

def set_cache_with_ttl(key: str, value: dict, ttl: int = 3600):
    """Сохраняет в кеш с автоматическим истечением."""
    redis_client.setex(
        key,
        ex=ttl,  # Time To Live
        value=json.dumps(value)
    )

# Примеры TTL
TTL = {
    "user_profile": 3600,      # 1 час
    "product_list": 1800,       # 30 минут
    "session": 86400,           # 24 часа
    "temp_verification": 300,   # 5 минут
}

for key_type, ttl in TTL.items():
    redis_client.setex(f"cache:{key_type}", ex=ttl, value=data)

2. Поиск и удаление ключей

# Поиск по паттерну
def find_keys(pattern: str):
    """Находит все ключи по паттерну."""
    return redis_client.keys(pattern)

# Примеры
user_sessions = redis_client.keys("session:user:123:*")
expired_caches = redis_client.keys("cache:product:*")

# Удаление
for key in user_sessions:
    redis_client.delete(key)

# Или с pipeline для эффективности
pipeline = redis_client.pipeline()
for key in user_sessions:
    pipeline.delete(key)
pipeline.execute()

3. Мониторинг ключей

# Количество ключей
db_info = redis_client.info("keyspace")
print(f"Total keys: {db_info['db0']['keys']}")

# Размер памяти
memory_info = redis_client.info("memory")
print(f"Memory used: {memory_info['used_memory_human']}")

# Поиск больших ключей
def find_large_keys(pattern="*"):
    """Находит самые большие ключи."""
    keys = redis_client.keys(pattern)
    sizes = [
        (key, redis_client.memory_usage(key))
        for key in keys
    ]
    return sorted(sizes, key=lambda x: x[1], reverse=True)[:10]

large_keys = find_large_keys()
for key, size in large_keys:
    print(f"{key}: {size} bytes")

Стратегии именования ключей

Вариант 1: Простой (для小 projects)

# cache:entity_id:field
redis_client.set("cache:123:profile", user_profile)
redis_client.set("cache:123:preferences", preferences)

Вариант 2: Иерархический (рекомендуется)

# namespace:module:entity_type:entity_id:field
redis_client.set("app:cache:user:123:profile", data)
redis_client.set("app:session:abc123", session_data)
redis_client.set("app:queue:email:pending", queue_data)

Вариант 3: С хешированием (для длинных ключей)

import hashlib

def create_hash_key(prefix: str, *args):
    hash_value = hashlib.sha256(
        json.dumps(args).encode()
    ).hexdigest()[:16]
    return f"{prefix}:{hash_value}"

# Для кеша результата сложной функции
key = create_hash_key("cache:search", query, filters, page)
redis_client.set(key, results, ex=3600)

Реальный пример: Caching Layer

class RedisCacheLayer:
    def __init__(self, redis_client):
        self.redis = redis_client
    
    def cache_key(self, entity_type: str, entity_id: int) -> str:
        return f"cache:{entity_type}:{entity_id}"
    
    async def get(self, entity_type: str, entity_id: int, loader_func):
        """Cache-aside pattern."""
        key = self.cache_key(entity_type, entity_id)
        
        # Проверяем кеш
        cached = self.redis.get(key)
        if cached:
            return json.loads(cached)
        
        # Загружаем из источника
        data = await loader_func(entity_id)
        
        # Сохраняем в кеш
        self.redis.setex(key, ex=3600, value=json.dumps(data))
        return data
    
    def invalidate(self, entity_type: str, entity_id: int) -> None:
        """Инвалидирует кеш."""
        key = self.cache_key(entity_type, entity_id)
        self.redis.delete(key)

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

user = await cache.get(
    "user",
    123,
    loader_func=db.fetch_user
)

# После обновления очищаем кеш
cache.invalidate("user", 123)

Итог

Ключи Redis — это основа для организации данных. Правильно структурированные ключи:

1. Иерархичны — используют двоеточие для namespace
2. Информативны — понятно, что в них хранится
3. Уникальны — без коллизий
4. Управляемы — легко удалять и искать
5. Вместимы — уместный TTL и размер

Хороший ключ Redis — это чистый, понятный контракт между кодом и хранилищем.