Redis 進階應用：Pub/Sub、Lua Script、分散式鎖
不只是快取的完整應用指南

String: SET key value / GET key / SETEX key 60 value（TTL 60秒）
        → 基本 Key-Value、計數器（INCR/DECR）、快取

Hash:   HSET user:1 name "Joseph" age 28
        HGET user:1 name          → "Joseph"
        HGETALL user:1            → 所有欄位
        → 物件儲存，比 JSON String 更節省記憶體（有 ziplist 優化）

List:   LPUSH queue task1 task2   → 從左側推入
        RPOP queue                 → 從右側取出（Queue 行為）
        LRANGE queue 0 -1          → 取得所有元素
        → 工作佇列、最新動態（Timeline）

Set:    SADD tags:post:1 redis backend database
        SMEMBERS tags:post:1       → 所有標籤
        SINTER tags:post:1 tags:post:2  → 兩篇文章的共同標籤
        → 去重、交集/聯集/差集運算

ZSet:   ZADD leaderboard 1500 "Joseph" 1200 "Alice"
(Sorted ZREVRANGE leaderboard 0 9 WITHSCORES   → 前10名（含分數）
  Set)  ZREVRANK leaderboard "Joseph"           → Joseph 的排名
        → 排行榜、帶權重的任務佇列、延遲佇列

import Redis from 'ioredis';

const publisher = new Redis();
const subscriber = new Redis(); // Pub/Sub 需要獨立連線！
// 一旦連線進入 subscribe 模式，就不能再發送其他命令
// 因此 publisher 和 subscriber 必須是兩個獨立的 Redis 連線

// ─── 訂閱頻道 ───────────────────────────────────────────
await subscriber.subscribe('notifications', 'chat:general');

subscriber.on('message', (channel, message) => {
  const data = JSON.parse(message);
  console.log(`[${channel}] Received:`, data);

  // 根據頻道處理不同邏輯
  if (channel === 'notifications') {
    sendPushNotification(data);
  } else if (channel === 'chat:general') {
    broadcastToWebSocket(data);
  }
});

// ─── 發布訊息 ───────────────────────────────────────────
await publisher.publish('notifications', JSON.stringify({
  type: 'NEW_ORDER',
  orderId: '12345',
  userId: 'user:001',
  timestamp: Date.now(),
}));

// 回傳值：成功收到訊息的訂閱者數量（0 = 沒有人在線）
// → Pub/Sub 的 Fire and Forget 特性：沒人訂閱就直接丟棄

// ─── Pattern 訂閱（訂閱所有 chat: 開頭的頻道）───────────
await subscriber.psubscribe('chat:*');

subscriber.on('pmessage', (pattern, channel, message) => {
  console.log(`Pattern "${pattern}" matched "${channel}":`, message);
  // pattern  = "chat:*"
  // channel  = "chat:general" 或 "chat:room-1" 等實際頻道名
  // message  = 訊息內容
});

情境：限量商品庫存 = 1，兩台 API Server 同時收到購買請求

Server 1: 讀取庫存 = 1，判斷「有庫存，可以購買」
Server 2: 讀取庫存 = 1，判斷「有庫存，可以購買」
Server 1: 執行扣庫存 → 庫存 = 0（成功建立訂單）
Server 2: 執行扣庫存 → 庫存 = -1（超賣了！）

這就是「讀-改-寫」（Read-Modify-Write）的 Race Condition。
在多台 Server 的環境中，這是必然會發生的問題，不是偶發。

解法一：資料庫樂觀鎖（有 Retry 成本）
解法二：資料庫 SELECT FOR UPDATE（高併發時會鎖住大量請求）
解法三：Redis 分散式鎖（高效、低延遲、跨服務）← 今天的主角

import Redlock from 'redlock';
import Redis from 'ioredis';

const redis = new Redis({ host: 'localhost', port: 6379 });

const redlock = new Redlock(
  [redis],  // 生產環境建議傳入奇數個 Redis 實例（如 3 或 5），提高可靠性
  {
    retryCount: 3,    // 獲取鎖失敗後最多重試 3 次
    retryDelay: 200,  // 每次重試間隔 200ms
    retryJitter: 200, // 加入隨機抖動（0~200ms），防止多個請求同時重試產生「驚群效應」
    // 驚群效應：所有等待的請求在同一時刻重試，瞬間產生大量競爭
  },
);

