Redis - гарячий шар поряд з PostgreSQL/MySQL, не primary store. У типовому backend-проєкті - шість ролей:

Мінімальний docker compose - Redis 8.6 + Redis Insight UI. Лежить у sandbox/docker-compose.yaml.

services:
  redis:
    image: redis:8.6
    command: ["redis-server", "/usr/local/etc/redis/redis.conf"]
    ports: ["6379:6379"]
    volumes:
      - ./redis.conf:/usr/local/etc/redis/redis.conf:ro
      - redis-data:/data
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 5s

  insight:
    image: redis/redisinsight:latest
    ports: ["5540:5540"]
    depends_on: { redis: { condition: service_healthy } }

volumes:
  redis-data:

cd ~/study/courses/redis/sandbox
docker compose up -d
docker compose exec redis redis-cli ping   # → PONG
# Insight UI: http://localhost:5540  (host=redis, port=6379)

PHP клієнт (phpredis)

// composer require ext-redis (PECL: pecl install redis)
$r = new \Redis();
$r->pconnect('127.0.0.1', 6379);   // persistent для PHP-FPM
$r->set('hello', 'world', ['ex' => 60]);
echo $r->get('hello');              // "world"

Python клієнт (redis-py)

# pip install redis
import redis

r = redis.Redis(host='127.0.0.1', port=6379, decode_responses=True)
r.set('hello', 'world', ex=60)
print(r.get('hello'))   # "world"

# Async варіант для FastAPI:
import redis.asyncio as aioredis
r = aioredis.Redis(host='127.0.0.1', port=6379, decode_responses=True)
await r.set('hello', 'world', ex=60)

Не намагайся пам'ятати усі 240+ команд Redis. Для дебагу свого коду вистачає шести.

docker compose exec redis redis-cli
# або:  redis-cli -h 127.0.0.1 -p 6379 -n 0

PING                          # → PONG  (живий?)
DBSIZE                        # → (integer) 42  (скільки ключів у поточній DB)

# Подивитися конкретний ключ:
TYPE user:42                  # → string | hash | list | set | zset | stream
TTL user:42                   # → 60 (сек) | -1 (без TTL) | -2 (нема ключа)
MEMORY USAGE user:42          # → 1234  (байтів)
OBJECT ENCODING user:42       # → raw | int | embstr | listpack | hashtable

# Перебір ключів за патерном:
SCAN 0 MATCH user:* COUNT 100  # cursor-based, безпечний
# Повторюй з новим cursor поки не повернеться 0.

KEYS user:*                    # ⛔ O(N) по всій базі - блокує сервер. Тільки у dev.

Чому SCAN, а не KEYS

Cache-aside (lazy loading) - 90% case-ів. Логіка: спершу читаю кеш, miss → іду в DB → кладу в кеш → повертаю.

Атомарний SET з TTL

# ⛔ ПОГАНО - два кроки, ризик "вічного" ключа якщо crash між ними:
SET user:42:cache "{json}"
EXPIRE user:42:cache 300

# ✓ ДОБРЕ - однією командою:
SET user:42:cache "{json}" EX 300
# або з NX (тільки якщо нема):
SET user:42:cache "{json}" NX EX 300

Jitter: розтягни TTL

Якщо 1000 ключів отримали TTL рівно 3600s - всі вони протухнуть одночасно і всі miss-и одночасно вдарять у DB. Розв'язання: додай випадковий offset.

// PHP
$ttl = 3600 + random_int(0, 600);    // 60-70 хв, не строго 60
$cache->set("user:$id", $data, $ttl);

# Python
import random
ttl = 3600 + random.randint(0, 600)
r.set(f"user:{id}", data, ex=ttl)

Laravel дає три фасади для більшості Redis-задач: Cache (cache-aside), Cache::tags() (tag invalidation), Cache::lock() (distributed lock).

// config/database.php - окрема connection під cache
'redis' => [
    'client' => env('REDIS_CLIENT', 'phpredis'),
    'default' => [
        'host'     => env('REDIS_HOST', '127.0.0.1'),
        'port'     => env('REDIS_PORT', '6379'),
        'database' => env('REDIS_DB', '0'),
    ],
    'cache' => [
        'database' => env('REDIS_CACHE_DB', '1'),   // окрема DB index
    ],
],

// .env
CACHE_STORE=redis
REDIS_CLIENT=phpredis

Cache-aside через remember

use Illuminate\Support\Facades\Cache;

// canonical pattern:
$user = Cache::remember("user:$id", 300, fn() => User::find($id));

// з jitter (Laravel не робить це автоматично):
$ttl = 300 + random_int(0, 60);
$user = Cache::remember("user:$id", $ttl, fn() => User::find($id));

// Atomic add (тільки якщо ще нема):
Cache::add("counter:$id", 0, 60);
Cache::increment("counter:$id");

