Redis Refresher

Setup & Environment (Docker-based)

The fastest way to get Redis running is Docker. No installation of Redis itself is needed on the host — only redis-cli (or RedisInsight) for the client side.

Start Redis with Docker

# Start Redis 7 in the background, exposed on host port 6379
docker run -d \
  --name redis-refresher \
  -p 6379:6379 \
  redis:7-alpine

# Verify the container is running
docker ps --filter name=redis-refresher

Install redis-cli

# macOS: Homebrew installs redis-cli (and redis-server, but we use Docker)
brew install redis

# Verify
redis-cli --version
# redis-cli 7.x.x

# OR: use the CLI inside the container without installing anything on host
docker exec -it redis-refresher redis-cli

Verify the Connection

# Connect to the Docker container's Redis
redis-cli -h 127.0.0.1 -p 6379

# Inside the REPL:
127.0.0.1:6379> PING
PONG

127.0.0.1:6379> SET hello world
OK

127.0.0.1:6379> GET hello
"world"

127.0.0.1:6379> EXIT

Python Client

# Install the redis-py library
pip install redis

# Or with uv
uv pip install redis

import redis

# Connect to local Redis (defaults: host='localhost', port=6379, db=0)
r = redis.Redis(host='127.0.0.1', port=6379, db=0, decode_responses=True)

# decode_responses=True means GET returns str, not bytes
print(r.ping())       # True
print(r.set('hello', 'world'))   # True
print(r.get('hello'))            # 'world'

GUI: RedisInsight

# Free desktop app — best way to browse keys visually
brew install --cask redisinsight

# Open it, add a connection: host=127.0.0.1, port=6379

Cleanup

# Stop and remove the container when done
docker stop redis-refresher && docker rm redis-refresher

Core Concepts

Redis is an in-memory data structure store that can be used as a database, cache, message broker, and streaming engine. Understanding why it is fast and how it persists data is the foundation for using it correctly.

Why Redis Is Fast

• In-memory storage — all data lives in RAM. Reads and writes are sub-millisecond because there is no disk I/O on the hot path.
• Single-threaded event loop — one thread handles all commands. No locking overhead, no context-switch cost. Redis 6+ added I/O threading for network, but command execution remains single-threaded.
• Simple data model — operations are O(1) or O(log N) by design. No query planning, no joins.
• Efficient data structures — internally Redis uses skip lists, hash tables, and intsets tuned for memory and speed.

Data Type Overview

TTL and Key Expiration

Every key can have a time-to-live (TTL) in milliseconds or seconds. Redis uses a combination of lazy expiration (checked on access) and active expiration (background scan of a random sample of volatile keys) to reclaim memory.

# Set a key with a 60-second TTL
SET session:abc123 "{...}" EX 60

# Check remaining TTL (returns -1 if no expiry, -2 if key gone)
TTL session:abc123
# (integer) 58

# Set TTL on an existing key
EXPIRE mykey 300

# Persist (remove TTL)
PERSIST mykey

# Millisecond precision
PEXPIRE mykey 1500
PTTL mykey

Persistence Options (Overview)

Redis offers two persistence mechanisms — covered in depth in the Persistence & Durability section:

• RDB (Redis Database) — point-in-time snapshots. Compact, fast restarts, slight data loss risk on crash.
• AOF (Append-Only File) — log of every write command. Near-zero data loss, larger files, slower restart.
• RDB + AOF hybrid — recommended for production. Uses RDB base with AOF tail for durability.

Strings

The most basic Redis type. A string value is binary-safe — it can hold any byte sequence up to 512 MB. Strings are used for caching, counters, session tokens, and feature flags.

Basic Operations

# SET and GET
SET name "Alice"
GET name                    # "Alice"

# Set with TTL (EX = seconds, PX = milliseconds)
SET token:abc "jwt_payload" EX 3600
TTL token:abc               # 3600

# NX = only set if key does NOT exist (atomic "create if absent")
SET lock:resource "owner_1" NX EX 30
# (nil) if key already exists — no overwrite

# XX = only set if key EXISTS
SET counter 0 XX            # (nil) if counter doesn't exist

# GET and SET in one atomic operation (Redis 6.2+)
GETSET name "Bob"           # returns "Alice", sets to "Bob"
GETDEL name                 # returns "Bob", deletes key

Multi-key Operations

# Set/get multiple keys atomically
MSET user:1:name "Alice" user:1:age "30" user:2:name "Bob"
MGET user:1:name user:1:age user:2:name
# 1) "Alice"
# 2) "30"
# 3) "Bob"

# Append to a string
SET greeting "Hello"
APPEND greeting ", World!"
GET greeting                # "Hello, World!"
STRLEN greeting             # (integer) 13

Atomic Counters

INCR is atomic — it is safe to call from multiple clients simultaneously. This is the canonical Redis pattern for counters, rate limiters, and ID generators.

# Initialize and increment
SET page_views 0
INCR page_views             # (integer) 1
INCR page_views             # (integer) 2
INCRBY page_views 10        # (integer) 12
DECR page_views             # (integer) 11
DECRBY page_views 5         # (integer) 6

# Float increment
SET price 9.99
INCRBYFLOAT price 0.01      # "10"

import redis

r = redis.Redis(host='127.0.0.1', port=6379, decode_responses=True)

# Atomic counter — safe from multiple threads/processes
r.set('visits', 0)
r.incr('visits')            # 1
r.incrby('visits', 9)       # 10

# Conditional set (NX = not exists)
acquired = r.set('lock:job_42', 'worker_1', nx=True, ex=30)
if acquired:
    print("Lock acquired, processing job")
else:
    print("Job already being processed")

Lists

Redis Lists are linked lists of strings. They support O(1) push/pop from both ends, making them ideal for queues (RPUSH / BLPOP) and stacks (LPUSH / LPOP).

Push, Pop, and Range