async function purchaseProduct(productId: string, userId: string) {
  const lockKey = `lock:product:${productId}`;
  const lockTTL = 10_000; // 鎖的最長持有時間：10 秒（防止死鎖）

  // 嘗試獲取鎖（會等待重試，直到成功或超過 retryCount）
  const lock = await redlock.acquire([lockKey], lockTTL);

  try {
    // ─── 臨界區（Critical Section）─────────────────────────
    // 這段程式碼在任何時間點，整個系統中只有一個執行緒在跑

    const stock = await getProductStock(productId);

    if (stock <= 0) {
      throw new Error('Out of stock');
    }

    await decreaseStock(productId, 1);
    const order = await createOrder({ productId, userId, quantity: 1 });

    console.log(`Order ${order.id} created for user ${userId}`);
    return order;
    // ─── 臨界區結束 ─────────────────────────────────────────

  } finally {
    // 確保鎖一定被釋放（即使上面拋出錯誤）
    // finally 區塊保證無論成功或失敗都會執行
    await lock.release();
  }
}

// ─── 自行實作（理解原理）────────────────────────────────────
// Redis 的 SET NX EX 是原子操作，這是分散式鎖的核心
async function acquireLock(key: string, ttlSeconds: number): Promise<string | null> {
  // 使用隨機 token 作為鎖的值（用於確認只有鎖的持有者才能釋放）
  const token = `${Date.now()}-${Math.random()}`;

  // SET key token NX EX ttl
  // NX：Only Set if Not eXists（只有 key 不存在時才設定）
  // EX：設定過期時間（秒），防止程序崩潰後鎖永遠不釋放（死鎖）
  const result = await redis.set(key, token, 'EX', ttlSeconds, 'NX');

  // result 為 'OK' 表示成功獲取鎖，null 表示鎖已被其他人持有
  return result === 'OK' ? token : null;
}

async function releaseLock(key: string, token: string): Promise<boolean> {
  // 必須先確認 token 是自己的，才能刪除
  // 這段必須是原子操作，否則「確認」和「刪除」之間可能有 Race Condition
  // → 用 Lua Script 保證原子性（見 Section 4）
  const luaScript = `
    if redis.call('GET', KEYS[1]) == ARGV[1] then
      return redis.call('DEL', KEYS[1])
    else
      return 0
    end
  `;

  const result = await redis.eval(luaScript, 1, key, token) as number;
  return result === 1;
}

-- scripts/deduct-stock.lua
-- KEYS[1] = 庫存的 Key（如 "stock:product:123"）
-- ARGV[1] = 要扣除的數量
-- ARGV[2] = 操作日誌的識別碼（userId + timestamp）

local stock = tonumber(redis.call('GET', KEYS[1]))

-- 判斷商品是否存在
if stock == nil then
  return -1  -- 商品不存在
end

-- 判斷庫存是否足夠
if stock < tonumber(ARGV[1]) then
  return -2  -- 庫存不足（回傳負數讓應用層判斷錯誤類型）
end

-- 扣減庫存
local new_stock = stock - tonumber(ARGV[1])
redis.call('SET', KEYS[1], new_stock)

-- 寫入操作日誌（可以之後用 LRANGE 查詢所有扣減記錄）
redis.call('LPUSH', 'deduct-log', ARGV[2])

-- 如果庫存歸零，發布 sold-out 事件
if new_stock == 0 then
  redis.call('PUBLISH', 'inventory:sold-out', KEYS[1])
end

return new_stock  -- 回傳扣減後的庫存數量

-- 整個 Script 是原子的：
-- 不可能發生「讀到庫存 = 1，另一個請求也讀到 1，然後兩個都扣」的 Race Condition

import fs from 'fs';
import Redis from 'ioredis';

const redis = new Redis();

// ─── 載入 Script（SCRIPT LOAD）────────────────────────────
// 使用 EVALSHA 而不是 EVAL 的原因：
//   EVAL：每次都傳送完整的 Script 內容（浪費頻寬）
//   EVALSHA：只傳 SHA1 Hash，Redis 從 Script Cache 取出執行（省頻寬）

const script = fs.readFileSync('./scripts/deduct-stock.lua', 'utf8');
const sha = await redis.script('LOAD', script);
// sha 像是 "a42059b356c875f0717db19a51f6aaca9ae659ea"

