redis-patterns

Redis 模式

为高性能应用实现常见的 Redis 模式。

核心工作流

1. 设置连接：配置 Redis 客户端
2. 选择模式：缓存、会话、队列等
3. 实现操作：使用适当的 TTL 执行 CRUD
4. 处理错误：重新连接、回退方案
5. 监控性能：内存、延迟
6. 优化：流水线、集群

连接设置

// redis/client.ts
import { Redis } from 'ioredis';

// Single instance
export const redis = new Redis({
  host: process.env.REDIS_HOST || 'localhost',
  port: parseInt(process.env.REDIS_PORT || '6379'),
  password: process.env.REDIS_PASSWORD,
  db: parseInt(process.env.REDIS_DB || '0'),

  // Connection options
  maxRetriesPerRequest: 3,
  retryStrategy(times) {
    const delay = Math.min(times * 50, 2000);
    return delay;
  },

  // Performance options
  enableReadyCheck: true,
  enableOfflineQueue: true,
  connectTimeout: 10000,

  // TLS for production
  tls: process.env.NODE_ENV === 'production' ? {} : undefined,
});

// Event handlers
redis.on('connect', () => console.log('Redis connecting...'));
redis.on('ready', () => console.log('Redis ready'));
redis.on('error', (err) => console.error('Redis error:', err));
redis.on('close', () => console.log('Redis connection closed'));

// Cluster connection
export const cluster = new Redis.Cluster([
  { host: 'redis-node-1', port: 6379 },
  { host: 'redis-node-2', port: 6379 },
  { host: 'redis-node-3', port: 6379 },
], {
  redisOptions: {
    password: process.env.REDIS_PASSWORD,
  },
  scaleReads: 'slave',
  maxRedirections: 16,
});

// Graceful shutdown
process.on('SIGTERM', async () => {
  await redis.quit();
});

缓存模式

// patterns/cache.ts
import { redis } from './client';

interface CacheOptions {
  ttl?: number;  // seconds
  prefix?: string;
}

export class Cache {
  private prefix: string;
  private defaultTTL: number;

  constructor(options: CacheOptions = {}) {
    this.prefix = options.prefix || 'cache:';
    this.defaultTTL = options.ttl || 3600;
  }

  private key(key: string): string {
    return `${this.prefix}${key}`;
  }

  async get<T>(key: string): Promise<T | null> {
    const data = await redis.get(this.key(key));
    if (!data) return null;

    try {
      return JSON.parse(data) as T;
    } catch {
      return data as unknown as T;
    }
  }

  async set<T>(key: string, value: T, ttl?: number): Promise<void> {
    const serialized = typeof value === 'string'
      ? value
      : JSON.stringify(value);

    await redis.setex(this.key(key), ttl || this.defaultTTL, serialized);
  }

  async getOrSet<T>(
    key: string,
    fetcher: () => Promise<T>,
    ttl?: number
  ): Promise<T> {
    const cached = await this.get<T>(key);
    if (cached !== null) return cached;

    const value = await fetcher();
    await this.set(key, value, ttl);
    return value;
  }

  async delete(key: string): Promise<void> {
    await redis.del(this.key(key));
  }

  async deletePattern(pattern: string): Promise<void> {
    const keys = await redis.keys(this.key(pattern));
    if (keys.length > 0) {
      await redis.del(...keys);
    }
  }

  // Cache with stale-while-revalidate
  async getStale<T>(
    key: string,
    fetcher: () => Promise<T>,
    options: { ttl: number; staleTTL: number }
  ): Promise<T> {
    const cacheKey = this.key(key);
    const staleKey = `${cacheKey}:stale`;

    const cached = await redis.get(cacheKey);
    if (cached) {
      return JSON.parse(cached);
    }

    // Check stale data
    const stale = await redis.get(staleKey);
    if (stale) {
      // Return stale, refresh in background
      this.refreshCache(key, fetcher, options).catch(console.error);
      return JSON.parse(stale);
    }

    return this.refreshCache(key, fetcher, options);
  }

  private async refreshCache<T>(
    key: string,
    fetcher: () => Promise<T>,
    options: { ttl: number; staleTTL: number }
  ): Promise<T> {
    const value = await fetcher();
    const serialized = JSON.stringify(value);

    const pipeline = redis.pipeline();
    pipeline.setex(this.key(key), options.ttl, serialized);
    pipeline.setex(`${this.key(key)}:stale`, options.staleTTL, serialized);
    await pipeline.exec();

    return value;
  }
}

// Usage
const cache = new Cache({ prefix: 'user:', ttl: 3600 });

async function getUser(id: string) {
  return cache.getOrSet(`profile:${id}`, async () => {
    return await db.users.findById(id);
  }, 1800);
}

会话存储