Tag invalidation

// Тільки на Redis/Memcached driver (file/database driver не підтримує):
Cache::tags(['users', "user:$id"])->put("user:$id:profile", $data, 300);

// Інвалідуй всі ключі з тегом 'users' одразу:
Cache::tags(['users'])->flush();

Distributed lock

$lock = Cache::lock('export:run', 60);

if ($lock->get()) {
    try {
        // тільки один worker з усіх pod-ів зайде сюди
    } finally {
        $lock->release();
    }
}

// Або з блокуванням до 5 сек:
Cache::lock('export:run', 60)->block(5, function () {
    // ...
});

Symfony Cache реалізує PSR-6 і PSR-16. У production найчастіший вибір - cache.adapter.redis з DSN-конфігом.

# config/packages/cache.yaml
framework:
    cache:
        app: cache.adapter.redis
        default_redis_provider: '%env(REDIS_DSN)%'
        pools:
            cache.user:
                adapter: cache.adapter.redis_tag_aware  # для invalidateTags()
                default_lifetime: 3600
            cache.api:
                adapter: cache.adapter.redis
                default_lifetime: 60

# .env
REDIS_DSN=redis://localhost:6379/0
# З паролем: redis://:password@localhost:6379/0
# Кластер:   redis://h1:6379,h2:6379,h3:6379

Cache-aside через CacheInterface

use Symfony\Contracts\Cache\CacheInterface;
use Symfony\Contracts\Cache\ItemInterface;

class UserRepository
{
    public function __construct(private CacheInterface $cache) {}

    public function find(int $id): User
    {
        return $this->cache->get("user.$id", function (ItemInterface $item) use ($id) {
            $item->expiresAfter(300);
            return $this->db->findUser($id);
        });
    }
}

Tag invalidation з RedisTagAwareAdapter

use Symfony\Contracts\Cache\TagAwareCacheInterface;

class UserRepository
{
    public function __construct(private TagAwareCacheInterface $cache) {}

    public function find(int $id): User
    {
        return $this->cache->get("user.$id", function (ItemInterface $item) use ($id) {
            $item->expiresAfter(300);
            $item->tag(['users', "user.$id"]);   // помічаємо тегами
            return $this->db->findUser($id);
        });
    }

    public function invalidateAllUsers(): void
    {
        $this->cache->invalidateTags(['users']);  // один виклик скидає все
    }
}

У Python два світи: Django (sync) і FastAPI (async). Готові бібліотеки покривають обидва.

Django + django-redis

# pip install django-redis
# settings.py
CACHES = {
    "default": {
        "BACKEND": "django_redis.cache.RedisCache",
        "LOCATION": "redis://127.0.0.1:6379/1",
        "OPTIONS": {
            "CLIENT_CLASS": "django_redis.client.DefaultClient",
            "CONNECTION_POOL_KWARGS": {"max_connections": 50},
        },
        "KEY_PREFIX": "myapp",
        "TIMEOUT": 300,
    }
}
SESSION_ENGINE = "django.contrib.sessions.backends.cache"
SESSION_CACHE_ALIAS = "default"

from django.core.cache import cache

# Canonical pattern - get_or_set:
def get_user(user_id: int):
    return cache.get_or_set(
        f"user:{user_id}",
        lambda: User.objects.get(pk=user_id),
        timeout=300,
    )

# Tag-like (django-redis specific):
cache.set_many({"user:1": u1, "user:2": u2}, timeout=300)
cache.delete_pattern("user:*")   # SCAN-based, безпечно

FastAPI + fastapi-cache2

# pip install fastapi-cache2[redis]
from fastapi import FastAPI
from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend
from fastapi_cache.decorator import cache
import redis.asyncio as aioredis

app = FastAPI()

@app.on_event("startup")
async def startup():
    r = aioredis.from_url("redis://localhost:6379/0")
    FastAPICache.init(RedisBackend(r), prefix="myapp")

@app.get("/users/{user_id}")
@cache(expire=300)
async def get_user(user_id: int):
    return await db.fetch_user(user_id)   # тільки якщо cache miss

aiocache: альтернатива з більшою гнучкістю

# pip install aiocache
from aiocache import cached, Cache

@cached(ttl=300, cache=Cache.REDIS, key_builder=lambda f, *a, **kw: f"user:{a[0]}")
async def get_user(user_id: int):
    return await db.fetch_user(user_id)

Hot-key з TTL 60s протух. Одночасно 100 request-ів прийшли і всі попали у cache miss. Усі 100 одночасно б'ють SELECT в DB. БД лягає. Кеш "захист" перетворився на amplifier - це cache stampede (thundering herd).

Спосіб 1: TTL jitter (статистичний)

Розтягни TTL випадково. Дешево, але не гарантує - якщо у тебе один hot key, jitter не допоможе.