// ─── 執行 Lua Script ────────────────────────────────────
async function deductStock(
  productId: string,
  quantity: number,
  userId: string,
): Promise<number> {
  const result = await redis.evalsha(
    sha,                                   // Script 的 SHA1 Hash
    1,                                     // KEYS 的數量（下面的參數分為 KEYS 和 ARGV）
    `stock:${productId}`,                // KEYS[1]
    quantity.toString(),                    // ARGV[1]
    `${userId}:${Date.now()}`,          // ARGV[2]
  ) as number;

  // 解析 Lua Script 的回傳值
  switch (result) {
    case -1:
      throw new Error(`Product ${productId} not found`);
    case -2:
      throw new Error(`Insufficient stock for product ${productId}`);
    default:
      console.log(`Stock deducted. Remaining: ${result}`);
      return result;
  }
}

// ─── 錯誤處理：NOSCRIPT（Script Cache 被清除）─────────────
async function safeDeductStock(
  productId: string,
  quantity: number,
  userId: string,
): Promise<number> {
  try {
    return await deductStock(productId, quantity, userId);
  } catch (err: unknown) {
    // 如果是 NOSCRIPT 錯誤（Redis 重啟後 Script Cache 清空），重新 LOAD
    if (err instanceof Error && err.message.includes('NOSCRIPT')) {
      const script = fs.readFileSync('./scripts/deduct-stock.lua', 'utf8');
      sha = await redis.script('LOAD', script);
      return await deductStock(productId, quantity, userId);
    }
    throw err;
  }
}

const LEADERBOARD_KEY = 'game:leaderboard';

// ─── 設定分數（絕對值）────────────────────────────────────
async function setScore(userId: string, score: number) {
  // ZADD：如果成員已存在，更新分數；不存在則新增
  await redis.zadd(LEADERBOARD_KEY, score, userId);
}

// ─── 加分（相對值，原子操作）──────────────────────────────
async function addScore(userId: string, delta: number): Promise<number> {
  // ZINCRBY 是原子的：不需要先 GET 再 SET，天然防止 Race Condition
  const newScore = await redis.zincrby(LEADERBOARD_KEY, delta, userId);
  return parseFloat(newScore);
}

// ─── 取得前 N 名 ─────────────────────────────────────────
async function getTopPlayers(count = 10) {
  // ZREVRANGE：按分數從高到低排序（Rev = Reverse = 降序）
  // WITHSCORES：同時回傳分數
  const raw = await redis.zrevrange(LEADERBOARD_KEY, 0, count - 1, 'WITHSCORES');

  // raw = ['userId1', '1500', 'userId2', '1200', ...] 交錯排列
  const leaderboard = [];
  for (let i = 0; i < raw.length; i += 2) {
    leaderboard.push({
      rank: i / 2 + 1,          // 1-indexed 排名
      userId: raw[i],
      score: parseFloat(raw[i + 1]),
    });
  }

  return leaderboard;
}

// ─── 取得特定用戶的排名和分數 ─────────────────────────────
async function getUserRank(userId: string) {
  // ZREVRANK：從高到低的排名（0-indexed），不存在回傳 null
  const [rank, score] = await Promise.all([
    redis.zrevrank(LEADERBOARD_KEY, userId),
    redis.zscore(LEADERBOARD_KEY, userId),
  ]);

  return {
    rank: rank !== null ? rank + 1 : null,  // 轉為 1-indexed
    score: score ? parseFloat(score) : 0,
    inLeaderboard: rank !== null,
  };
}

// ─── 分頁取得排行榜 ───────────────────────────────────────
async function getLeaderboardPage(page: number, pageSize = 20) {
  const start = (page - 1) * pageSize;
  const end = start + pageSize - 1;

  const [raw, total] = await Promise.all([
    redis.zrevrange(LEADERBOARD_KEY, start, end, 'WITHSCORES'),
    redis.zcard(LEADERBOARD_KEY),   // 總成員數
  ]);

  const players = [];
  for (let i = 0; i < raw.length; i += 2) {
    players.push({
      rank: start + i / 2 + 1,
      userId: raw[i],
      score: parseFloat(raw[i + 1]),
    });
  }

  return {
    players,
    total,
    page,
    pageSize,
    totalPages: Math.ceil(total / pageSize),
  };
}