# LPUSH = insert at HEAD, RPUSH = insert at TAIL
RPUSH tasks "task_1"        # (integer) 1
RPUSH tasks "task_2"        # (integer) 2
LPUSH tasks "task_0"        # (integer) 3 — prepend

# Inspect
LLEN tasks                  # (integer) 3
LRANGE tasks 0 -1           # 1) "task_0"  2) "task_1"  3) "task_2"
LINDEX tasks 1              # "task_1"

# Pop from head or tail
LPOP tasks                  # "task_0"
RPOP tasks                  # "task_2"

# Pop multiple elements at once (Redis 6.2+)
RPUSH queue a b c d e
LPOP queue 3                # 1) "a"  2) "b"  3) "c"

Trim and Remove

# Keep only the most recent 100 items (sliding window log pattern)
RPUSH feed "event_1"
RPUSH feed "event_2"
# ... many more ...
LTRIM feed -100 -1          # Keep last 100 only

# Remove occurrences of a value
# LREM key count value
# count > 0: remove from head, count < 0: from tail, count = 0: all
RPUSH mylist "a" "b" "a" "c" "a"
LREM mylist 2 "a"           # removes 2 "a" from head → "b" "c" "a"

Blocking Pop (Queue Consumer)

BLPOP and BRPOP block until an element is available or the timeout expires. This is how you implement a reliable background worker queue without polling.

# Block up to 5 seconds waiting for an element
BLPOP jobs 5
# Returns: 1) "jobs"  2) "task_body"   (or nil on timeout)

import redis
import threading

r = redis.Redis(host='127.0.0.1', port=6379, decode_responses=True)

def producer():
    import time
    time.sleep(1)
    r.rpush('jobs', 'process_order_42')
    print("Producer: pushed job")

def consumer():
    print("Consumer: waiting for jobs...")
    # Block up to 10 seconds; returns (queue_name, value) or None
    result = r.blpop('jobs', timeout=10)
    if result:
        queue, job = result
        print(f"Consumer: processing {job} from {queue}")

t = threading.Thread(target=producer)
t.start()
consumer()
t.join()

Sets

Redis Sets are unordered collections of unique strings. Add, remove, and membership checks are O(1). Set operations (union, intersection, difference) are O(N).

Basic Operations

# Add members (duplicates are silently ignored)
SADD tags "python" "redis" "backend"   # (integer) 3
SADD tags "python"                     # (integer) 0 — already exists

# Check membership
SISMEMBER tags "redis"      # (integer) 1 (true)
SISMEMBER tags "java"       # (integer) 0 (false)

# List all members (order not guaranteed)
SMEMBERS tags               # 1) "python"  2) "redis"  3) "backend"

# Cardinality
SCARD tags                  # (integer) 3

# Remove
SREM tags "backend"         # (integer) 1

# Pop a random member (destructive)
SPOP tags                   # e.g. "python"

# Return random members without removing
SRANDMEMBER tags 2          # returns 2 random members

Set Operations

SADD user:1:skills "python" "redis" "sql"
SADD user:2:skills "python" "go" "sql"

# Intersection — skills both users share
SINTER user:1:skills user:2:skills
# 1) "python"  2) "sql"

# Union — all skills across both users
SUNION user:1:skills user:2:skills
# 1) "python"  2) "redis"  3) "sql"  4) "go"

# Difference — skills user:1 has that user:2 does not
SDIFF user:1:skills user:2:skills
# 1) "redis"

# Store result into a new key (non-blocking, returns count)
SINTERSTORE common:skills user:1:skills user:2:skills

r = redis.Redis(host='127.0.0.1', port=6379, decode_responses=True)

# Track unique visitors per day
r.sadd('visitors:2026-02-23', 'user:1001', 'user:1002', 'user:1001')
daily_uniques = r.scard('visitors:2026-02-23')  # 2

# Find users who visited both days (returning visitors)
r.sadd('visitors:2026-02-22', 'user:1001', 'user:1003')
returning = r.sinter('visitors:2026-02-22', 'visitors:2026-02-23')
# {'user:1001'}

Sorted Sets

Sorted Sets (ZSets) associate each member with a floating-point score. Members are always sorted by score. Internally backed by a skip list + hash table, giving O(log N) add/remove and O(log N + M) range queries.

Basic Operations

# ZADD key score member [score member ...]
ZADD leaderboard 1500 "alice"
ZADD leaderboard 2300 "bob"
ZADD leaderboard 1800 "charlie"
ZADD leaderboard 2300 "dave"    # same score as bob — ordered alphabetically

# Get rank (0-indexed, low score = rank 0)
ZRANK leaderboard "alice"       # (integer) 0
ZRANK leaderboard "bob"         # (integer) 2

# Get rank from the highest score end
ZREVRANK leaderboard "bob"      # (integer) 0  (top of leaderboard)

# Get score of a member
ZSCORE leaderboard "charlie"    # "1800"

# Cardinality
ZCARD leaderboard               # (integer) 4

# Remove
ZREM leaderboard "dave"

Range Queries

# Range by index (ascending, inclusive)
ZRANGE leaderboard 0 -1             # all members, lowest first
ZRANGE leaderboard 0 -1 WITHSCORES # include scores

# Range by index (descending — top of leaderboard)
ZREVRANGE leaderboard 0 2 WITHSCORES
# 1) "bob"    2) "2300"
# 3) "charlie" 4) "1800"
# 5) "alice"  6) "1500"

# Range by score
ZRANGEBYSCORE leaderboard 1600 2500 WITHSCORES
# charlie (1800), bob (2300)

ZRANGEBYSCORE leaderboard -inf +inf LIMIT 0 3   # pagination

# Count members in a score range
ZCOUNT leaderboard 1600 2500    # (integer) 2

Increment Score

# Atomically add to a member's score
ZINCRBY leaderboard 500 "alice"     # alice: 1500 → 2000
ZSCORE leaderboard "alice"          # "2000"