// patterns/session.ts
import { redis } from './client';
import { nanoid } from 'nanoid';

interface Session {
  id: string;
  userId: string;
  data: Record<string, any>;
  createdAt: number;
  expiresAt: number;
}

export class SessionStore {
  private prefix = 'session:';
  private userPrefix = 'user:sessions:';
  private ttl = 86400 * 7; // 7 days

  private key(sessionId: string): string {
    return `${this.prefix}${sessionId}`;
  }

  async create(userId: string, data: Record<string, any> = {}): Promise<Session> {
    const session: Session = {
      id: nanoid(32),
      userId,
      data,
      createdAt: Date.now(),
      expiresAt: Date.now() + this.ttl * 1000,
    };

    const pipeline = redis.pipeline();

    // Store session
    pipeline.setex(this.key(session.id), this.ttl, JSON.stringify(session));

    // Track user's sessions
    pipeline.sadd(`${this.userPrefix}${userId}`, session.id);
    pipeline.expire(`${this.userPrefix}${userId}`, this.ttl);

    await pipeline.exec();

    return session;
  }

  async get(sessionId: string): Promise<Session | null> {
    const data = await redis.get(this.key(sessionId));
    if (!data) return null;

    const session = JSON.parse(data) as Session;

    // Check expiration
    if (session.expiresAt < Date.now()) {
      await this.destroy(sessionId);
      return null;
    }

    return session;
  }

  async update(sessionId: string, data: Record<string, any>): Promise<void> {
    const session = await this.get(sessionId);
    if (!session) throw new Error('Session not found');

    session.data = { ...session.data, ...data };

    await redis.setex(
      this.key(sessionId),
      this.ttl,
      JSON.stringify(session)
    );
  }

  async refresh(sessionId: string): Promise<void> {
    const session = await this.get(sessionId);
    if (!session) return;

    session.expiresAt = Date.now() + this.ttl * 1000;

    await redis.setex(
      this.key(sessionId),
      this.ttl,
      JSON.stringify(session)
    );
  }

  async destroy(sessionId: string): Promise<void> {
    const session = await this.get(sessionId);
    if (!session) return;

    const pipeline = redis.pipeline();
    pipeline.del(this.key(sessionId));
    pipeline.srem(`${this.userPrefix}${session.userId}`, sessionId);
    await pipeline.exec();
  }

  async destroyAllForUser(userId: string): Promise<void> {
    const sessionIds = await redis.smembers(`${this.userPrefix}${userId}`);

    if (sessionIds.length > 0) {
      const keys = sessionIds.map(id => this.key(id));
      await redis.del(...keys, `${this.userPrefix}${userId}`);
    }
  }
}

速率限制

// patterns/rate-limiter.ts
import { redis } from './client';

interface RateLimitResult {
  allowed: boolean;
  remaining: number;
  resetAt: number;
}

export class RateLimiter {
  // Fixed window rate limiting
  async fixedWindow(
    key: string,
    limit: number,
    windowSeconds: number
  ): Promise<RateLimitResult> {
    const redisKey = `ratelimit:fixed:${key}`;
    const now = Math.floor(Date.now() / 1000);
    const window = Math.floor(now / windowSeconds);
    const windowKey = `${redisKey}:${window}`;

    const count = await redis.incr(windowKey);

    if (count === 1) {
      await redis.expire(windowKey, windowSeconds);
    }

    return {
      allowed: count <= limit,
      remaining: Math.max(0, limit - count),
      resetAt: (window + 1) * windowSeconds * 1000,
    };
  }

  // Sliding window rate limiting
  async slidingWindow(
    key: string,
    limit: number,
    windowSeconds: number
  ): Promise<RateLimitResult> {
    const redisKey = `ratelimit:sliding:${key}`;
    const now = Date.now();
    const windowStart = now - windowSeconds * 1000;

    const pipeline = redis.pipeline();

    // Remove old entries
    pipeline.zremrangebyscore(redisKey, 0, windowStart);

    // Add current request
    pipeline.zadd(redisKey, now, `${now}:${Math.random()}`);

    // Count requests in window
    pipeline.zcard(redisKey);

    // Set expiration
    pipeline.expire(redisKey, windowSeconds);

    const results = await pipeline.exec();
    const count = results?.[2]?.[1] as number;

    return {
      allowed: count <= limit,
      remaining: Math.max(0, limit - count),
      resetAt: now + windowSeconds * 1000,
    };
  }

  // Token bucket rate limiting
  async tokenBucket(
    key: string,
    bucketSize: number,
    refillRate: number,  // tokens per second
    tokensNeeded: number = 1
  ): Promise<RateLimitResult> {
    const redisKey = `ratelimit:bucket:${key}`;
    const now = Date.now();

    // Lua script for atomic token bucket
    const script = `
      local key = KEYS[1]
      local bucket_size = tonumber(ARGV[1])
      local refill_rate = tonumber(ARGV[2])
      local tokens_needed = tonumber(ARGV[3])
      local now = tonumber(ARGV[4])

      local bucket = redis.call('HMGET', key, 'tokens', 'last_refill')
      local tokens = tonumber(bucket[1]) or bucket_size
      local last_refill = tonumber(bucket[2]) or now

      -- Calculate refill
      local elapsed = (now - last_refill) / 1000
      local refill = elapsed * refill_rate
      tokens = math.min(bucket_size, tokens + refill)

      -- Check if enough tokens
      local allowed = 0
      if tokens >= tokens_needed then
        tokens = tokens - tokens_needed
        allowed = 1
      end

      -- Save state
      redis.call('HMSET', key, 'tokens', tokens, 'last_refill', now)
      redis.call('EXPIRE', key, math.ceil(bucket_size / refill_rate) + 1)

      return {allowed, tokens}
    `;

    const result = await redis.eval(
      script,
      1,
      redisKey,
      bucketSize,
      refillRate,
      tokensNeeded,
      now
    ) as [number, number];

    return {
      allowed: result[0] === 1,
      remaining: Math.floor(result[1]),
      resetAt: now + Math.ceil((tokensNeeded - result[1]) / refillRate) * 1000,
    };
  }
}

// Express middleware
export function rateLimitMiddleware(
  limiter: RateLimiter,
  options: { limit: number; window: number }
) {
  return async (req: Request, res: Response, next: NextFunction) => {
    const key = req.ip || 'anonymous';
    const result = await limiter.slidingWindow(key, options.limit, options.window);

    res.setHeader('X-RateLimit-Limit', options.limit);
    res.setHeader('X-RateLimit-Remaining', result.remaining);
    res.setHeader('X-RateLimit-Reset', result.resetAt);

    if (!result.allowed) {
      return res.status(429).json({
        error: 'Too Many Requests',
        retryAfter: Math.ceil((result.resetAt - Date.now()) / 1000),
      });
    }

    next();
  };
}

分布式锁

// patterns/lock.ts
import { redis } from './client';
import { nanoid } from 'nanoid';

export class DistributedLock {
  private prefix = 'lock:';

  async acquire(
    resource: string,
    ttlMs: number = 10000
  ): Promise<string | null> {
    const lockKey = `${this.prefix}${resource}`;
    const lockValue = nanoid();
    const ttlSeconds = Math.ceil(ttlMs / 1000);

    const acquired = await redis.set(
      lockKey,
      lockValue,
      'EX',
      ttlSeconds,
      'NX'
    );

    return acquired === 'OK' ? lockValue : null;
  }

  async release(resource: string, lockValue: string): Promise<boolean> {
    const lockKey = `${this.prefix}${resource}`;

    // Lua script for atomic release
    const script = `
      if redis.call('GET', KEYS[1]) == ARGV[1] then
        return redis.call('DEL', KEYS[1])
      else
        return 0
      end
    `;

    const result = await redis.eval(script, 1, lockKey, lockValue);
    return result === 1;
  }

  async extend(
    resource: string,
    lockValue: string,
    ttlMs: number
  ): Promise<boolean> {
    const lockKey = `${this.prefix}${resource}`;
    const ttlSeconds = Math.ceil(ttlMs / 1000);

    const script = `
      if redis.call('GET', KEYS[1]) == ARGV[1] then
        return redis.call('EXPIRE', KEYS[1], ARGV[2])
      else
        return 0
      end
    `;

    const result = await redis.eval(script, 1, lockKey, lockValue, ttlSeconds);
    return result === 1;
  }

  async withLock<T>(
    resource: string,
    fn: () => Promise<T>,
    options: { ttl?: number; retries?: number; retryDelay?: number } = {}
  ): Promise<T> {
    const { ttl = 10000, retries = 3, retryDelay = 100 } = options;

    let lockValue: string | null = null;
    let attempts = 0;

    while (attempts < retries) {
      lockValue = await this.acquire(resource, ttl);
      if (lockValue) break;

      attempts++;
      await new Promise(resolve => setTimeout(resolve, retryDelay));
    }

    if (!lockValue) {
      throw new Error(`Failed to acquire lock for ${resource}`);
    }

    try {
      return await fn();
    } finally {
      await this.release(resource, lockValue);
    }
  }
}

// Usage
const lock = new DistributedLock();

async function processPayment(orderId: string) {
  return lock.withLock(`order:${orderId}`, async () => {
    // Critical section - only one process can execute
    const order = await db.orders.findById(orderId);
    if (order.status !== 'pending') {
      throw new Error('Order already processed');
    }

    await processStripePayment(order);
    await db.orders.update(orderId, { status: 'paid' });
  });
}

