Redis Caching

Redis Caching Skill — Data Structures, Caching Patterns, and Production Best Practices

Overview

This skill provides comprehensive guidance for implementing Redis caching strategies, including all Redis data structures (strings, hashes, lists, sets, sorted sets, streams, bitmaps, hyperloglogs, geospatial), caching patterns (cache-aside, write-through, write-behind, write-around), session management, pub/sub messaging, Lua scripting, transactions, pipelines, clustering, Sentinel, replication, persistence (RDB, AOF), memory management, and production best practices across multiple client libraries.

When to Use

Use this skill when:

• Implementing cache-aside, write-through, write-behind, or write-around caching patterns
• Managing Redis sessions for web applications
• Using Redis pub/sub for real-time messaging
• Working with Redis Streams for message queuing
• Performing string operations (GET, SET, INCR, DECR, APPEND)
• Managing hash operations (HSET, HGET, HGETALL, HDEL)
• Using list operations (LPUSH, RPUSH, LPOP, RPOP, LRANGE)
• Implementing set operations (SADD, SREM, SMEMBERS, SINTER, SUNION)
• Working with sorted sets (ZADD, ZREM, ZRANGE, ZREVRANGE, ZINCRBY)
• Implementing stream processing (XADD, XREAD, XGROUP, XACK)
• Using bitmap operations (SETBIT, GETBIT, BITCOUNT)
• Implementing hyperloglog counting (PFADD, PFCOUNT, PFMERGE)
• Using geospatial operations (GEOADD, GEORADIUS, GEOHASH)
• Writing Lua scripts for atomic operations
• Implementing transactions with MULTI/EXEC
• Using pipelines for batch operations
• Setting up Redis clustering for horizontal scaling
• Configuring Redis Sentinel for high availability
• Implementing Redis replication for read scaling
• Configuring RDB persistence
• Setting up AOF persistence
• Managing memory and maxmemory configuration
• Implementing eviction policies (LRU, LFU, random, ttl)
• Managing TTL expiration strategies
• Setting up connection pooling with connection managers
• Using Redisson, Lettuce, Jedis, ioredis, node-redis, or redis-py clients

Do NOT use this skill when:

• Building a relational database schema (use database-design skill)
• Implementing application business logic (use backend-developer skill)
• Setting up message queues like RabbitMQ or Kafka (use microservices-architecture skill)
• Configuring full-text search (use elasticsearch or meilisearch)
• Implementing object storage (use cloud storage solutions)
• Setting up graph databases (use Neo4j or Amazon Neptune)
• Building time-series databases (use InfluxDB or TimescaleDB)

Why avoid: Redis is an in-memory data store optimized for caching, sessions, and real-time operations. It is not a replacement for relational databases, search engines, or object storage. Use the right tool for the job.

Core Concepts

Redis Data Structures

| Structure | Use Case | Time Complexity | Memory Efficiency | | --------------- | -------------------------------------- | --------------- | -------------------------- | | String | Counter, cache, rate limiting | O(1) | High | | Hash | Object storage, user profiles | O(1) | High (sparse optimization) | | List | Queue, stack, feed | O(1) push/pop | Medium | | Set | Tags, unique members, intersections | O(1) add/remove | High | | Sorted Set | Leaderboards, rankings, priority queue | O(log N) | Medium | | Stream | Message queue, event sourcing | O(1) add | Low (append-only) | | Bitmap | Feature flags, presence tracking | O(1) | Very High | | HyperLogLog | Unique visitor counting | O(1) | Very High (12KB per key) | | Geospatial | Location-based queries | O(log N) | Medium |

Caching Patterns

| Pattern | Write Strategy | Read Strategy | Best For | | ----------------- | ---------------------------------------- | ---------------------------------------------- | -------------------------- | | Cache-Aside | App writes to DB, invalidates cache | App reads cache first, falls back to DB | Most common, simple | | Write-Through | App writes to cache and DB synchronously | App reads from cache | Data consistency critical | | Write-Behind | App writes to cache, async write to DB | App reads from cache | Write performance critical | | Write-Around | App writes directly to DB | App reads cache, falls back to DB, writes back | Reduce cache pollution |

Persistence Options

| Option | Mechanism | Durability | Performance | Recovery Time | | ------------- | ----------------------- | ----------------------------------------- | ----------- | --------------- | | RDB | Point-in-time snapshots | Lower (can lose data since last snapshot) | Higher | Faster | | AOF | Append-only command log | Higher (can configure fsync frequency) | Lower | Slower (replay) | | RDB + AOF | Both mechanisms | Highest | Medium | Medium |

Client Libraries by Language

| Language | Recommended Client | Alternative | Features | | ----------- | ------------------ | --------------- | ------------------------------ | | Node.js | ioredis | node-redis | Clustering, Sentinel, Lua | | Python | redis-py | redis.asyncio | Pipelines, Pub/Sub, Streams | | Java | Lettuce | Jedis, Redisson | Reactive, clustering, Sentinel | | Go | go-redis | redigo | Pipelines, pub/sub, scripting | | Ruby | redis-rb | redis-namespace | Connection pooling, scripting |