$ttl = 3600 + random_int(0, 600);   // +/-10%

Спосіб 2: Lock-on-miss

Перший процес з miss бере lock (SET NX EX) і обчислює. Решта чекають lock або повертають stale-дані. Класичний паттерн.

// Псевдокод
$data = Cache::get($key);
if ($data === null) {
    $lock = Cache::lock("lock:$key", 10);
    if ($lock->get()) {
        try {
            $data = expensiveQuery();
            Cache::put($key, $data, 300);
        } finally {
            $lock->release();
        }
    } else {
        usleep(100_000);                    // wait 100ms
        $data = Cache::get($key);           // спробуй знову
    }
}

Спосіб 3: Probabilistic early expiration (XFetch)

За якийсь час до реального TTL один з N запитів вирішує: "оновлю-но я кеш проактивно". TTL ще не сплив - 99% запитів отримують hit, 1% (probabilistic) рефрешать.

// Symfony - вбудовано:
$value = $cache->get('key', function (ItemInterface $item) {
    $item->expiresAfter(300);
    return computeValue();
}, beta: 2.0);  // higher beta = більш агресивний early refresh

# Python - руками (probabilistic):
import math, random, time
def get_with_xfetch(key, ttl, compute, beta=1.0):
    cached = r.get(key)
    if cached:
        expire_at = float(r.get(f"{key}:exp") or 0)
        # ймовірність рефрешу зростає з наближенням до expire
        if random.random() < -math.log(random.random()) * beta * (expire_at - time.time()):
            new_val = compute()
            r.set(key, new_val, ex=ttl)
            r.set(f"{key}:exp", time.time() + ttl, ex=ttl)
            return new_val
        return cached
    # ... звичайний cache-aside

Канонічний lock pattern на одному Redis - три рядки. Але є дві пастки, які роблять "наївну" реалізацію баговою.

# Acquire (atomic, NX = тільки якщо нема):
SET lock:export "token-uuid-7f3a..." NX PX 30000
# → OK    (захопили) | (nil) (хтось вже тримає)

# Worker робить роботу...

# Release - Lua compare-and-delete:
EVAL "if redis.call('GET',KEYS[1])==ARGV[1] then
        return redis.call('DEL',KEYS[1])
      else return 0 end" 1 lock:export "token-uuid-7f3a..."

Пастка 1: чому random token

Сценарій без token:

1. Worker A захопив lock:job з PX 30000.
2. A залип на 35 секунд (GC pause, мережа).
3. Lock протух за 30s, Worker B захопив.
4. A прокинувся, виконав DEL lock:job - видалив лок B.
5. Worker C захопив. Тепер B і C працюють одночасно. Critical section зламана.

З random token: A при release робить compare-and-delete. Бачить, що значення вже не його - не чіпає.

Пастка 2: TTL менший за роботу

Якщо PX 30000, а робота може зайняти 60s - це баг. Lock протухне посередині роботи, інший worker захопить, два будуть працювати.

Два рішення:

• TTL з запасом: PX = worst-case × 2.
• Refresh усередині job-у: продовжуєш PX поки працюєш (Symfony, python-redis-lock auto-renewal вміють).

Laravel: Cache::lock

use Illuminate\Support\Facades\Cache;

// 1) Простий try-acquire:
$lock = Cache::lock('export:run', 60);
if ($lock->get()) {
    try {
        $this->runExport();
    } finally {
        $lock->release();
    }
}

// 2) Closure-варіант (release у finally автоматично):
Cache::lock('export:run', 60)->get(function () {
    $this->runExport();
});

// 3) Блокувальний (чекає до 5s):
Cache::lock('export:run', 60)->block(5, function () {
    $this->runExport();
});

// 4) Передати lock у job, який release-нула інший процес:
$lock = Cache::lock('export:run', 60, 'shared-owner-id');
ProcessExportJob::dispatch($lock->owner());

Symfony: LockFactory + RedisStore

# config/packages/lock.yaml
framework:
    lock:
        default: 'redis://%env(REDIS_HOST)%:%env(REDIS_PORT)%'

use Symfony\Component\Lock\LockFactory;

class ExportService
{
    public function __construct(private LockFactory $lockFactory) {}

    public function run(): void
    {
        $lock = $this->lockFactory->createLock('export:run', ttl: 30);

        if (!$lock->acquire()) {
            return;   // вже працює інший worker
        }

        try {
            while (!$this->done()) {
                $this->processChunk();

                // refresh якщо лишилося менше 5с до expire:
                if ($lock->getRemainingLifetime() < 5) {
                    $lock->refresh();
                }
            }
        } finally {
            $lock->release();
        }
    }
}

У Python немає вбудованого фасаду на рівні Laravel/Symfony. Дві популярні бібліотеки покривають sync і async кейси.

Sync: python-redis-lock

# pip install python-redis-lock
import redis
import redis_lock