发布/订阅模式

// patterns/pubsub.ts
import { Redis } from 'ioredis';

// Separate connections for pub/sub
const subscriber = new Redis(process.env.REDIS_URL!);
const publisher = new Redis(process.env.REDIS_URL!);

type MessageHandler<T> = (message: T, channel: string) => void | Promise<void>;

export class PubSub {
  private handlers: Map<string, Set<MessageHandler<any>>> = new Map();

  async subscribe<T>(channel: string, handler: MessageHandler<T>): Promise<void> {
    if (!this.handlers.has(channel)) {
      this.handlers.set(channel, new Set());
      await subscriber.subscribe(channel);
    }

    this.handlers.get(channel)!.add(handler);
  }

  async unsubscribe(channel: string, handler?: MessageHandler<any>): Promise<void> {
    const handlers = this.handlers.get(channel);
    if (!handlers) return;

    if (handler) {
      handlers.delete(handler);
      if (handlers.size === 0) {
        this.handlers.delete(channel);
        await subscriber.unsubscribe(channel);
      }
    } else {
      this.handlers.delete(channel);
      await subscriber.unsubscribe(channel);
    }
  }

  async publish<T>(channel: string, message: T): Promise<void> {
    await publisher.publish(channel, JSON.stringify(message));
  }

  async subscribePattern<T>(pattern: string, handler: MessageHandler<T>): Promise<void> {
    if (!this.handlers.has(pattern)) {
      this.handlers.set(pattern, new Set());
      await subscriber.psubscribe(pattern);
    }

    this.handlers.get(pattern)!.add(handler);
  }
}

// Initialize message handling
subscriber.on('message', (channel, message) => {
  const pubsub = new PubSub();
  const handlers = pubsub['handlers'].get(channel);
  if (!handlers) return;

  const parsed = JSON.parse(message);
  handlers.forEach(handler => handler(parsed, channel));
});

// Usage
const pubsub = new PubSub();

// Subscribe to user events
await pubsub.subscribe<{ userId: string; action: string }>('user:events', async (msg) => {
  console.log(`User ${msg.userId} performed ${msg.action}`);
});

// Publish event
await pubsub.publish('user:events', { userId: '123', action: 'login' });

排行榜模式

// patterns/leaderboard.ts
import { redis } from './client';

export class Leaderboard {
  private key: string;

  constructor(name: string) {
    this.key = `leaderboard:${name}`;
  }

  async addScore(member: string, score: number): Promise<void> {
    await redis.zadd(this.key, score, member);
  }

  async incrementScore(member: string, increment: number): Promise<number> {
    return redis.zincrby(this.key, increment, member);
  }

  async getTop(count: number): Promise<Array<{ member: string; score: number }>> {
    const results = await redis.zrevrange(this.key, 0, count - 1, 'WITHSCORES');

    const entries: Array<{ member: string; score: number }> = [];
    for (let i = 0; i < results.length; i += 2) {
      entries.push({
        member: results[i],
        score: parseFloat(results[i + 1]),
      });
    }

    return entries;
  }

  async getRank(member: string): Promise<number | null> {
    const rank = await redis.zrevrank(this.key, member);
    return rank !== null ? rank + 1 : null;
  }

  async getScore(member: string): Promise<number | null> {
    const score = await redis.zscore(this.key, member);
    return score !== null ? parseFloat(score) : null;
  }

  async getAroundMember(
    member: string,
    count: number
  ): Promise<Array<{ member: string; score: number; rank: number }>> {
    const rank = await redis.zrevrank(this.key, member);
    if (rank === null) return [];

    const start = Math.max(0, rank - Math.floor(count / 2));
    const results = await redis.zrevrange(this.key, start, start + count - 1, 'WITHSCORES');

    const entries: Array<{ member: string; score: number; rank: number }> = [];
    for (let i = 0; i < results.length; i += 2) {
      entries.push({
        member: results[i],
        score: parseFloat(results[i + 1]),
        rank: start + i / 2 + 1,
      });
    }

    return entries;
  }
}

最佳实践

1. 连接池：重复使用连接
2. 流水线：批量执行多条命令
3. 为所有键设置 TTL：防止内存泄漏
4. 键命名：使用一致的前缀
5. Lua 脚本：执行原子操作
6. 支持集群：面向水平扩展进行设计
7. 错误处理：优雅降级
8. 内存管理：监控并设置 maxmemory

输出检查清单

每个 Redis 实现都应包含：

• 采用重试策略的连接
• 适当的键前缀/命名空间
• 为所有键设置 TTL
• 错误处理和回退方案
• 优雅关闭
• 批量操作使用流水线
• 原子操作使用 Lua 脚本
• 内存监控
• 兼容集群的操作
• 连接池