// ─── 取得某分數範圍的玩家（按分數篩選）──────────────────────
async function getPlayersByScoreRange(minScore: number, maxScore: number) {
  // ZRANGEBYSCORE：按分數範圍查詢（從低到高）
  // 用 ZREVRANGEBYSCORE 則是從高到低
  const raw = await redis.zrangebyscore(
    LEADERBOARD_KEY,
    minScore,
    maxScore,
    'WITHSCORES',
  );

  const players = [];
  for (let i = 0; i < raw.length; i += 2) {
    players.push({ userId: raw[i], score: parseFloat(raw[i + 1]) });
  }

  return players;
}

──────────────────────────────────────────────────────────
問題 1：記憶體超過限制
──────────────────────────────────────────────────────────

症狀：OOM（Out of Memory），Redis 拒絕寫入新資料
解法：設定 maxmemory 和淘汰策略（eviction policy）

  # redis.conf
  maxmemory 2gb                    # 限制最大記憶體使用量
  maxmemory-policy allkeys-lru     # 最常用的淘汰策略

  淘汰策略選擇指南：
  allkeys-lru    : 所有 Key 中，淘汰最近最少使用的（最常用）
  volatile-lru   : 只淘汰有 TTL 的 Key（適合只把部分資料當快取的場景）
  allkeys-random : 隨機淘汰（不推薦）
  noeviction     : 記憶體滿了就報錯（預設值，適合不能丟資料的場景）

──────────────────────────────────────────────────────────
問題 2：熱 Key（Hot Key）— 單一 Key 請求量過大
──────────────────────────────────────────────────────────

症狀：某個 Key 承受了大量請求，導致 Redis 該實例 CPU 100%
     （例如：首頁推薦資料、爆紅文章的點贊數）

解法一：本地快取（Local Cache）
  在應用層加一層 LRU Cache（如 node-lru-cache）
  把超熱的 Key 快取在 Server 記憶體中（TTL 1~5 秒）
  這樣大部分請求根本不打到 Redis

解法二：Key 加隨機後綴（複製多份）
  原本 "product:hot:1" 承受所有流量
  改為 "product:hot:1:0" / "product:hot:1:1" / "product:hot:1:2"
  寫入時更新所有副本，讀取時隨機選一個
  → 流量分散到多個 Key，避免集中

──────────────────────────────────────────────────────────
問題 3：BigKey — 單一 Value 過大
──────────────────────────────────────────────────────────

症狀：讀取或刪除某個 Key 時，Redis 阻塞數秒
     （Redis 單執行緒，一個大操作會卡住所有其他請求）

典型 BigKey：
  String > 10MB、List/Hash/Set/ZSet 成員數超過 10,000

解法：
  ✓ 壓縮 Value（gzip 後再存）
  ✓ 拆分大 Hash：user:1:profile / user:1:settings（分散欄位）
  ✓ 用 HSCAN / SSCAN 分批讀取（不要用 HGETALL / SMEMBERS 一次取全部）
  ✓ 刪除大 Key 用 UNLINK（非同步刪除，不阻塞）而不是 DEL

──────────────────────────────────────────────────────────
問題 4：慢日誌排查（slowlog）
──────────────────────────────────────────────────────────

# 設定：記錄執行時間超過 10ms 的命令
CONFIG SET slowlog-log-slower-than 10000   # 單位：微秒（10000 = 10ms）
CONFIG SET slowlog-max-len 128              # 最多記錄 128 條

# 查看最近 10 條慢日誌
SLOWLOG GET 10
# 每條包含：唯一 ID、時間戳、執行時間（微秒）、命令與參數

# 清空慢日誌
SLOWLOG RESET

單節點 Redis（開發/小型生產）：
  → 簡單，適合 Cache 場景（資料可以丟失）
  → 不適合分散式鎖（節點掛掉，鎖也消失）

Redis Sentinel（主從 + 自動切換）：
  → 一主多從 + 3 個 Sentinel 監控節點
  → 主節點掛掉時，Sentinel 自動選出新的主節點
  → 適合中型系統，提供 HA 但不提供水平擴展

Redis Cluster（分片 + HA）：
  → 資料自動分散到多個節點（16384 個 Hash Slot）
  → 每個節點有副本，節點掛掉自動切換
  → 適合超大規模系統，提供水平擴展 + HA
  → 注意：Cluster 模式下，多 Key 命令（MGET、Pipeline）
    需要所有 Key 在同一個 Slot，否則報錯

Cloud 託管（最推薦）：
  → AWS ElastiCache、GCP Memorystore、Upstash
  → 自動處理 HA、備份、版本升級，工程師不用管運維