Caching, Pub/Sub & Patterns Caching Patterns // Cache-aside (lazy loading) — most common async function getUser(id) { const cacheKey = `user:${id}`; const cache…
Caching, Pub/Sub & Patterns
Caching Patterns
// Cache-aside (lazy loading) — most common
async function getUser(id) {
const cacheKey = `user:${id}`;
const cached = await redis.get(cacheKey);
if (cached) return JSON.parse(cached);
const user = await db.findUser(id);
await redis.set(cacheKey, JSON.stringify(user), 'EX', 300); // 5 min TTL
return user;
}
// Write-through — update cache on every write
async function updateUser(id, data) {
const user = await db.updateUser(id, data);
await redis.set(`user:${id}`, JSON.stringify(user), 'EX', 300);
return user;
}
// Cache invalidation on write
async function deleteUser(id) {
await db.deleteUser(id);
await redis.del(`user:${id}`);
}
// Hash-based caching for partial updates
async function updateUserField(id, field, value) {
await db.updateUser(id, { [field]: value });
await redis.hset(`user:${id}`, field, value);
}
// Multi-key invalidation pattern (tag-based)
await redis.sadd(`tag:user:${userId}:keys`, `posts:${userId}`, `activity:${userId}`);
// On user update, get all related keys and delete them
const keys = await redis.smembers(`tag:user:${userId}:keys`);
if (keys.length) await redis.del(...keys);
Rate Limiting
// Fixed window rate limiter
async function checkRateLimit(ip, limit = 100, windowSec = 60) {
const key = `rate:${ip}:${Math.floor(Date.now() / (windowSec * 1000))}`;
const count = await redis.incr(key);
if (count === 1) await redis.expire(key, windowSec);
return count <= limit;
}
// Sliding window with sorted set (more accurate)
async function slidingWindowLimit(ip, limit = 100, windowMs = 60000) {
const key = `rate:sliding:${ip}`;
const now = Date.now();
const windowStart = now - windowMs;
const requestId = `${now}-${Math.random()}`;
const pipe = redis.pipeline();
pipe.zadd(key, now, requestId);
pipe.zremrangebyscore(key, 0, windowStart); // remove old requests
pipe.zcard(key);
pipe.expire(key, Math.ceil(windowMs / 1000));
const results = await pipe.exec();
const count = results[2][1];
return count <= limit;
}
Pub/Sub
// Publisher (ioredis)
import Redis from 'ioredis';
const pub = new Redis();
await pub.publish('notifications', JSON.stringify({
type: 'NEW_MESSAGE',
userId: 42,
text: 'Hello!',
}));
// Subscriber (separate connection required!)
const sub = new Redis();
sub.subscribe('notifications', 'alerts');
sub.on('message', (channel, message) => {
const data = JSON.parse(message);
console.log(`[channel:${channel}]`, data);
});
// Pattern subscribe
sub.psubscribe('user:*:events');
sub.on('pmessage', (pattern, channel, message) => {
console.log(pattern, channel, message);
});
// Note: sub connection can ONLY subscribe — cannot run other commands
// Use a separate connection for regular commands
Distributed Lock
// Simple distributed lock with SET NX
async function acquireLock(resource, ttlMs = 5000) {
const lockKey = `lock:${resource}`;
const lockValue = crypto.randomUUID(); // unique ID for this lock holder
const result = await redis.set(lockKey, lockValue, 'PX', ttlMs, 'NX');
return result === 'OK' ? lockValue : null;
}
async function releaseLock(resource, lockValue) {
// Atomic check-and-delete using Lua to prevent releasing someone else's lock
const script = `
if redis.call("get", KEYS[1]) == ARGV[1] then
return redis.call("del", KEYS[1])
else
return 0
end
`;
return redis.eval(script, 1, `lock:${resource}`, lockValue);
}
// Usage
const lockValue = await acquireLock('send-email', 10000);
if (!lockValue) {
throw new Error('Could not acquire lock');
}
try {
await sendEmail();
} finally {
await releaseLock('send-email', lockValue);
}
Transactions & Pipelining
// Pipeline — batch commands (no atomicity guarantee, but fewer round trips)
const pipe = redis.pipeline();
pipe.set('k1', 'v1');
pipe.get('k1');
pipe.incr('counter');
const results = await pipe.exec(); // [[null, 'OK'], [null, 'v1'], [null, 1]]
// Transaction (MULTI/EXEC — atomic, all or nothing)
const results = await redis
.multi()
.set('balance:alice', 900)
.set('balance:bob', 1100)
.exec(); // either both succeed or both fail
// WATCH — optimistic locking
async function transfer(from, to, amount) {
await redis.watch(`balance:${from}`);
const fromBalance = parseInt(await redis.get(`balance:${from}`));
if (fromBalance < amount) {
await redis.unwatch();
throw new Error('Insufficient funds');
}
const result = await redis
.multi()
.decrby(`balance:${from}`, amount)
.incrby(`balance:${to}`, amount)
.exec(); // returns null if WATCH detected a change
if (!result) throw new Error('Transaction aborted (concurrent modification)');
}