Redis Caching Patterns: Practical Use Cases Guide
In this tutorial, you'll learn about Redis Caching Patterns: Practical Use Cases Guide. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.
Redis is an in-memory data structure store that serves as a cache, Message Broker, and database for use cases including session management, real-time analytics, Rate Limiting, distributed locking, and publish/subscribe messaging.
What You'll Learn
You will implement cache-aside, write-through, and write-behind caching patterns, use Redis for sessions, Rate Limiting, distributed locks, leaderboards, and Message Queues, and understand eviction policies and persistence trade-offs.
Why Redis Caching Matters
Database queries are 100-1000x slower than in-memory lookups. Doda Browser uses Redis to cache user session data and frequently accessed bookmarks, reducing page load times from 450ms to 12ms and handling 50,000 requests per second on a single instance.
Redis Caching Learning Path
flowchart LR A[Redis Basics] --> B[Caching Patterns] B --> C[Redis Use Cases] B:::current classDef current fill:#f90,color:#fff,stroke:#333,stroke-width:2px
Prerequisites: Familiarity with Redis data structures and basic commands. Understanding of MySQL or PostgreSQL is helpful.
Cache-Aside Pattern (Lazy Loading)
The application checks Redis first. On a miss, it reads from the database and populates the cache.
import redis
import json
cache = redis.Redis(host='localhost', port=6379, decode_responses=True)
def get_user(user_id):
# Try cache first
cached = cache.get(f"user:{user_id}")
if cached:
return json.loads(cached)
# Cache miss: load from database
user = db.fetch_one(
"SELECT id, name, email FROM users WHERE id = %s",
(user_id,)
)
if user:
# Populate cache with TTL
cache.setex(
f"user:{user_id}",
3600, # 1 hour TTL
json.dumps(user)
)
return user
Pros: Resistant to failures (database remains source of truth). Cons: Cache stampede on first request (three database queries for three concurrent misses).
Solving Cache Stampede
import threading
def get_user_with_mutex(user_id):
cache_key = f"user:{user_id}"
cached = cache.get(cache_key)
if cached:
return json.loads(cached)
# Try to acquire lock (only one process loads from DB)
lock_key = f"lock:user:{user_id}"
if cache.setnx(lock_key, "locked"):
cache.expire(lock_key, 10) # Auto-release after 10s
try:
user = db.fetch_one("SELECT * FROM users WHERE id = %s", (user_id,))
if user:
cache.setex(cache_key, 3600, json.dumps(user))
return user
finally:
cache.delete(lock_key)
else:
# Another process is loading, wait and retry
import time
time.sleep(0.1)
return get_user_with_mutex(user_id)
Write-Through Cache
Every write goes to cache first, which then writes to the database synchronously.
def update_user(user_id, data):
cache_key = f"user:{user_id}"
# Write to cache first
cache.setex(cache_key, 3600, json.dumps(data))
# Then write to database
db.execute(
"UPDATE users SET name = %s, email = %s WHERE id = %s",
(data['name'], data['email'], user_id)
)
Pros: Cache always consistent with database. Cons: Higher write latency, cache and database must both succeed or transaction fails.
Write-Behind (Write-Back) Cache
Writes go to cache and are asynchronously persisted to the database.
import queue
import threading
write_queue = queue.Queue()
def update_user_write_behind(user_id, data):
cache_key = f"user:{user_id}"
cache.setex(cache_key, 3600, json.dumps(data))
write_queue.put((user_id, data))
def batch_writer():
while True:
batch = []
for _ in range(100):
try:
batch.append(write_queue.get(timeout=5))
except queue.Empty:
break
if batch:
with db.transaction():
for user_id, data in batch:
db.execute(
"UPDATE users SET name = %s, email = %s WHERE id = %s",
(data['name'], data['email'], user_id)
)
# Start background writer
threading.Thread(target=batch_writer, daemon=True).start()
Pros: Very fast writes, batch persistence. Cons: Data loss risk on crash (unsaved writes in queue).
Session Storage
Redis with TTL is ideal for user sessions.
import uuid
def create_session(user_id):
session_id = str(uuid.uuid4())
session_data = {
'user_id': user_id,
'created_at': time.time(),
'ip_address': request.remote_addr,
'user_agent': request.user_agent
}
# Store session with 30-minute TTL (sliding expiration)
cache.setex(f"session:{session_id}", 1800, json.dumps(session_data))
return session_id
def get_session(session_id):
data = cache.get(f"session:{session_id}")
if data:
session = json.loads(data)
# Slide expiration on each access
cache.expire(f"session:{session_id}", 1800)
return session
return None
def delete_session(session_id):
cache.delete(f"session:{session_id}")
Rate Limiting
Redis is ideal for Rate Limiting due to its atomic increment operations.
Sliding Window with Sorted Sets
def check_rate_limit(user_id, limit=100, window_seconds=60):
key = f"ratelimit:{user_id}"
now = time.time()
window_start = now - window_seconds
# Remove old entries outside window
cache.zremrangebyscore(key, 0, window_start)
# Count current requests
request_count = cache.zcard(key)
if request_count >= limit:
return False # Rate limited
# Add current request
cache.zadd(key, {str(now): now})
cache.expire(key, window_seconds)
return True # Request allowed
Fixed Window with INCR
def check_rate_limit_fixed(user_id, limit=100, window_seconds=60):
key = f"ratelimit:{user_id}:{int(time.time() / window_seconds)}"
count = cache.incr(key)
if count == 1:
cache.expire(key, window_seconds + 1)
return count <= limit
Performance comparison:
| Method | Memory per Request | Accuracy | Redis Commands |
|---|---|---|---|
| Fixed window INCR | 8 bytes | Moderate | 1-2 |
| Sliding Window Sorted Set | 50 bytes | Perfect | 3-4 |
| Token bucket | 16 bytes | Good | 2-3 |
Distributed Locks
Redis SETNX (or SET with NX) provides distributed locking across services.
def acquire_lock(lock_name, acquire_timeout=10, lock_timeout=30):
lock_key = f"lock:{lock_name}"
identifier = str(uuid.uuid4())
end = time.time() + acquire_timeout
while time.time() < end:
if cache.set(lock_key, identifier, nx=True, ex=lock_timeout):
return identifier # Lock acquired
time.sleep(0.01) # Wait 10ms before retry
return None # Lock acquisition timed out
def release_lock(lock_name, identifier):
lock_key = f"lock:{lock_name}"
# Only release if we still own the lock (prevent releasing someone else's)
if cache.get(lock_key) == identifier:
cache.delete(lock_key)
# Usage
lock_id = acquire_lock("payment:order:12345")
if lock_id:
try:
process_payment(12345)
finally:
release_lock("payment:order:12345", lock_id)
Leaderboards and Counting
Redis sorted sets are perfect for real-time leaderboards.
def update_score(game_id, player_id, score):
key = f"leaderboard:{game_id}"
cache.zadd(key, {player_id: score})
def get_top_players(game_id, count=10):
key = f"leaderboard:{game_id}"
# Get top players with scores, highest first
return cache.zrevrange(key, 0, count - 1, withscores=True)
def get_player_rank(game_id, player_id):
key = f"leaderboard:{game_id}"
rank = cache.zrevrank(key, player_id)
if rank is not None:
return rank + 1 # Ranks are 0-indexed in Redis
return None
# Example
update_score("game1", "alice", 1500)
update_score("game1", "bob", 1200)
update_score("game1", "charlie", 1800)
print(get_top_players("game1", 2))
# Output: [('charlie', 1800.0), ('alice', 1500.0)]
Message Queue (List-based)
Redis lists can serve as simple Message Queues.
# Producer
def send_email_task(to, subject, body):
task = json.dumps({
'to': to,
'subject': subject,
'body': body,
'created_at': time.time()
})
cache.lpush("email_queue", task)
# Consumer (worker)
def process_email_queue():
while True:
_, task_data = cache.brpop("email_queue", timeout=5)
if task_data:
task = json.loads(task_data)
send_email(task['to'], task['subject'], task['body'])
Eviction Policies
| Policy | Behavior | Use Case |
|---|---|---|
| noeviction | Return error on writes when full | Never lose data |
| allkeys-lru | Evict least recently used keys | General caching |
| allkeys-lfu | Evict least frequently used keys | Hot data caching |
| volatile-lru | Evict LRU among keys with TTL | Session storage |
| volatile-ttl | Evict keys with shortest TTL | Short-lived data |
# redis.conf
maxmemory 4gb
maxmemory-policy allkeys-lru
Common Redis Caching Errors
1. No TTL on Cache Keys
Keys without TTL stay in memory forever, eventually filling Redis and triggering eviction of other keys.
2. Cache Stampede Without Mitigation
When a popular cache key expires, multiple concurrent requests all hit the database. Use mutex locks or early recomputation.
3. Storing Large Objects
Redis works best with values under 100KB. Large objects consume memory and increase network latency. Consider compression or splitting.
4. Using Redis as the Primary Database
Redis is an in-memory store with limited persistence guarantees. Use it as a cache on top of a durable database.
5. Ignoring Network Latency
Every Redis command has network overhead. Use pipelining or Lua scripts for multiple commands.
# BAD: N+1 network round trips
for user_id in user_ids:
cache.get(f"user:{user_id}")
# GOOD: Single round trip with pipeline
pipe = cache.pipeline()
for user_id in user_ids:
pipe.get(f"user:{user_id}")
results = pipe.execute()
6. No Monitoring for Eviction Rate
If Redis is evicting keys, you need more memory or a better eviction policy. Monitor evicted_keys metric.
7. Using KEYS Command in Production
KEYS blocks the entire Redis instance. Use SCAN for production.
# BAD: Blocks Redis for seconds
cache.keys("user:*")
# GOOD: Non-blocking iteration
cursor = 0
while cursor != 0:
cursor, keys = cache.scan(cursor, "user:*", count=100)
for key in keys:
process(key)
Practice Questions
1. What is the cache-aside pattern and when should you use it?
The application checks cache first, loads from database on miss, and populates the cache. Best for read-heavy workloads with acceptable first-request latency.
2. How does Redis compare to Memcached?
Redis supports rich data structures (lists, sets, sorted sets, hashes) and persistence. Memcached is simpler (only strings) but uses less memory overhead per key.
3. What eviction policy is best for general caching?
allkeys-lru. It evicts the least recently used keys when memory is full, which matches most caching workloads.
4. How do you implement a distributed lock in Redis?
Use SET key value NX EX timeout to atomically create a key only if it does not exist. Only the process that created the key can delete it.
5. Challenge: Design a caching strategy for a news website.
The site has 1M articles, 10K new articles/day, 50M daily page views. Most traffic goes to 5% of articles. Answer: (1) Cache articles with allkeys-lfu eviction to keep hot articles in memory. (2) TTL of 1 hour for articles, 5 minutes for homepage. (3) Use cache-aside with mutex to prevent stampede for breaking news. (4) Pre-warm popular articles on deploy. (5) Use pipeline for batch article fetches. (6) Monitor eviction rate and increase memory if evicted_keys > 0.
FAQ
Try It Yourself
Set up a Redis caching layer:
- Start Redis locally or with Docker
- Implement cache-aside for a slow database query (simulate with time.sleep)
- Add a TTL of 60 seconds
- Measure response time: first request (cache miss) vs subsequent requests (cache hit)
- Implement Rate Limiting for an API endpoint
- Create a real-time leaderboard with sorted sets
- Monitor memory usage and eviction rate
What's Next
You have learned Redis caching patterns, Rate Limiting, distributed locks, leaderboards, and Message Queues. Start by adding cache-aside to your slowest database queries and measuring the latency improvement.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro