api-database-redis

Redis 模式

快速指南： 将 Redis 用作内存数据存储，用于缓存、会话管理、速率限制、发布/订阅消息传递和作业队列。主要使用 ioredis（v5.x）作为客户端，因为它具有出色的 TypeScript 支持、Cluster/Sentinel 集成、自动流水线和 Lua 脚本功能。只有在需要 Redis Stack 模块（JSON、Search、TimeSeries）时才使用 node-redis（v5.x）。对于 BullMQ 工作进程，始终设置 maxRetriesPerRequest: null；为发布/订阅订阅者使用独立连接；对原子多步操作，通过 defineCommand 定义 Lua 脚本。

---

<critical_requirements>

重要：使用此技能前

所有代码必须遵循 CLAUDE.md 中的项目约定（kebab-case、命名导出、导入顺序、import type、命名常量）

（必须为发布/订阅订阅者使用独立的 Redis 连接——已订阅的连接会进入特殊模式，无法执行其他命令）

（传递给 BullMQ 的任何 ioredis 连接都必须设置 maxRetriesPerRequest: null——BullMQ 要求无限重试，未设置时会抛出错误）

（任何需要在多个 Redis 命令间保证原子性的操作，都必须使用 Lua 脚本（defineCommand 或 eval）——即使使用流水线，分开的命令也不具备原子性）

（必须处理每个 Redis 客户端实例的 error 事件——未处理的错误会导致 Node.js 进程崩溃）

</critical_requirements>

---

示例

• 核心模式 -- ioredis/node-redis 连接、错误处理、重新连接、集群、哨兵、流水线、事务
• 缓存模式 -- Cache-aside、写穿透、失效、缓存雪崩防护、多键流水线
• 数据结构 -- 使用类型化辅助工具处理字符串、哈希、列表、集合、有序集合
• 会话 -- Express connect-redis（v9+ 需要 node-redis）、Hono 手动中间件
• 发布/订阅 -- 发布/订阅、事件广播、模式订阅
• 速率限制 -- 滑动窗口（Lua）、令牌桶（Lua）、中间件集成
• 队列与锁 -- BullMQ 作业队列、带消费者组的 Redis Streams、分布式锁

其他资源：

• reference.md -- 命令速查表、连接选项、反模式、生产环境检查清单

---

自动检测： Redis、ioredis、node-redis、createClient、RedisStore、BullMQ、Queue、Worker、发布/订阅、MULTI、EXEC、pipeline、Lua 脚本、defineCommand、xadd、xread、cache-aside、速率限制、会话存储、connect-redis、Redis.Cluster、Sentinel

适用场景：

• 缓存数据库查询或 API 响应（cache-aside、写穿透）
• 为 Express/Hono/Fastify 应用存储会话
• 分布式速率限制（滑动窗口、令牌桶）
• 使用发布/订阅进行实时消息传递
• 使用 BullMQ 队列处理后台作业
• 使用有序集合构建排行榜、计数器和实时分析
• 使用 Lua 脚本实现分布式锁和原子操作

涵盖的关键模式：

• ioredis 连接的设置、配置和错误处理
• 数据结构（字符串、哈希、列表、集合、有序集合、流）
• 使用 TTL 管理实现 Cache-aside 和写穿透缓存
• 使用 connect-redis 存储会话
• 使用 Lua 脚本实现速率限制（滑动窗口、令牌桶）
• 使用独立连接进行发布/订阅消息传递
• 使用 Redis Streams 构建持久化消息队列
• 使用 BullMQ 构建带重试和调度功能的作业队列
• 流水线和事务（MULTI/EXEC）
• 使用 Lua 脚本实现原子操作
• 使用 Cluster 模式和 Sentinel 实现高可用性

不适用场景：

• 关系型数据的主数据库（请使用关系型数据库）
• 需要复杂查询的文档存储（请使用文档数据库）
• 大型二进制文件存储（请使用 S3/对象存储）
• 未配置持久化但必须在内存完全丢失后仍保留的数据

---

## 理念