r = redis.Redis(host='127.0.0.1', port=6379, decode_responses=True)

# Build a leaderboard
r.zadd('scores', {'alice': 1500, 'bob': 2300, 'charlie': 1800})

# Increment score
r.zincrby('scores', 200, 'alice')       # alice is now 1700

# Top 3 players (highest score first)
top3 = r.zrevrange('scores', 0, 2, withscores=True)
for rank, (name, score) in enumerate(top3, 1):
    print(f"#{rank} {name}: {int(score)}")
# #1 bob: 2300
# #2 alice: 1700
# #3 charlie: 1800

Hashes

Hashes store field-value pairs under a single key, equivalent to a dictionary or a row in a table. They are the natural fit for storing objects because related fields are co-located and you can update individual fields without fetching and reserializing the entire object.

Basic Operations

# HSET key field value [field value ...]
HSET user:1001 name "Alice" email "[email protected]" age 30 role "engineer"

# Get a single field
HGET user:1001 name         # "Alice"

# Get multiple fields
HMGET user:1001 name email  # 1) "Alice"  2) "[email protected]"

# Get everything
HGETALL user:1001
# 1) "name"   2) "Alice"
# 3) "email"  4) "[email protected]"
# 5) "age"    6) "30"
# 7) "role"   8) "engineer"

# Check if field exists
HEXISTS user:1001 email     # (integer) 1

# Delete a field
HDEL user:1001 role

# Count fields
HLEN user:1001              # (integer) 3

# List keys and values separately
HKEYS user:1001             # name, email, age
HVALS user:1001             # Alice, [email protected], 30

Numeric Fields

# Integer increment on a hash field (atomic)
HSET stats:api calls 0 errors 0
HINCRBY stats:api calls 1
HINCRBY stats:api calls 1
HINCRBYFLOAT stats:api latency_ms 12.5

HGETALL stats:api
# calls: 2, errors: 0, latency_ms: 12.5

r = redis.Redis(host='127.0.0.1', port=6379, decode_responses=True)

# Store a user object as a hash
r.hset('user:1001', mapping={
    'name': 'Alice',
    'email': '[email protected]',
    'age': 30,
    'login_count': 0,
})

# Update a single field without fetching everything
r.hset('user:1001', 'age', 31)

# Atomic counter on a hash field
r.hincrby('user:1001', 'login_count', 1)

# Retrieve as a dict
user = r.hgetall('user:1001')
print(user)
# {'name': 'Alice', 'email': '[email protected]', 'age': '31', 'login_count': '1'}

# Hash vs. JSON string: prefer hash when you update individual fields frequently.
# Use JSON string (via SET) when you always read/write the whole object at once.

Streams

Redis Streams (added in Redis 5.0) are an append-only log data structure. Each entry has a unique auto-generated ID (millisecond-sequence) and a set of fields. Streams support consumer groups for fan-out delivery with acknowledgement — making them suitable for event sourcing and durable message queues.

XADD — Append to Stream

# XADD key ID field value [field value ...]
# Use * for auto-generated ID: <millseconds>-<sequence>
XADD events * type "page_view" user "alice" path "/home"
# Returns: "1708600000000-0"

XADD events * type "click" user "alice" element "signup-btn"
XADD events * type "page_view" user "bob"   path "/pricing"

# Length of stream
XLEN events             # (integer) 3

XRANGE — Read by ID Range

# Read all entries
XRANGE events - +
# 1) 1) "1708600000000-0"
#    2) 1) "type"  2) "page_view"  3) "user"  4) "alice" ...