Caching Pattern Implementations

Cache-Aside Pattern (Lazy Loading)

The most common caching pattern. Application is responsible for loading data into cache.

# Python (redis-py) - Cache-Aside Pattern
import redis
import json

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

def get_user(user_id: int) -> dict:
    cache_key = f"user:{user_id}"

    # 1. Try cache first
    cached = redis_client.get(cache_key)
    if cached:
        return json.loads(cached)

    # 2. Cache miss - load from database
    user = database.query("SELECT * FROM users WHERE id = %s", (user_id,))
    if not user:
        return None

    # 3. Store in cache with TTL
    redis_client.setex(cache_key, 3600, json.dumps(user))  # 1 hour TTL

    return user

def update_user(user_id: int, user_data: dict):
    # 1. Update database first
    database.execute(
        "UPDATE users SET name = %s, email = %s WHERE id = %s",
        (user_data['name'], user_data['email'], user_id)
    )

    # 2. Invalidate cache
    redis_client.delete(f"user:{user_id}")

// Node.js (ioredis) - Cache-Aside Pattern
const Redis = require('ioredis');
const redis = new Redis({ host: 'localhost', port: 6379 });

async function getProduct(productId) {
  const cacheKey = `product:${productId}`;

  // Try cache first
  let cached = await redis.get(cacheKey);
  if (cached) {
    return JSON.parse(cached);
  }

  // Cache miss - load from database
  const product = await db.products.findById(productId);
  if (!product) {
    return null;
  }

  // Store in cache with TTL
  await redis.setex(cacheKey, 3600, JSON.stringify(product));

  return product;
}

async function updateProduct(productId, productData) {
  await db.products.update(productId, productData);
  await redis.del(`product:${productId}`); // Invalidate cache
}

Write-Through Pattern

Application writes to both cache and database synchronously.

# Python - Write-Through Pattern
def save_user(user_id: int, user_data: dict):
    cache_key = f"user:{user_id}"

    # Write to both cache and database
    serialized = json.dumps(user_data)

    # 1. Write to cache
    redis_client.setex(cache_key, 3600, serialized)

    # 2. Write to database (synchronous)
    database.execute(
        "INSERT INTO users (id, name, email) VALUES (%s, %s, %s) "
        "ON CONFLICT (id) DO UPDATE SET name = %s, email = %s",
        (user_id, user_data['name'], user_data['email'],
         user_data['name'], user_data['email'])
    )

// Node.js - Write-Through Pattern
async function saveProduct(productId, productData) {
  const cacheKey = `product:${productId}`;
  const serialized = JSON.stringify(productData);

  // Write to both cache and database
  await redis.setex(cacheKey, 3600, serialized);
  await db.products.upsert(productId, productData);
}

Write-Behind Pattern (Write-Back)

Application writes to cache only, database is updated asynchronously.

# Python - Write-Behind Pattern (using background thread)
import threading
import queue

write_queue = queue.Queue()

def background_writer():
    """Background thread that batches writes to database"""
    batch = []
    while True:
        try:
            item = write_queue.get(timeout=1)
            batch.append(item)

            # Flush batch every 100 items or every second
            if len(batch) >= 100:
                database.batch_update(batch)
                batch.clear()
        except queue.Empty:
            if batch:
                database.batch_update(batch)
                batch.clear()

# Start background writer thread
threading.Thread(target=background_writer, daemon=True).start()

def update_user_async(user_id: int, user_data: dict):
    cache_key = f"user:{user_id}"

    # 1. Write to cache immediately
    redis_client.setex(cache_key, 3600, json.dumps(user_data))

    # 2. Queue for async database write
    write_queue.put((user_id, user_data))

Write-Around Pattern

Application writes directly to database, cache is populated on read.

# Python - Write-Around Pattern
def update_user(user_id: int, user_data: dict):
    # Write directly to database, bypass cache
    database.execute(
        "UPDATE users SET name = %s, email = %s WHERE id = %s",
        (user_data['name'], user_data['email'], user_id)
    )

    # Delete from cache if it exists (will be re-populated on next read)
    redis_client.delete(f"user:{user_id}")

def get_user(user_id: int) -> dict:
    cache_key = f"user:{user_id}"

    cached = redis_client.get(cache_key)
    if cached:
        return json.loads(cached)

    # Load from database and populate cache
    user = database.query("SELECT * FROM users WHERE id = %s", (user_id,))
    if user:
        redis_client.setex(cache_key, 3600, json.dumps(user))

    return user

Redis Data Structure Operations

String Operations

# Basic string operations
redis_client.set("key", "value")              # SET key value
redis_client.get("key")                       # GET key -> "value"
redis_client.setex("key", 3600, "value")      # SETEX key 3600 value (with TTL)
redis_client.setnx("lock:key", "locked")      # SETNX (set if not exists)
redis_client.append("key", " suffix")         # APPEND key " suffix"
redis_client.mset({"k1": "v1", "k2": "v2"})  # MSET k1 v1 k2 v2
redis_client.mget(["k1", "k2"])              # MGET k1 k2 -> ["v1", "v2"]