Redis 是一种内存数据存储，可用作缓存、消息代理和流处理引擎。核心原则：使用 Redis 存储快速、临时或实时数据，而不是将其用作主数据库。

核心原则：

1. 缓存，而非存储——Redis 用于辅助主数据库。缓存经常访问的数据，但始终要在其他地方保留真实数据源。
2. 原子操作——对于涉及多个键的操作，请使用 Lua 脚本或 MULTI/EXEC。单条 Redis 命令是原子的，但命令序列不是。
3. 职责分离——使用不同的 Redis 数据库（或键前缀）分别处理缓存、会话和队列。Pub/Sub 使用独立连接。
4. 为所有内容设置 TTL——内存有限。每个缓存键都应设置过期时间。在 SET 命令中使用 EX（秒）或 PX（毫秒）。
5. 优雅地处理故障——Redis 是缓存，不是数据库。如果 Redis 宕机，应用程序应优雅降级（绕过缓存，直接使用数据库）。

---

核心模式

模式 1：ioredis 连接设置

为 ioredis 配置适当的错误处理和重连策略。完整示例（包括 node-redis 和集群配置）请参阅 examples/core.md。

// ✅ Good Example - Proper ioredis setup with error handling
import Redis from "ioredis";

const RETRY_DELAY_BASE_MS = 50;
const RETRY_DELAY_MAX_MS = 2000;

function createRedisClient(): Redis {
  const url = process.env.REDIS_URL;
  if (!url) {
    throw new Error("REDIS_URL environment variable is required");
  }

  const client = new Redis(url, {
    maxRetriesPerRequest: 3,
    retryStrategy(times) {
      return Math.min(times * RETRY_DELAY_BASE_MS, RETRY_DELAY_MAX_MS);
    },
    lazyConnect: true,
  });

  client.on("error", (err) => {
    console.error("Redis connection error:", err.message);
  });

  return client;
}

export { createRedisClient };

优点： 验证环境变量，使用命名常量表示重试延迟，lazyConnect 可避免连接未就绪时建立连接，错误事件处理程序可防止进程崩溃

// ❌ Bad Example - No error handling, hardcoded config
import Redis from "ioredis";
const redis = new Redis("redis://localhost:6379");
// No error handler, no retry strategy, hardcoded URL

缺点： 缺少错误事件处理程序会导致 Node.js 进程在连接失败时崩溃，硬编码 URL 会妨碍针对不同环境进行配置

---

模式 2：旁路缓存

先检查缓存，未命中时回退到数据库，并填充缓存。有关写穿、失效、缓存击穿预防和多键缓存模式，请参阅 examples/caching.md。

// ✅ Good Example - Generic cache-aside helper
const CACHE_TTL_SECONDS = 300;

async function cacheAside<T>(
  redis: Redis,
  key: string,
  fetcher: () => Promise<T>,
  ttlSeconds: number = CACHE_TTL_SECONDS,
): Promise<T> {
  const cached = await redis.get(key);
  if (cached !== null) {
    return JSON.parse(cached) as T;
  }

  const data = await fetcher();

  // Fire-and-forget cache write
  redis.set(key, JSON.stringify(data), "EX", ttlSeconds).catch((err) => {
    console.error(`Cache write failed for ${key}:`, err.message);
  });

  return data;
}

优点： 通用的 cacheAside<T> 适用于任何数据类型，后台写入缓存可避免缓存故障阻塞响应，TTL 可配置

---

模式 3：流水线与事务

批量执行命令以提高网络效率（流水线）或保证原子性（MULTI/EXEC）。完整示例请参阅 examples/core.md。

// ✅ Pipeline - batch for network efficiency (NOT atomic)
const pipeline = redis.pipeline();
pipeline.hset(key, { name: user.name, email: user.email });
pipeline.expire(key, TTL_SECONDS);
await pipeline.exec();

// ✅ Transaction - atomic execution with optimistic locking
await redis.watch(fromKey);
const results = await redis
  .multi()
  .decrby(fromKey, amount)
  .incrby(toKey, amount)
  .exec();