r = redis.Redis()

# Context manager (auto-release у finally):
with redis_lock.Lock(r, "export:run", expire=30, auto_renewal=True):
    run_export()

# Без context manager:
lock = redis_lock.Lock(r, "export:run", expire=30)
if lock.acquire(blocking=False):
    try:
        run_export()
    finally:
        lock.release()
else:
    return  # хтось працює

# auto_renewal=True - бібліотека запускає background thread,
# що продовжує lock кожні expire/3 секунд, поки не зробиш release.

Async: aioredlock (для FastAPI/asyncio)

# pip install aioredlock
from aioredlock import Aioredlock, LockError

lock_manager = Aioredlock([{
    "host": "localhost",
    "port": 6379,
}])

# Context manager:
async with await lock_manager.lock("export:run", lock_timeout=30) as lock:
    await run_export_async()
    # lock auto-extend через self.extend() при потребі
    if lock.valid:
        await lock.extend()

# Не забути закрити при shutdown:
await lock_manager.destroy()

Native: коли все це не треба

import uuid

def try_lock(r, name, ttl_ms=30000):
    token = uuid.uuid4().hex
    if r.set(f"lock:{name}", token, nx=True, px=ttl_ms):
        return token
    return None

def release_lock(r, name, token):
    lua = """
    if redis.call('GET', KEYS[1]) == ARGV[1] then
        return redis.call('DEL', KEYS[1])
    else return 0 end
    """
    return r.eval(lua, 1, f"lock:{name}", token)

token = try_lock(r, "export:run")
if token:
    try:
        run_export()
    finally:
        release_lock(r, "export:run", token)

Redis має дві природні структури для черги. Кожен фреймворк вибрав одну з них під капот.

Laravel Queue: чому ще List

Історичні причини - List був стабільним з Redis 2.x. Laravel робить atomic Lua-script: RPOPLPUSH queues:emails queues:emails:reserved + ZADD у "delayed-retry" sorted set з timestamp. Якщо worker не delete job за retry_after секунд - інший worker reclaims.

Symfony Messenger: чому Streams

Symfony стартував з Streams після їх стабілізації у Redis 5+ (2018). PEL дає природний at-least-once з retry без Lua-обгортки.

Python: Celery і RQ

Celery використовує комбінацію List + sorted set (нижче деталі). RQ - чистий List + Lua. Stream-based варіанти у Python менш популярні (Faust був, але не активний).

// config/queue.php
'connections' => [
    'redis' => [
        'driver'       => 'redis',
        'connection'   => 'default',
        'queue'        => env('REDIS_QUEUE', 'default'),
        'retry_after'  => 90,           // повертає у чергу через 90с
        'block_for'    => 5,            // BRPOP timeout (sec)
        'after_commit' => true,         // dispatch тільки після DB commit
    ],
],
// .env
QUEUE_CONNECTION=redis

Створення job-у

// app/Jobs/SendInvoiceEmail.php
class SendInvoiceEmail implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    public int $tries = 3;
    public int $backoff = 60;       // 60s між retry
    public int $timeout = 120;      // worker kill після 120s
    public bool $deleteWhenMissingModels = true;   // не fail якщо модель видалена

    public function __construct(public int $invoiceId) {}

    public function handle(InvoiceMailer $mailer): void
    {
        $mailer->send(Invoice::findOrFail($this->invoiceId));
    }

    public function failed(\Throwable $e): void
    {
        Log::error('Email job failed', ['invoice' => $this->invoiceId, 'error' => $e]);
    }
}

Dispatch і worker

// Dispatch (з контролера/сервісу):
SendInvoiceEmail::dispatch($invoice->id);                       // одразу
SendInvoiceEmail::dispatch($invoice->id)->delay(now()->addMinutes(10));
SendInvoiceEmail::dispatch($invoice->id)->onQueue('emails');    // окрема черга

# Worker (запускається через supervisor / systemd):
php artisan queue:work redis \
    --queue=emails,default \
    --tries=3 \
    --max-time=3600 \
    --max-jobs=1000 \
    --sleep=3

# Або горизонтальне масштабування:
php artisan horizon

Idempotency

// На випадок retry - захист від подвійної відправки:
public function handle(InvoiceMailer $mailer): void
{
    $key = "email-sent:invoice:{$this->invoiceId}";

    if (Cache::add($key, 1, 86400)) {   // SET NX EX - atomic
        $mailer->send(Invoice::findOrFail($this->invoiceId));
    }
}

# config/packages/messenger.yaml
framework:
    messenger:
        transports:
            async:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                options:
                    stream: 'messages'
                    group: 'symfony'
                    consumer: '%env(HOSTNAME)%'        # ВАЖЛИВО: унікально
                    delete_after_ack: true
                    stream_max_entries: 10000          # auto-trim
                    redeliver_timeout: 3600            # claim забутих через 1 год

            failed: 'doctrine://default?queue_name=failed'

        failure_transport: failed

        routing:
            'App\Message\SendInvoiceEmail': async