# Atomic counters
redis_client.incr("page:views")               # INCR -> 1
redis_client.incrby("page:views", 10)         # INCRBY 10 -> 11
redis_client.decr("page:views")               # DECR -> 10
redis_client.incrbyfloat("price", 2.5)        # INCRBYFLOAT 2.5

Hash Operations

# Hash operations - ideal for object storage
redis_client.hset("user:1001", mapping={      # HSET multiple fields
    "name": "John Doe",
    "email": "john@example.com",
    "age": "30"
})
redis_client.hget("user:1001", "name")        # HGET -> "John Doe"
redis_client.hgetall("user:1001")             # HGETALL -> {name, email, age}
redis_client.hdel("user:1001", "age")         # HDEL field
redis_client.hincrby("user:1001", "age", 1)   # HINCRBY field 1
redis_client.hexists("user:1001", "email")    # HEXISTS -> True
redis_client.hlen("user:1001")                # HLEN -> 2
redis_client.hkeys("user:1001")               # HKEYS -> [name, email]

List Operations

# List operations - queue and stack patterns
redis_client.lpush("queue:tasks", "task1")    # LPUSH -> 1
redis_client.rpush("queue:tasks", "task2")    # RPUSH -> 2
redis_client.lpop("queue:tasks")              # LPOP -> "task1" (FIFO queue)
redis_client.rpop("queue:tasks")              # RPOP -> "task2" (LIFO stack)
redis_client.lrange("queue:tasks", 0, -1)     # LRANGE 0 -1 -> all elements
redis_client.llen("queue:tasks")              # LLEN -> length
redis_client.lindex("queue:tasks", 0)         # LINDEX 0 -> first element

# Blocking pop (for consumer queues)
redis_client.blpop("queue:tasks", timeout=5)  # BLPOP with 5s timeout
redis_client.brpop("queue:tasks", timeout=5)  # BRPOP with 5s timeout

Set Operations

# Set operations - unique members and set algebra
redis_client.sadd("post:1:tags", "python", "redis", "caching")  # SADD
redis_client.srem("post:1:tags", "caching")                      # SREM
redis_client.smembers("post:1:tags")                             # SMEMBERS -> {python, redis}
redis_client.sismember("post:1:tags", "python")                  # SISMEMBER -> True
redis_client.scard("post:1:tags")                                # SCARD -> 2

# Set algebra
redis_client.sadd("user:1:interests", "tech", "sports", "music")
redis_client.sadd("user:2:interests", "tech", "gaming", "music")
redis_client.sinter("user:1:interests", "user:2:interests")      # SINTER -> {tech, music}
redis_client.sunion("user:1:interests", "user:2:interests")      # SUNION -> {tech, sports, music, gaming}
redis_client.sdiff("user:1:interests", "user:2:interests")       # SDIFF -> {sports}

Sorted Set Operations

# Sorted set operations - leaderboards and rankings
redis_client.zadd("leaderboard", {           # ZADD multiple members
    "player1": 1500,
    "player2": 2300,
    "player3": 1800
})
redis_client.zincrby("leaderboard", 100, "player1")  # ZINCRBY -> 1600
redis_client.zrem("leaderboard", "player3")           # ZREM member

# Range queries
redis_client.zrange("leaderboard", 0, -1)             # ZRANGE (ascending)
redis_client.zrevrange("leaderboard", 0, -1)          # ZREVRANGE (descending)
redis_client.zrange("leaderboard", 0, 9, withscores=True)  # Top 10 with scores

# Rankings
redis_client.zrank("leaderboard", "player1")          # ZRANK (ascending)
redis_client.zrevrank("leaderboard", "player1")       # ZREVRANK (descending)
redis_client.zscore("leaderboard", "player1")         # ZSCORE -> 1600.0
redis_client.zcard("leaderboard")                     # ZCARD -> count

Stream Operations

# Stream operations - message queuing
redis_client.xadd("orders:stream", {"order_id": "123", "amount": "99.99"})
# -> "1697012345678-0"

redis_client.xadd("orders:stream", {"order_id": "124", "amount": "149.99"})
# -> "1697012345679-0"

# Read stream
redis_client.xread({"orders:stream": "0-0"}, count=10)  # Read from beginning
redis_client.xread({"orders:stream": "$"}, count=10, block=5000)  # Block for new entries

# Consumer groups
redis_client.xgroup_create("orders:stream", "workers", id="0-0", mkstream=True)
redis_client.xreadgroup("workers", "worker-1", {"orders:stream": ">-"}, count=1)
redis_client.xack("orders:stream", "workers", ["1697012345678-0"])

# Stream info
redis_client.xlen("orders:stream")  # Stream length

Bitmap Operations

# Bitmap operati

…

## Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [Omar-Obando](https://github.com/Omar-Obando)
- **Source:** [Omar-Obando/qwen-orchestrator](https://github.com/Omar-Obando/qwen-orchestrator)
- **License:** MIT
- **Homepage:** https://qwen.ai/qwencode

Install and usage instructions live in the source repository linked above.