if (!results) {
  /* WATCH detected change, retry */
}

优点： 流水线可减少往返次数，MULTI/EXEC 可保证原子性，WATCH 可实现乐观锁

// ❌ Bad Example - Non-atomic balance transfer
await redis.decrby("balance:user1", 100);
await redis.incrby("balance:user2", 100);
// Crash between commands causes data loss

缺点： 两条独立命令不具备原子性，两条命令之间发生崩溃会导致数据不一致

---

模式 4：使用 Lua 脚本保证原子性

Lua 脚本在 Redis 服务端以原子方式执行。有关基于 Lua 的速率限制器完整示例，请参阅 examples/rate-limiting.md。

// ✅ Define custom atomic command
redis.defineCommand("rateLimit", {
  numberOfKeys: 1,
  lua: `
    local key = KEYS[1]
    local limit = tonumber(ARGV[1])
    local window = tonumber(ARGV[2])
    local now = tonumber(ARGV[3])
    redis.call('ZREMRANGEBYSCORE', key, 0, now - window)
    local count = redis.call('ZCARD', key)
    if count < limit then
      redis.call('ZADD', key, now, now .. '-' .. math.random(1000000))
      redis.call('EXPIRE', key, window)
      return 1
    end
    return 0
  `,
});

优点： defineCommand 内部使用 EVALSHA，性能更高；Lua 脚本是原子的，不会发生竞态条件

---

模式 5：Pub/Sub 消息传递

订阅和发布必须使用独立连接。有关事件广播系统，请参阅 examples/pub-sub.md。

// ✅ Good Example - Separate connections
const publisher = new Redis(url);
const subscriber = new Redis(url); // MUST be separate

await subscriber.subscribe("notifications");
subscriber.on("message", (channel, message) => {
  handleNotification(JSON.parse(message));
});

await publisher.publish("notifications", JSON.stringify(data));

优点： 发布和订阅使用独立连接（Redis 协议要求），两者都配置了错误处理程序

// ❌ Bad Example - Same connection
const redis = new Redis();
await redis.subscribe("channel");
await redis.set("key", "value"); // THROWS: connection is in subscriber mode

缺点： 已订阅的连接无法执行非 Pub/Sub 命令

---

模式 6：BullMQ 作业队列

支持重试、调度和优先级的健壮作业队列。有关包含 Worker、事件和优雅关闭的完整示例，请参阅 examples/queues.md。

// ✅ Good Example - BullMQ connection factory
function createBullMQConnection(): Redis {
  return new Redis(process.env.REDIS_URL!, {
    maxRetriesPerRequest: null, // REQUIRED for BullMQ
  });
}

const emailQueue = new Queue<EmailJobData>(QUEUE_NAME, {
  connection: createBullMQConnection(),
  defaultJobOptions: {
    attempts: MAX_RETRY_ATTEMPTS,
    backoff: { type: "exponential", delay: BACKOFF_DELAY_MS },
  },
});

优点： BullMQ 要求设置 maxRetriesPerRequest: null，每个 Queue/Worker 使用独立连接

// ❌ Bad Example - Missing required config
const connection = new Redis(); // Missing maxRetriesPerRequest: null
const queue = new Queue("emails", { connection });
// BullMQ will throw: "maxRetriesPerRequest must be null"

缺点： BullMQ 要求无限重试；如果不设置为 null，ioredis 会在达到设定的尝试次数后放弃

---

模式 7：Redis Streams

使用消费者组实现持久、有序的消息日志。有关包含待处理消息恢复功能的完整生产者/消费者实现，请参阅 examples/queues.md。

// Producer
const entryId = await redis.xadd(
  STREAM_KEY,
  "*",
  "orderId",
  id,
  "action",
  action,
);

// Consumer group setup (idempotent)
try {
  await redis.xgroup("CREATE", STREAM_KEY, GROUP_NAME, "0", "MKSTREAM");
} catch (err) {
  if (!(err instanceof Error) || !err.message.includes("BUSYGROUP")) throw err;
}

