Redis: кэширование, очереди и ограничение скорости

Вы добавили Redis в свой стек, но используете его только как простое хранилище ключ-значение. Redis предлагает мощные структуры данных и команды, которые позволяют элегантно решать задачи кэширования, организации очередей и ограничения скорости. Это руководство покажет, как эффективно реализовать эти паттерны.

Почему Redis для этих сценариев?

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

Кэширование с помощью Redis

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

Базовый паттерн Cache-Aside

В паттерне cache-aside ваше приложение сначала проверяет Redis. Если данных там нет (промах кэша), оно извлекает их из базы данных, а затем сохраняет в Redis для следующего раза.

import redis
import json

r = redis.Redis(host='localhost', port=6379, db=0)

def get_user(user_id):
    cache_key = f'user:{user_id}'
    cached = r.get(cache_key)
    if cached:
        return json.loads(cached)
    user = fetch_user_from_db(user_id)  # ваш вызов БД
    r.setex(cache_key, 3600, json.dumps(user))  # кэш на 1 час
    return user

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

• Всегда устанавливайте время жизни (TTL), чтобы избежать устаревших данных.
• Используйте осмысленные имена ключей, например user:123:profile.
• Обрабатывайте инвалидацию кэша: при изменении данных обновляйте или удаляйте запись в кэше.
• Рассмотрите использование хэшей Redis для хранения нескольких полей объекта, чтобы сэкономить память.

Продвинутые паттерны кэширования

Для высоконагруженных приложений рассмотрите следующие паттерны:

• Write-through: обновление кэша и базы данных одновременно.
• Write-behind: сначала обновляется кэш, затем асинхронно база данных.
• Refresh-ahead: проактивное обновление записей кэша до истечения их срока действия.

Очереди с помощью Redis

Списки и потоки Redis идеально подходят для создания простых очередей сообщений. Используйте списки для базовых очередей задач, а потоки — для более сложной обработки событий.

Простая очередь задач с использованием списков

Используйте LPUSH для добавления задач и BRPOP для их обработки. BRPOP блокируется до появления задачи, что эффективно для воркеров.

# Производитель
r.lpush('jobs', json.dumps({'task': 'send_email', 'to': 'user@example.com'}))

# Потребитель (воркер)
while True:
    _, job_data = r.brpop('jobs', timeout=30)
    job = json.loads(job_data)
    process_job(job)

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

• Используйте отдельные списки для разных типов задач или приоритетов.
• Реализуйте очередь необработанных сообщений (dead-letter queue) для неудачных задач.
• Используйте BRPOPLPUSH (или BLMOVE в Redis 6.2+) для атомарного перемещения задач в список обработки, обеспечивая надёжность.
• Отслеживайте длину очереди с помощью LLEN для масштабирования воркеров.

Redis Streams для событийного sourcing

Redis Streams (появившиеся в Redis 5.0) предоставляют более надёжную очередь с группами потребителей, подтверждениями и воспроизведением. Используйте XADD для добавления событий и XREADGROUP для их потребления.

# Добавить событие
r.xadd('events', {'type': 'order_placed', 'order_id': '123'})

# Группа потребителей
r.xgroup_create('events', 'order_processors', id='0', mkstream=True)
while True:
    events = r.xreadgroup('order_processors', 'worker1', {'events': '>'}, count=10, block=5000)
    for event in events:
        process_event(event)
        r.xack('events', 'order_processors', event.id)

Ограничение скорости с помощью Redis

Ограничение скорости защищает ваши API от злоупотреблений и обеспечивает справедливое использование. Атомарные операции Redis делают его идеальным для подсчёта запросов.

Ограничение скорости с фиксированным окном

Увеличивайте счётчик для каждого пользователя и устанавливайте срок действия. Если счётчик превышает лимит, блокируйте запрос.

def is_allowed(user_id, limit=100, window=60):
    key = f'rate_limit:{user_id}'
    current = r.incr(key)
    if current == 1:
        r.expire(key, window)
    return current