# .env
MESSENGER_TRANSPORT_DSN=redis://localhost:6379/messages

Message + Handler

// src/Message/SendInvoiceEmail.php
final class SendInvoiceEmail
{
    public function __construct(public readonly int $invoiceId) {}
}

// src/MessageHandler/SendInvoiceEmailHandler.php
use Symfony\Component\Messenger\Attribute\AsMessageHandler;

#[AsMessageHandler]
final class SendInvoiceEmailHandler
{
    public function __construct(private InvoiceMailer $mailer) {}

    public function __invoke(SendInvoiceEmail $msg): void
    {
        $this->mailer->send($msg->invoiceId);
    }
}

Dispatch і consume

// Dispatch
$this->messageBus->dispatch(new SendInvoiceEmail($invoice->getId()));

// З delay і retry stamps:
use Symfony\Component\Messenger\Stamp\DelayStamp;

$this->messageBus->dispatch(
    new SendInvoiceEmail($invoice->getId()),
    [new DelayStamp(10_000)],   // через 10 секунд
);

# Consumer (k8s/supervisor):
php bin/console messenger:consume async \
    --keepalive \
    --time-limit=3600 \
    --memory-limit=128M \
    --limit=1000

# Failed messages
php bin/console messenger:failed:show
php bin/console messenger:failed:retry
php bin/console messenger:failed:remove 7

# pip install celery redis
# tasks.py
from celery import Celery

app = Celery(
    'myapp',
    broker='redis://localhost:6379/0',
    backend='redis://localhost:6379/1',   # result backend (опційно)
)

app.conf.update(
    task_acks_late=True,           # ack після завершення (не на старті)
    task_reject_on_worker_lost=True,
    task_default_retry_delay=60,
    task_max_retries=3,
    broker_transport_options={
        'visibility_timeout': 3600,  # 1 година - DEFAULT, часто треба збільшити
    },
)

@app.task(bind=True, autoretry_for=(Exception,), retry_backoff=True)
def send_invoice_email(self, invoice_id: int):
    try:
        send_email(invoice_id)
    except SmtpError as e:
        raise self.retry(exc=e, countdown=60, max_retries=5)

# Dispatch
send_invoice_email.delay(invoice_id)                          # одразу
send_invoice_email.apply_async((invoice_id,), countdown=600)  # через 10 хв
send_invoice_email.apply_async((invoice_id,), eta=tomorrow)   # на конкретний час

# Worker
celery -A tasks worker --loglevel=info --concurrency=4

# Beat (для periodic tasks):
celery -A tasks beat --loglevel=info

# Flower (UI на :5555):
celery -A tasks flower

Gotcha: visibility_timeout

Celery з Redis broker працює так: task береться з queue, починає виконуватися. Якщо worker впав - інший worker через visibility_timeout (default 1 година) візьме той самий task знову.

Idempotency

@app.task(bind=True)
def send_invoice_email(self, invoice_id: int):
    # SET NX EX - atomic, повертає True якщо встановив
    if not r.set(f"sent:invoice:{invoice_id}", 1, nx=True, ex=86400):
        return  # вже відправлено
    send_email(invoice_id)

Celery - потужний, але великий. Для невеликої кодової бази або проєктів, де "магія" Celery overkill - є простіші опції.

RQ (Redis Queue)

# pip install rq
import redis
from rq import Queue

r = redis.Redis()
q = Queue('default', connection=r)

# Звичайна функція - не task class:
def send_invoice_email(invoice_id: int):
    # ...
    pass

# Enqueue:
job = q.enqueue(send_invoice_email, 42)
job = q.enqueue(send_invoice_email, 42, job_timeout=300, retry=Retry(max=3))

# Status:
print(job.get_status())  # queued | started | finished | failed
print(job.result)

# Worker
rq worker default high low
# Або з RQ Dashboard:  rq-dashboard

Dramatiq

# pip install dramatiq[redis]
import dramatiq
from dramatiq.brokers.redis import RedisBroker

dramatiq.set_broker(RedisBroker(host='localhost'))

@dramatiq.actor(max_retries=3, time_limit=60_000)
def send_invoice_email(invoice_id: int):
    # ...
    pass

# Dispatch:
send_invoice_email.send(42)
send_invoice_email.send_with_options(args=(42,), delay=10_000)  # 10s

# Worker
dramatiq tasks

Матриця вибору

1. Fixed window (INCR + EXPIRE)

# 100 req/min на user_id:
# key = "rl:user:42:"

INCR rl:user:42:2026-05-17T11:30
# Якщо результат 1 - тільки що створили, треба EXPIRE:
EXPIRE rl:user:42:2026-05-17T11:30 60
# Якщо результат > 100 - deny.