// Consumer read + acknowledge
const results = await redis.xreadgroup(
  "GROUP",
  GROUP_NAME,
  consumerName,
  "COUNT",
  "10",
  "BLOCK",
  "5000",
  "STREAMS",
  STREAM_KEY,
  ">",
);
await redis.xack(STREAM_KEY, GROUP_NAME, messageId);

优点： MKSTREAM 会在流不存在时创建流，BUSYGROUP 处理可实现幂等设置，XACK 用于确认处理完成

---

## 性能优化

• 自动流水线——启用 enableAutoPipelining: true，将同一事件循环周期内发出的命令批量处理。此功能不适用于 WATCH/MULTI 或阻塞命令。设置方法请参见 examples/core.md。
• 手动流水线——使用 redis.pipeline() 显式批量处理命令。请参见 examples/core.md。
• 键过期时间——为所有缓存键设置 TTL。使用具名常量（TTL_SHORT_SECONDS = 60、TTL_MEDIUM_SECONDS = 300 等）。请参见 examples/core.md。
• 使用 SCAN 而非 KEYS——生产环境中切勿使用 KEYS（会阻塞 Redis）。请使用 redis.scanStream({ match: "pattern", count: 100 })。
• 使用 UNLINK 而非 DEL——删除大型键时使用 unlink（非阻塞删除）。

---

<decision_framework>

决策框架

选择哪个 Redis 客户端？

Which Redis client should I use?
├─ Need Redis Stack modules (JSON, Search, TimeSeries)? -> node-redis (v5.x)
├─ Using BullMQ for job queues? -> ioredis (BullMQ requires it)
├─ Need Cluster or Sentinel support? -> ioredis (built-in, battle-tested)
├─ Need auto-pipelining? -> ioredis (enableAutoPipelining option)
└─ General caching/sessions/pub-sub? -> ioredis (recommended default)

选择哪种缓存策略？

How should I cache this data?
├─ Read-heavy, tolerates brief staleness? -> Cache-aside with TTL
├─ Needs strong consistency after writes? -> Write-through (update DB + invalidate cache)
├─ Write-heavy, can tolerate brief data loss? -> Write-behind (async cache update)
└─ Data changes rarely? -> Cache-aside with long TTL + manual invalidation

选择哪种数据结构？

What Redis data structure should I use?
├─ Simple key-value (cache, sessions)? -> Strings (GET/SET)
├─ Object with multiple fields? -> Hashes (HSET/HGET)
├─ Ordered ranking/leaderboard? -> Sorted Sets (ZADD/ZRANGE)
├─ Queue (FIFO/LIFO)? -> Lists (LPUSH/RPOP)
├─ Unique collection (tags, categories)? -> Sets (SADD/SMEMBERS)
├─ Persistent message log with consumers? -> Streams (XADD/XREAD)
└─ Rate limiting (sliding window)? -> Sorted Sets + Lua script

选择哪种消息模式？

How should I implement real-time messaging?
├─ Fire-and-forget broadcast? -> Pub/Sub (no persistence)
├─ Need message persistence and replay? -> Streams with consumer groups
├─ Need reliable job processing with retries? -> BullMQ (built on Redis)
└─ Need request-reply pattern? -> Pub/Sub with correlation IDs

原子性决策

Do I need atomicity across multiple commands?
├─ YES -> Are the commands on the same key?
│   ├─ YES -> Use a single atomic command (INCR, SETNX, etc.)
│   └─ NO -> Use Lua script (defineCommand)
├─ NO, but I want batching -> Use pipeline (non-atomic, single round-trip)
└─ Need optimistic locking? -> Use WATCH + MULTI/EXEC

</decision_framework>

---

集成指南

常见集成模式：

• 数据库缓存——使用旁路缓存模式缓存查询结果；写入时使缓存失效
• HTTP 框架会话——会话存储中间件、速率限制中间件
• BullMQ 作业队列——基于 Redis 构建（要求 ioredis 配置 maxRetriesPerRequest: null）
• WebSocket 扩展——使用 Redis 适配器，在多个服务之间分发 WebSocket 连接