# Read from a specific ID (exclusive: use (ID)
XRANGE events (1708600000000-0 +

# Read last N entries
XREVRANGE events + - COUNT 2

XREAD — Tail the Stream

# Read up to 10 new messages starting after ID "0"
XREAD COUNT 10 STREAMS events 0

# Block waiting for new messages (like tail -f)
# Use "$" to only see messages arriving after this call
XREAD COUNT 1 BLOCK 0 STREAMS events $

Consumer Groups

Consumer groups allow multiple consumers to share the work of processing a stream. Each message is delivered to exactly one consumer in the group. Consumers must ACK messages to remove them from the pending list.

# Create a consumer group starting at the beginning (0) or live ($)
XGROUP CREATE events workers $ MKSTREAM

# Consumer "worker-1" reads 1 undelivered message
XREADGROUP GROUP workers worker-1 COUNT 1 STREAMS events >
# ">" means: give me messages not yet delivered to anyone in this group

# Acknowledge processing (removes from pending list)
XACK events workers 1708600000000-0

# Inspect what's pending (delivered but not yet ACKed)
XPENDING events workers - + 10

import redis
import time

r = redis.Redis(host='127.0.0.1', port=6379, decode_responses=True)

# Producer: append events
r.xadd('orders', {'order_id': 'ord_1', 'amount': '59.99', 'user': 'alice'})
r.xadd('orders', {'order_id': 'ord_2', 'amount': '12.00', 'user': 'bob'})

# Create consumer group (ignore if already exists)
try:
    r.xgroup_create('orders', 'fulfillment', id='0', mkstream=True)
except redis.exceptions.ResponseError:
    pass  # group already exists

# Worker: read and process
messages = r.xreadgroup(
    groupname='fulfillment',
    consumername='worker-1',
    streams={'orders': '>'},  # '>' = new, undelivered messages
    count=10,
    block=2000,               # block up to 2 seconds
)

if messages:
    for stream_name, entries in messages:
        for entry_id, fields in entries:
            print(f"Processing order {fields['order_id']} for {fields['user']}")
            # ... do the work ...
            r.xack('orders', 'fulfillment', entry_id)
            print(f"ACKed {entry_id}")

Pub/Sub

Redis Pub/Sub implements the publish-subscribe messaging pattern. Publishers push messages to channels; subscribers receive them in real time. Messages are fire-and-forget — they are not persisted and are lost if no subscriber is listening.

Commands

# Terminal 1: subscriber
SUBSCRIBE notifications
# Waiting for messages...

# Terminal 2: publisher
PUBLISH notifications "User alice just signed up"
# (integer) 1  — number of subscribers who received it

# Pattern subscription: subscribe to all channels matching a glob
PSUBSCRIBE user:*       # matches user:login, user:logout, user:purchase, etc.

# Unsubscribe
UNSUBSCRIBE notifications
PUNSUBSCRIBE user:*

# List active channels and subscription counts
PUBSUB CHANNELS         # list channels with at least one subscriber
PUBSUB NUMSUB channel1  # subscriber count for a channel

import redis
import threading
import time

def subscriber():
    # A separate connection is required for pub/sub
    r_sub = redis.Redis(host='127.0.0.1', port=6379, decode_responses=True)
    p = r_sub.pubsub()
    p.subscribe('notifications')

    print("Subscriber: listening...")
    for message in p.listen():
        if message['type'] == 'message':
            print(f"Subscriber received: {message['data']}")
            break  # exit after first message

def publisher():
    r_pub = redis.Redis(host='127.0.0.1', port=6379, decode_responses=True)
    time.sleep(0.5)   # let subscriber connect first
    recipient_count = r_pub.publish('notifications', 'Hello from publisher!')
    print(f"Publisher: sent to {recipient_count} subscribers")

t = threading.Thread(target=subscriber)
t.start()
publisher()
t.join()

Limitations and When to Use Streams Instead

Transactions & Scripting

Redis provides two mechanisms for atomic multi-operation sequences: MULTI/EXEC transactions and Lua scripting. Neither supports rollback on business logic errors — they only guarantee that commands run without interleaving.

MULTI / EXEC

# Queue multiple commands, then execute atomically
MULTI
INCR balance:alice
DECR balance:bob
SET transfer:log "alice->bob:100"
EXEC
# 1) (integer) 1100
# 2) (integer) 900
# 3) OK

# Discard a queued transaction
MULTI
SET foo bar
DISCARD   # clears the queue, no commands executed

WATCH — Optimistic Locking

WATCH implements optimistic concurrency control (OCC). If a watched key changes between WATCH and EXEC, the transaction returns nil (no-op). The client must retry.

import redis

r = redis.Redis(host='127.0.0.1', port=6379, decode_responses=True)

def transfer(from_key: str, to_key: str, amount: int) -> bool:
    """Atomic transfer using WATCH + MULTI/EXEC (optimistic locking)."""
    with r.pipeline() as pipe:
        while True:
            try:
                pipe.watch(from_key, to_key)   # start watching

                from_balance = int(pipe.get(from_key) or 0)
                if from_balance < amount:
                    pipe.reset()
                    return False  # insufficient funds

                # Enter MULTI block — commands are now queued, not executed
                pipe.multi()
                pipe.decrby(from_key, amount)
                pipe.incrby(to_key, amount)
                pipe.execute()   # atomically execute; raises WatchError if key changed
                return True

            except redis.WatchError:
                # Another client modified the keys — retry
                continue

r.set('balance:alice', 1000)
r.set('balance:bob', 500)
success = transfer('balance:alice', 'balance:bob', 200)
print(f"Transfer {'succeeded' if success else 'failed'}")
print(r.get('balance:alice'))  # 800
print(r.get('balance:bob'))    # 700

Lua Scripting

Lua scripts run atomically — no other command executes while the script is running. This is the preferred way to implement custom atomic operations that MULTI/EXEC cannot express (e.g., conditional logic mid-execution).

# EVAL script numkeys key [key ...] arg [arg ...]
# KEYS[1], KEYS[2]... and ARGV[1], ARGV[2]... inside the script

# Atomic "get-and-increment-if-below-limit"
EVAL "
  local current = tonumber(redis.call('GET', KEYS[1])) or 0
  if current < tonumber(ARGV[1]) then
    return redis.call('INCR', KEYS[1])
  else
    return -1
  end
" 1 counter 100

import redis

r = redis.Redis(host='127.0.0.1', port=6379, decode_responses=True)

# Load script once, reference by SHA1 for efficiency
SET_IF_LESS = """
local current = tonumber(redis.call('GET', KEYS[1])) or 0
if current < tonumber(ARGV[1]) then
    return redis.call('INCR', KEYS[1])
else
    return -1
end
"""

# Register once
sha = r.script_load(SET_IF_LESS)

r.set('counter', 98)
print(r.evalsha(sha, 1, 'counter', 100))  # 99 (incremented)
print(r.evalsha(sha, 1, 'counter', 100))  # 100 (incremented)
print(r.evalsha(sha, 1, 'counter', 100))  # -1 (limit reached)

Persistence & Durability

By default Redis is an in-memory store — a crash without persistence means data loss. Choosing the right persistence strategy is a production-critical decision that trades off performance, durability, and recovery time.

RDB — Point-in-Time Snapshots

RDB produces a compact binary snapshot of the entire dataset at a given moment. Redis forks the process to create the snapshot in the background (BGSAVE), so production traffic is unaffected.

# Manual snapshot (blocks until done — avoid in production)
SAVE

# Non-blocking background snapshot
BGSAVE

# Check snapshot status
LASTSAVE      # Unix timestamp of last successful save

# In redis.conf — auto-save rules:
# save <seconds> <changes>
# save 900 1      # if at least 1 key changed in 900 seconds
# save 300 10     # if at least 10 keys changed in 300 seconds
# save 60 10000   # if at least 10000 keys changed in 60 seconds

AOF — Append-Only File

AOF logs every write command. On restart Redis replays the log to reconstruct state. The fsync policy controls the durability vs. performance trade-off.

# In redis.conf
appendonly yes
appendfilename "appendonly.aof"

# fsync policy options:
# appendfsync always   — fsync after every write (safest, slowest ~1K ops/sec)
# appendfsync everysec — fsync every second (default, at most 1 sec data loss)
# appendfsync no       — OS decides (fastest, potentially several seconds loss)

appendfsync everysec

# AOF rewrite (compact the log in the background)
BGREWRITEAOF

RDB + AOF Hybrid (Recommended)

# redis.conf — enable both
save 900 1
appendonly yes
appendfsync everysec

# Redis 4.0+ hybrid format: AOF file begins with RDB snapshot, followed by
# incremental AOF commands. Combines fast load time with minimal data loss.
aof-use-rdb-preamble yes

Recovery Scenarios

Production Use Cases

This section covers the most common real-world Redis patterns with complete Python implementations using the redis-py library.

Session Store

Redis is the canonical session store for horizontally-scaled web applications. Sessions are serialized as JSON strings (or hashes) with a TTL. Any app server can look up any session without affinity.

import redis
import json
import secrets
import time

r = redis.Redis(host='127.0.0.1', port=6379, decode_responses=True)

SESSION_TTL = 3600  # 1 hour

def create_session(user_id: str, metadata: dict) -> str:
    """Create a session and return the session token."""
    token = secrets.token_urlsafe(32)
    key = f"session:{token}"
    payload = json.dumps({
        'user_id': user_id,
        'created_at': int(time.time()),
        **metadata,
    })
    r.set(key, payload, ex=SESSION_TTL)
    return token

def get_session(token: str) -> dict | None:
    """Retrieve session data, returning None if expired or not found."""
    key = f"session:{token}"
    data = r.get(key)
    if data is None:
        return None
    # Sliding expiry: reset TTL on each access
    r.expire(key, SESSION_TTL)
    return json.loads(data)

def invalidate_session(token: str) -> None:
    r.delete(f"session:{token}")

# Usage
token = create_session('user:1001', {'role': 'admin', 'ip': '1.2.3.4'})
session = get_session(token)
print(session)  # {'user_id': 'user:1001', 'created_at': ..., 'role': 'admin', ...}
invalidate_session(token)
print(get_session(token))  # None

Cache-Aside Pattern

Cache-aside (lazy loading) is the most common caching pattern: the application checks the cache first, fetches from the database on a miss, then populates the cache.

import redis
import json
from typing import Callable, TypeVar

r = redis.Redis(host='127.0.0.1', port=6379, decode_responses=True)

T = TypeVar('T')

def cache_aside(
    key: str,
    fetch_fn: Callable[[], T],
    ttl: int = 300,
    serializer=json,
) -> T:
    """
    Generic cache-aside wrapper.

    1. Check cache
    2. On miss: call fetch_fn (e.g., DB query)
    3. Populate cache with TTL
    4. Return result
    """
    cached = r.get(key)
    if cached is not None:
        return serializer.loads(cached)

    # Cache miss — fetch from source of truth
    value = fetch_fn()
    if value is not None:
        r.set(key, serializer.dumps(value), ex=ttl)
    return value

# Example: cache a database query result
def get_user_from_db(user_id: str) -> dict:
    # Simulate slow DB call
    import time; time.sleep(0.05)
    return {'id': user_id, 'name': 'Alice', 'email': '[email protected]'}

user = cache_aside(
    key='user:1001',
    fetch_fn=lambda: get_user_from_db('1001'),
    ttl=300,
)
print(user)  # fetched from DB
user = cache_aside(
    key='user:1001',
    fetch_fn=lambda: get_user_from_db('1001'),
    ttl=300,
)
print(user)  # served from cache

Rate Limiting

Two canonical implementations: fixed window (simple, slight burst at window boundary) and sliding window (accurate, uses sorted sets).

import redis
import time

r = redis.Redis(host='127.0.0.1', port=6379, decode_responses=True)

# ── Fixed Window ─────────────────────────────────────────────────────────────
def fixed_window_allow(user_id: str, limit: int, window_sec: int) -> bool:
    """
    Allow up to `limit` requests per `window_sec` second window.
    Simple and cheap: one INCR + one EXPIRE per request.
    """
    key = f"ratelimit:fixed:{user_id}:{int(time.time()) // window_sec}"
    count = r.incr(key)
    if count == 1:
        r.expire(key, window_sec)  # set TTL only on first request in window
    return count <= limit

# ── Sliding Window (Sorted Set) ───────────────────────────────────────────────
def sliding_window_allow(user_id: str, limit: int, window_sec: int) -> bool:
    """
    Allow up to `limit` requests in any rolling `window_sec` second period.
    Uses a sorted set where score = timestamp.
    More accurate than fixed window; no burst at window boundary.
    """
    now = time.time()
    key = f"ratelimit:sliding:{user_id}"
    window_start = now - window_sec

    pipe = r.pipeline()
    # Remove requests outside the window
    pipe.zremrangebyscore(key, '-inf', window_start)
    # Add current request (score=timestamp, member=unique request ID)
    pipe.zadd(key, {str(now): now})
    # Count requests in window
    pipe.zcard(key)
    # Expire the key after the window (cleanup)
    pipe.expire(key, window_sec + 1)
    results = pipe.execute()

    request_count = results[2]
    return request_count <= limit

# Usage
for i in range(12):
    allowed = sliding_window_allow('user:1001', limit=10, window_sec=60)
    print(f"Request {i+1}: {'ALLOWED' if allowed else 'BLOCKED'}")

Distributed Locking

A distributed lock prevents multiple instances from processing the same job concurrently. The naive SETNX approach has a correctness flaw — use the atomic SET key value NX EX form instead.

import redis
import uuid
import time
import contextlib

r = redis.Redis(host='127.0.0.1', port=6379, decode_responses=True)

@contextlib.contextmanager
def distributed_lock(resource: str, ttl: int = 30):
    """
    Context manager for a simple single-node distributed lock.

    Uses SET NX EX (atomic) to acquire.
    Uses a Lua script to release — prevents a client from deleting
    a lock it doesn't own (e.g., if TTL expired and another client acquired it).
    """
    lock_key = f"lock:{resource}"
    # Unique value so we only release our own lock
    lock_value = str(uuid.uuid4())

    # Atomic acquire: SET if Not eXists with EXpiry
    acquired = r.set(lock_key, lock_value, nx=True, ex=ttl)
    if not acquired:
        raise RuntimeError(f"Could not acquire lock for {resource}")

    # Lua script: only delete if the value matches (atomic check-and-delete)
    release_script = """
        if redis.call("GET", KEYS[1]) == ARGV[1] then
            return redis.call("DEL", KEYS[1])
        else
            return 0
        end
    """
    try:
        yield
    finally:
        r.eval(release_script, 1, lock_key, lock_value)

# Usage
try:
    with distributed_lock('job:process_report_42', ttl=30):
        print("Processing report — exclusively locked")
        time.sleep(0.1)  # simulate work
        print("Done — lock released automatically")
except RuntimeError as e:
    print(f"Could not run: {e}")

Real-time Leaderboard

import redis

r = redis.Redis(host='127.0.0.1', port=6379, decode_responses=True)

LEADERBOARD = 'game:leaderboard'

def record_score(user_id: str, score_delta: float) -> float:
    """Add score_delta to a user's total. Returns new total."""
    return r.zincrby(LEADERBOARD, score_delta, user_id)

def get_rank(user_id: str) -> int | None:
    """Returns 1-indexed rank from the top (1 = highest score)."""
    rank = r.zrevrank(LEADERBOARD, user_id)
    return None if rank is None else rank + 1

def get_top_n(n: int = 10) -> list[dict]:
    """Return the top N players with their scores."""
    entries = r.zrevrange(LEADERBOARD, 0, n - 1, withscores=True)
    return [
        {'rank': i + 1, 'user': user, 'score': int(score)}
        for i, (user, score) in enumerate(entries)
    ]

def get_surrounding(user_id: str, window: int = 2) -> list[dict]:
    """Return a user's rank plus `window` players above and below."""
    rank = r.zrevrank(LEADERBOARD, user_id)
    if rank is None:
        return []
    start = max(0, rank - window)
    end = rank + window
    entries = r.zrevrange(LEADERBOARD, start, end, withscores=True)
    return [
        {'rank': start + i + 1, 'user': u, 'score': int(s)}
        for i, (u, s) in enumerate(entries)
    ]

# Demo
for name, score in [('alice', 1500), ('bob', 2300), ('charlie', 1800), ('dave', 900)]:
    r.zadd(LEADERBOARD, {name: score})

record_score('alice', 300)   # alice: 1800

print(get_top_n(3))
# [{'rank': 1, 'user': 'bob', 'score': 2300},
#  {'rank': 2, 'user': 'alice', 'score': 1800},
#  {'rank': 3, 'user': 'charlie', 'score': 1800}]

print(f"Alice's rank: {get_rank('alice')}")  # 2 or 3 depending on tie-breaking

Message Queue

import redis
import json
import time

r = redis.Redis(host='127.0.0.1', port=6379, decode_responses=True)

QUEUE = 'jobs:email'

# ── Simple Queue (List-based) ─────────────────────────────────────────────────
def enqueue(queue: str, payload: dict) -> None:
    """Push a job to the tail of the queue."""
    r.rpush(queue, json.dumps(payload))

def dequeue(queue: str, timeout: int = 5) -> dict | None:
    """Pop a job from the head (blocking). Returns None on timeout."""
    result = r.blpop(queue, timeout=timeout)
    if result is None:
        return None
    _, raw = result
    return json.loads(raw)

# ── Reliable Queue (with processing list) ────────────────────────────────────
PROCESSING = 'jobs:email:processing'

def reliable_dequeue(queue: str, processing_queue: str) -> dict | None:
    """
    Atomically move job from pending to processing list (BRPOPLPUSH).
    If the worker crashes, jobs remain in the processing list for recovery.
    """
    # BRPOPLPUSH is deprecated in Redis 6.2 — use LMOVE instead
    raw = r.lmove(queue, processing_queue, 'LEFT', 'RIGHT', timeout=5)
    return json.loads(raw) if raw else None

def ack_job(processing_queue: str, payload: dict) -> None:
    """Remove from processing list after successful completion."""
    r.lrem(processing_queue, 1, json.dumps(payload))

# Producer
enqueue(QUEUE, {'to': '[email protected]', 'subject': 'Welcome!'})
enqueue(QUEUE, {'to': '[email protected]',   'subject': 'Your order shipped'})

# Consumer
job = dequeue(QUEUE)
if job:
    print(f"Sending email to {job['to']}: {job['subject']}")

Full-text Search with RediSearch

RediSearch is a Redis module (bundled in Redis Stack) that adds full-text search, secondary indexing, and aggregation capabilities.

# Start Redis Stack (includes RediSearch, RedisJSON, etc.)
docker run -d \
  --name redis-stack \
  -p 6379:6379 \
  redis/redis-stack-server:latest

# Create an index over hash keys with prefix "product:"
FT.CREATE product_idx
  ON HASH PREFIX 1 product:
  SCHEMA
    name TEXT WEIGHT 2.0
    description TEXT
    category TAG
    price NUMERIC

# Add some products
HSET product:1 name "Redis in Action" description "Deep dive into Redis" category "book" price 39
HSET product:2 name "Redis Cookbook" description "Redis recipes and patterns" category "book" price 29

# Full-text search
FT.SEARCH product_idx "redis patterns" RETURN 3 name description price

# Filter + full-text
FT.SEARCH product_idx "@category:{book} @price:[0 35]" SORTBY price ASC

import redis

# Requires redis-py 4.x+ with RediSearch support
r = redis.Redis(host='127.0.0.1', port=6379, decode_responses=True)

# Create index programmatically
from redis.commands.search.field import TextField, NumericField, TagField
from redis.commands.search.indexDefinition import IndexDefinition, IndexType

try:
    r.ft('product_idx').create_index(
        [
            TextField('name', weight=2.0),
            TextField('description'),
            TagField('category'),
            NumericField('price'),
        ],
        definition=IndexDefinition(prefix=['product:'], index_type=IndexType.HASH),
    )
except Exception:
    pass  # index already exists

# Index documents
r.hset('product:1', mapping={'name': 'Redis in Action', 'description': 'Deep dive', 'category': 'book', 'price': 39})
r.hset('product:2', mapping={'name': 'Redis Cookbook', 'description': 'Recipes and patterns', 'category': 'book', 'price': 29})

# Search
from redis.commands.search.query import Query
results = r.ft('product_idx').search(Query('redis').return_fields('name', 'price'))
for doc in results.docs:
    print(f"{doc.name}: ${doc.price}")

Clustering & High Availability

A standalone Redis server is a single point of failure. Redis provides two HA solutions: Sentinel for automatic failover of a primary-replica setup, and Redis Cluster for horizontal sharding across multiple nodes.

Replication

# In replica's redis.conf (or via CLI)
REPLICAOF 192.168.1.10 6379

# Check replication status
INFO replication
# role: slave
# master_host: 192.168.1.10
# master_link_status: up
# master_replid: abc123...
# master_repl_offset: 12345

Redis Sentinel

Sentinel monitors Redis instances and performs automatic failover: it promotes a replica to primary when the primary is unreachable for a configurable duration.

# sentinel.conf
sentinel monitor mymaster 192.168.1.10 6379 2
# "2" = quorum (how many sentinels must agree to trigger failover)

sentinel down-after-milliseconds mymaster 5000
sentinel failover-timeout mymaster 60000
sentinel parallel-syncs mymaster 1

# Start sentinel
redis-server /etc/redis/sentinel.conf --sentinel

from redis.sentinel import Sentinel

# Connect through Sentinel — auto-discovers primary and replicas
sentinel = Sentinel(
    [('sentinel-1', 26379), ('sentinel-2', 26379), ('sentinel-3', 26379)],
    socket_timeout=0.5,
    decode_responses=True,
)

# Gets the current primary (auto-failover transparent to the client)
primary = sentinel.master_for('mymaster', socket_timeout=0.5)
replica = sentinel.slave_for('mymaster', socket_timeout=0.5)

primary.set('key', 'value')          # write to primary
value = replica.get('key')           # read from replica (may lag slightly)

Redis Cluster

Redis Cluster shards data across up to 1,000 nodes using hash slots (0–16383). Each key is mapped to a slot via CRC16(key) % 16384. Each primary node owns a range of slots and has at least one replica.

# Create a cluster from 6 nodes (3 primary + 3 replica)
redis-cli --cluster create \
  127.0.0.1:7000 127.0.0.1:7001 127.0.0.1:7002 \
  127.0.0.1:7003 127.0.0.1:7004 127.0.0.1:7005 \
  --cluster-replicas 1

# Check cluster status
redis-cli -p 7000 CLUSTER INFO
redis-cli -p 7000 CLUSTER NODES

# Which slot does a key hash to?
redis-cli -p 7000 CLUSTER KEYSLOT mykey
# (integer) 14687 — owned by a specific shard

from redis.cluster import RedisCluster

# redis-py handles slot routing transparently
rc = RedisCluster(
    host='127.0.0.1',
    port=7000,
    decode_responses=True,
)

rc.set('user:1001', 'Alice')  # routed to the shard owning this key's slot
rc.set('user:1002', 'Bob')    # may go to a different shard
print(rc.get('user:1001'))    # 'Alice'

Memory Management

Redis lives in RAM — understanding memory usage and eviction policies is critical for keeping costs controlled and preventing OOM crashes.

maxmemory and Eviction Policies

# redis.conf — set a memory cap
maxmemory 2gb

# Eviction policy (what to do when cap is reached)
maxmemory-policy allkeys-lru

# View current memory usage
INFO memory
# used_memory_human: 1.25G
# maxmemory_human: 2.00G
# maxmemory_policy: allkeys-lru

Inspect Memory Usage

# Memory usage of a specific key (in bytes, including overhead)
MEMORY USAGE user:1001
# (integer) 312

# Memory doctor — Redis analyzes and gives advice
MEMORY DOCTOR

# Find large keys (slow on large datasets — use on a replica)
redis-cli --bigkeys

# Memory stats breakdown
MEMORY STATS

# Scan for keys over 10KB using redis-cli
redis-cli --memkeys --memkeys-samples 0

SCAN for Safe Key Enumeration

# NEVER use KEYS in production — it blocks the server for O(N)
# KEYS pattern  -- DO NOT USE on large databases

# SCAN is non-blocking (cursor-based iteration)
# Returns (cursor, [keys...]) — cursor=0 means iteration complete
SCAN 0 MATCH "user:*" COUNT 100
# 1) "1234"     ← new cursor
# 2) 1) "user:1001" 2) "user:1002" ...

SCAN 1234 MATCH "user:*" COUNT 100
# ... continue until cursor returns 0

def scan_keys(pattern: str, count: int = 100) -> list[str]:
    """Safe non-blocking key scan. Never use r.keys() in production."""
    cursor = 0
    all_keys = []
    while True:
        cursor, keys = r.scan(cursor=cursor, match=pattern, count=count)
        all_keys.extend(keys)
        if cursor == 0:
            break
    return all_keys

large_keys = scan_keys('cache:product:*')
print(f"Found {len(large_keys)} product cache keys")

Key Expiration Strategies

• Always set TTLs on cache keys — prevents unbounded memory growth.
• Use shorter TTLs for frequently updated data — reduces stale reads.
• Add jitter to TTLs to prevent cache stampede: ttl = base_ttl + random.randint(0, 30)
• Sliding TTL — reset the TTL on each access to keep hot keys alive.
• Avoid very large keys — values over 10 KB slow serialization and network transfer. Split into smaller keys if possible.

import random

def set_with_jitter(key: str, value: str, base_ttl: int = 300, jitter: int = 60) -> None:
    """Set a key with a jittered TTL to prevent cache stampede."""
    ttl = base_ttl + random.randint(0, jitter)
    r.set(key, value, ex=ttl)

Best Practices

Key Naming Conventions

Use colon-separated namespaces. This makes keys self-documenting, supports glob patterns in SCAN, and aligns with RedisInsight's tree view.

# Pattern: object-type:id:field
user:1001:profile
user:1001:sessions
user:1001:permissions

# Pattern: domain:entity:id
auth:token:abc123def456
cache:product:sku_9876
ratelimit:api:user_1001:2026-02-23

# Pattern: feature:env
feature_flag:dark_mode:prod
feature_flag:dark_mode:staging

# Avoid: flat names, spaces, special chars
user1001profile      # bad — no hierarchy
user 1001            # bad — spaces break CLI parsing

Pipelining for Bulk Operations

A pipeline batches multiple commands into a single network round trip. For N commands, pipelining reduces N RTTs to 1. Essential for bulk writes and multi-step operations that do not require intermediate results.

import redis
import time

r = redis.Redis(host='127.0.0.1', port=6379, decode_responses=True)

# Without pipeline: N round trips
start = time.time()
for i in range(1000):
    r.set(f'key:{i}', f'value:{i}')
print(f"Without pipeline: {time.time() - start:.3f}s")

# With pipeline: 1 round trip (batched)
start = time.time()
with r.pipeline() as pipe:
    for i in range(1000):
        pipe.set(f'key:{i}', f'value:{i}')
    pipe.execute()
print(f"With pipeline: {time.time() - start:.3f}s")
# Typically 10-50x faster on a remote Redis

Connection Pooling

Creating a new TCP connection for each Redis command is expensive. Use a connection pool — redis-py manages one automatically, but configure its size explicitly.

import redis

# Explicit pool configuration (default pool size is 10)
pool = redis.ConnectionPool(
    host='127.0.0.1',
    port=6379,
    db=0,
    max_connections=20,
    decode_responses=True,
)

# Application-wide singleton — do NOT create a new Redis() per request
redis_client = redis.Redis(connection_pool=pool)

# In a FastAPI/Flask app, create the pool at startup and reuse it
# Never create redis.Redis() inside a request handler

Monitoring: INFO and SLOWLOG

# Comprehensive server statistics
INFO all
INFO server       # version, uptime, config
INFO memory       # used_memory, fragmentation ratio
INFO stats        # ops/sec, hits/misses, expired keys
INFO replication  # role, connected replicas
INFO keyspace     # per-DB key count and expiry stats

# Real-time monitoring (1 update/sec)
redis-cli --stat

# Slow query log (commands taking over N microseconds)
# redis.conf: slowlog-log-slower-than 10000  (10ms)
SLOWLOG GET 10          # last 10 slow commands
SLOWLOG RESET           # clear the log
SLOWLOG LEN             # how many entries

import redis

r = redis.Redis(host='127.0.0.1', port=6379, decode_responses=True)

# Check cache hit ratio (should be > 90% for effective caching)
info = r.info('stats')
hits = int(info.get('keyspace_hits', 0))
misses = int(info.get('keyspace_misses', 0))
total = hits + misses
if total > 0:
    hit_rate = hits / total * 100
    print(f"Cache hit rate: {hit_rate:.1f}% ({hits}/{total})")

# Check memory fragmentation ratio
memory_info = r.info('memory')
frag_ratio = float(memory_info.get('mem_fragmentation_ratio', 1.0))
if frag_ratio > 1.5:
    print(f"WARNING: High memory fragmentation ratio: {frag_ratio:.2f}")
    print("Consider restarting Redis or running MEMORY PURGE")

Production Checklist

• Set maxmemory and an eviction policy appropriate for your use case. Never run production Redis without a memory cap.
• Use SCAN, never KEYS in application code or scripts that run against production.
• Pipeline bulk operations — any loop that does N Redis calls is a candidate for pipelining.
• Always set TTLs on cache keys — unbounded key growth is a common cause of OOM crashes.
• Use connection pooling — one pool per process, sized to your concurrency needs.
• Enable persistence — even for caches, RDB snapshots allow faster warm-up after restarts and reduce DB pressure.
• Monitor hit rate and memory — add keyspace_hits and used_memory to your metrics dashboard.
• Separate functional concerns by database number — cache in db 0, sessions in db 1, rate limits in db 2. Or use separate Redis instances.
• Secure with requirepass and bind to 127.0.0.1 or a private network. Never expose Redis to the public internet.
• Test failover — simulate a primary crash and verify your Sentinel or Cluster promotion works end-to-end before going live.

# Memory
maxmemory 4gb
maxmemory-policy allkeys-lru

# Persistence (hybrid recommended)
save 900 1
save 300 10
save 60 10000
appendonly yes
appendfsync everysec
aof-use-rdb-preamble yes

# Security
requirepass your_strong_password_here
bind 127.0.0.1

# Performance
tcp-keepalive 300
hz 10
lazyfree-lazy-eviction yes    # async eviction (Redis 4+)
lazyfree-lazy-expire yes

# Slow log
slowlog-log-slower-than 10000  # 10ms
slowlog-max-len 128

# Disable dangerous commands in production
rename-command FLUSHDB ""
rename-command FLUSHALL ""
rename-command DEBUG ""
rename-command CONFIG "CONFIG_b3f2c9"  # rename to a secret name