// PHP реалізація
$minute = floor(time() / 60);
$key    = "rl:user:$userId:$minute";

$count = $redis->incr($key);
if ($count === 1) {
    $redis->expire($key, 60);
}

if ($count > 100) {
    abort(429, 'Too many requests');
}

Просто, 5 рядків коду, 1 INCR на запит. Недолік: burst на межі вікна. 100 req на 59-й секунді хвилини A + 100 req на 1-й секунді хвилини B = 200 req за ~2 секунди (хоча "ліміт" 100/хв).

2. Sliding window log (Sorted Set)

# Запам'ятовуємо timestamp кожного req у sorted set:
ZADD rl:user:42 1747900000123 "uuid-of-this-req"
# Прибираємо все старше 60 секунд:
ZREMRANGEBYSCORE rl:user:42 0 (1747900000123 - 60000)
# Лічимо скільки лишилось:
ZCARD rl:user:42
# Якщо > 100 - deny.
EXPIRE rl:user:42 60

# Python (atomic через pipeline)
def is_allowed(r, user_id: int, limit: int = 100, window: int = 60) -> bool:
    now_ms = int(time.time() * 1000)
    key = f"rl:user:{user_id}"

    with r.pipeline() as p:
        p.zadd(key, {f"{now_ms}-{uuid.uuid4()}": now_ms})
        p.zremrangebyscore(key, 0, now_ms - window * 1000)
        p.zcard(key)
        p.expire(key, window)
        _, _, count, _ = p.execute()

    return count <= limit

Точне sliding window. Недолік: memory росте лінійно по traffic (один elem на кожен запит).

Коли який

Кожен фреймворк має вбудований rate limiter поверх Redis. Брати готовий.

Laravel: RateLimiter facade + middleware

// app/Providers/AppServiceProvider.php
use Illuminate\Cache\RateLimiting\Limit;
use Illuminate\Support\Facades\RateLimiter;

RateLimiter::for('api', function (Request $request) {
    return Limit::perMinute(100)->by($request->user()?->id ?: $request->ip());
});

RateLimiter::for('password-reset', function (Request $request) {
    return [
        Limit::perMinute(5)->by($request->ip()),
        Limit::perHour(20)->by($request->input('email')),
    ];
});

// routes/api.php
Route::middleware(['throttle:api'])->group(function () {
    Route::get('/users', ...);
});

Symfony: rate_limiter

# config/packages/rate_limiter.yaml
framework:
    rate_limiter:
        api:
            policy: 'sliding_window'
            limit: 100
            interval: '1 minute'
        password_reset:
            policy: 'token_bucket'
            limit: 5
            rate: { interval: '15 minutes', amount: 5 }

use Symfony\Component\RateLimiter\RateLimiterFactory;

class ApiController
{
    public function __construct(private RateLimiterFactory $apiLimiter) {}

    public function index(Request $request): Response
    {
        $limiter = $this->apiLimiter->create($request->getClientIp());
        if (false === $limiter->consume(1)->isAccepted()) {
            throw new TooManyRequestsHttpException();
        }
        return new Response('OK');
    }
}

Python: FastAPI + slowapi

# pip install slowapi
from slowapi import Limiter
from slowapi.util import get_remote_address
from fastapi import FastAPI, Request

limiter = Limiter(
    key_func=get_remote_address,
    storage_uri="redis://localhost:6379",
)
app = FastAPI()
app.state.limiter = limiter

@app.get("/api/data")
@limiter.limit("100/minute")
async def get_data(request: Request):
    return {"data": "..."}

Python: Django + django-ratelimit

# pip install django-ratelimit
from django_ratelimit.decorators import ratelimit

@ratelimit(key='user_or_ip', rate='100/m', method='GET', block=True)
def api_view(request):
    return JsonResponse({"data": "..."})

Шість core структур Redis. Кожна на одне-два призначення. Decision tree з прикладами.

Decision tree

Один скаляр? (counter, cache value)
  → String (SET / INCR)

Декілька полів одного об'єкта, апдейтиш частинами?
  → Hash (HSET / HGET)

Колекція з порядком вставки?
  → List (LPUSH / RPOP)

Колекція унікальних значень без порядку?
  → Set (SADD / SISMEMBER)

Колекція з sorting за числом?
  → Sorted Set (ZADD / ZRANGE)

Append-only log з replay і consumer groups?
  → Stream (XADD / XREADGROUP)

Лічильники - класична Redis-задача. INCR/HINCRBY атомарні безкоштовно (single-threaded core).

Денний counter API hits

# Key per-day:
INCR api:calls:2026-05-17
EXPIREAT api:calls:2026-05-17 1747958399   # абсолютний timestamp = end of day

# Або з floating TTL:
INCR api:calls:counter
EXPIRE api:calls:counter 86400   # ✗ скинеться кожен раз - НЕ це треба