替代对象／冲突对象：

• 内存缓存——Redis 可在多个应用实例之间提供分布式缓存
• 基于数据库的会话——Redis 会话速度更快，并能减轻数据库负载
• 简单消息队列——Redis Streams 和 BullMQ 可覆盖大多数队列使用场景；复杂路由或超大吞吐量场景请使用专用消息代理

---

<red_flags>

危险信号

高优先级问题：

• 将同一连接用于 Pub/Sub 订阅和常规命令——已订阅的连接无法执行非 Pub/Sub 命令
• BullMQ 连接缺少 maxRetriesPerRequest: null——如果未设置此项，BullMQ 会立即抛出错误
• 在生产环境中使用 KEYS 命令——扫描所有键时会阻塞整个 Redis 服务
• Redis 客户端没有 error 事件处理程序——未处理的错误会导致 Node.js 进程崩溃
• 在 Redis 中存储大型对象（> 1 MB）——会降低性能并浪费内存；应存储引用，并从对象存储中获取对象

中优先级问题：

• 缓存键缺少 TTL——会导致内存无限增长，直到 Redis 内存耗尽
• 对多个键使用 del 而不是 unlink——del 会阻塞 Redis；unlink 会异步释放内存
• 批量操作未使用流水线——每条命令都会单独进行一次网络往返
• 序列化／反序列化复杂对象时没有错误处理——缓存中的 JSON 格式错误会导致解析崩溃
• BullMQ Queue 和 Worker 共用单个 Redis 连接——二者需要各自的连接

常见错误：

• 误以为流水线命令是原子的——流水线会批量执行命令以提高网络效率，但不保证原子性（请使用 MULTI/EXEC 或 Lua）
• 忘记 hgetall 对不存在的键会返回空对象 {}（而不是 null）——请检查 Object.keys(result).length === 0
• 条件更新时使用 MULTI/EXEC 却未使用 WATCH——除非先对键执行 WATCH，否则事务会无条件执行
• 未处理 GET 返回的 null——缓存未命中时返回的是 null，不是 undefined
• 在生产环境中不使用 TLS 连接 Redis——凭据会以明文形式通过网络传输

注意事项和边缘情况：

• Redis HGETALL 会将所有值作为字符串返回——通过 HSET 存储的数字取出后会变成字符串，需要显式解析
• 使用 SET 覆盖键时，EXPIRE 设置的过期时间会被重置——如果对已有 TTL 的键执行 SET，TTL 会被移除，除非在 SET 命令中包含 EX/PX
• Pub/Sub 消息采用即发即弃模式——发布消息时如果没有订阅者正在监听，消息将永久丢失（如需持久化，请使用 Streams）
• Redis Cluster 不支持跨不同哈希槽的多键操作——使用 {hash-tag} 前缀可将相关键强制放到同一哈希槽
• WATCH 的作用范围是连接级别——并发请求共享连接时会相互干扰各自的 WATCH 状态
• ioredis 自动流水线不适用于 WATCH/MULTI 或阻塞命令（BRPOP、BLPOP、XREAD BLOCK）

</red_flags>

---

<critical_reminders>

关键提醒

所有代码必须遵循项目在 CLAUDE.md 中的约定（kebab-case、命名导出、导入顺序、import type、命名常量）

（必须为 Pub/Sub 订阅者使用单独的 Redis 连接——已订阅的连接会进入特殊模式，无法执行其他命令）

（传递给 BullMQ 的任何 ioredis 连接都必须设置 maxRetriesPerRequest: null——BullMQ 要求无限次重试，未设置时会抛出错误）

（任何需要跨多个 Redis 命令保证原子性的操作都必须使用 Lua 脚本（defineCommand 或 eval）——即使在流水线中，分开的命令也不具备原子性）

（必须处理每个 Redis 客户端实例的 error 事件——未处理的错误会导致 Node.js 进程崩溃）

不遵守这些规则会导致 Pub/Sub 失败、BullMQ 连接错误、竞态条件和应用程序崩溃。

</critical_reminders>