// PHP - atomic INCR + EXPIRE на створенні
$key = "api:calls:" . date('Y-m-d');
$count = $redis->incr($key);
if ($count === 1) {
    $redis->expireAt($key, strtotime('tomorrow'));
}

Per-user struct counters (HINCRBY)

# Один Hash з кількома counter-полями:
HINCRBY user:42:stats logins 1
HINCRBY user:42:stats views 1
HINCRBY user:42:stats messages_sent 1

HGETALL user:42:stats
# 1) "logins"        2) "15"
# 3) "views"         4) "127"
# 5) "messages_sent" 6) "8"

# Python
r.hincrby(f"user:{user_id}:stats", "logins", 1)
r.hincrby(f"user:{user_id}:stats", "views", 1)
stats = r.hgetall(f"user:{user_id}:stats")

Лічильник з reset (rate limit як приклад)

# Window key з timestamp:
SET counter:hour:$h "0" EX 3600 NX   # NX = тільки якщо нема
INCR counter:hour:$h

Float counters

INCRBYFLOAT balance:42 10.5    # для double-precision
HINCRBYFLOAT user:42 balance 10.5

Sessions у Redis

Чому Redis для session: shared store між multiple instances backend (load balancer перенаправляє між pod-ами), TTL автоматично, fast read.

// Laravel: .env
SESSION_DRIVER=redis
SESSION_LIFETIME=120
SESSION_CONNECTION=default

# Symfony: config/packages/framework.yaml
framework:
    session:
        handler_id: 'session.handler.redis'
        cookie_secure: auto
        cookie_samesite: lax
        gc_maxlifetime: 7200

services:
    Redis:
        class: Redis
        calls:
            - method: connect
              arguments: ['%env(REDIS_HOST)%', '%env(int:REDIS_PORT)%']

    session.handler.redis:
        class: Symfony\Component\HttpFoundation\Session\Storage\Handler\RedisSessionHandler
        arguments: ['@Redis']

# Django: settings.py
SESSION_ENGINE = "django.contrib.sessions.backends.cache"
SESSION_CACHE_ALIAS = "default"   # використовує django-redis backend
SESSION_COOKIE_AGE = 86400

Leaderboard на Sorted Set

# Додати/оновити score:
ZADD lb:players 1500 "alice"
ZADD lb:players 2300 "bob"
ZADD lb:players 1800 "carol"

# Тільки якщо новий score БІЛЬШИЙ (для high-score):
ZADD lb:players GT 2500 "alice"

# Топ-10 (descending):
ZRANGE lb:players 0 9 REV WITHSCORES

# Позиція гравця (descending rank, 0-indexed):
ZREVRANK lb:players "alice"     # → 0 (перше місце)
ZSCORE lb:players "alice"        # → "2500"

# Atomic інкремент score:
ZINCRBY lb:players 100 "alice"   # +100 поінтів

# Players навколо конкретного (rank 5-15):
ZRANGE lb:players 5 14 REV WITHSCORES

# Python приклад
def get_top10(r):
    return r.zrange("lb:players", 0, 9, desc=True, withscores=True)

def add_points(r, player: str, points: int):
    return r.zincrby("lb:players", points, player)

def get_rank(r, player: str) -> int | None:
    rank = r.zrevrank("lb:players", player)
    return rank + 1 if rank is not None else None

Сценарій: "у тебе у коді щось не працює". Не знаєш чому ключ повертається або не повертається. Шість команд закриють 95% дебагу.

# Уявімо, у коді ти кешуєш як "user:42":

EXISTS user:42                 # → 1 (є) | 0 (нема). Перевірка факту.
TYPE user:42                   # → string | hash | list | set | zset | stream
TTL user:42                    # → 60 (сек) | -1 (без TTL) | -2 (нема)
MEMORY USAGE user:42           # → 1234 (байтів)
OBJECT ENCODING user:42        # → raw | int | embstr | listpack | hashtable

# Прочитати значення відповідно до типу:
GET user:42                    # для string
HGETALL user:42                # для hash
LRANGE user:42 0 -1            # для list
SMEMBERS user:42               # для set
ZRANGE user:42 0 -1 WITHSCORES # для zset

Типові ситуації

SCAN замість KEYS

SCAN 0 MATCH user:* COUNT 100
# Перший виклик повертає (cursor, [keys...]).
# Повторюй з новим cursor поки cursor не стане 0.

# У PHP:
$cursor = 0;
do {
    [$cursor, $keys] = $redis->scan($cursor, ['MATCH' => 'user:*', 'COUNT' => 100]);
    foreach ($keys as $key) { /* process */ }
} while ($cursor != 0);

# У Python:
for key in r.scan_iter(match="user:*", count=100):
    process(key)

Сценарій: "у нас щось гальмує, чи це Redis?". Чотири інструменти, кожен закриває свій клас проблем.

1. INFO stats: hit rate і eviction

INFO stats
# # Stats
# keyspace_hits:182456
# keyspace_misses:24193
# evicted_keys:0
# expired_keys:8732
# ...

# Hit rate = hits / (hits + misses):
# 182456 / (182456 + 24193) = 88% - OK, >80% задовільно.

2. redis-cli --bigkeys

docker compose exec redis redis-cli --bigkeys
# -------- summary -------
# Biggest string found 'sessions:abc' has 524288 bytes
# Biggest hash   found 'user:42:cache' has 8192 fields
# Biggest list   found 'log:tail'      has 100000 items
# Biggest set    found 'tags:popular'  has 50000 members
# Biggest zset   found 'leaderboard'   has 1000000 members

Сканує весь keyspace і знаходить найбільші ключі по кожному типу. Безпечно для production (cursor-based, з throttle).

3. redis-cli --hotkeys (потребує allkeys-lfu)

# спершу налаштуй (у redis.conf):
# maxmemory-policy allkeys-lfu

docker compose exec redis redis-cli --hotkeys
# -------- summary -------
# hot key found with counter: 1024 keyname:'product:bestseller:42'
# hot key found with counter: 892  keyname:'config:global'

4. SLOWLOG: повільні команди

CONFIG SET slowlog-log-slower-than 10000   # threshold 10ms (у μs)
CONFIG SET slowlog-max-len 128             # тримати 128 останніх

# подивитися:
SLOWLOG GET 10
# 1) 1) (integer) 14                # ID
#    2) (integer) 1747900000        # unix timestamp
#    3) (integer) 25000             # тривалість у μs (25ms)
#    4) 1) "HGETALL"
#       2) "user:42:everything"     # сама команда
#    5) "127.0.0.1:54321"           # client
#    6) "client-name"

SLOWLOG RESET                              # очистити

1. KEYS у production

KEYS user:*       # ⛔ O(N) - блокує весь Redis на секунди при млн ключів
SCAN 0 MATCH user:* COUNT 100   # ✓ cursor-based, безпечно

2. SET + EXPIRE замість SET EX

SET key val             # ⛔ якщо процес упаде між двома командами -
EXPIRE key 60           #    key лишиться вічним

SET key val EX 60       # ✓ atomic

3. DEL big key (1M+ полів)

DEL big_hash            # ⛔ синхронне видалення блокує Redis
UNLINK big_hash         # ✓ async free, реальне видалення у background thread

4. HGETALL / SMEMBERS / LRANGE 0 -1 на великих

HGETALL user:42:everything   # ⛔ якщо 100k полів - 100k items у одній відповіді
HSCAN user:42:everything 0 COUNT 100   # ✓ cursor-based

5. Cache без TTL + ручна invalidation

Cache::forever($key, $data);         // ⛔ один пропущений edge case = stale назавжди
Cache::put($key, $data, 86400);      // ✓ навіть 24 години краще за "ніколи"

6. Python: connection-per-request без pool

def handle_request():
    r = redis.Redis(host='localhost')   # ⛔ нова TCP-сесія на кожен запит
    return r.get('key')

# ✓ Module-level singleton:
pool = redis.ConnectionPool(host='localhost', max_connections=50)
r = redis.Redis(connection_pool=pool)

def handle_request():
    return r.get('key')

7. PHP-FPM: connect замість pconnect

$r = new \Redis();
$r->connect('redis', 6379);    // ⛔ нова сесія на кожен FPM request
$r->pconnect('redis', 6379);   // ✓ persistent, переживає request

8. Lua script >50ms

-- ⛔ Lua блокує ВСІХ клієнтів на час виконання.
-- Якщо тут цикл по 1M елементах - весь Redis стоїть.
for i, v in ipairs(redis.call('LRANGE', KEYS[1], 0, -1)) do
    redis.call('HSET', 'result:' .. v, 'flag', 1)
end

-- ✓ Винось важку логіку в app, у Lua тільки compare-and-set / acquire-lock / dec-counter.

9. Cross-slot у Cluster без hash tag

# ⛔ якщо у Redis Cluster - ключі можуть бути на різних shard-ах:
MGET user:42:profile user:42:posts        # error: CROSSSLOT

# ✓ Hash tag {42} - однаковий slot:
MGET user:{42}:profile user:{42}:posts    # OK

10. SUBSCRIBE / PUBLISH замість Streams

# ⛔ Pub/Sub - at-most-once, без persistence:
SUBSCRIBE notifications
PUBLISH notifications "{...}"
# Subscriber offline → повідомлення втрачено

# ✓ Stream - persistent, replay, ack:
XADD events * type "notification" payload "{...}"
XREADGROUP GROUP workers worker-1 COUNT 10 BLOCK 5000 STREAMS events >