Redis 緩存策略與分佈式鎖實戰 2026 | 高併發架構核心組件

Redis 緩存策略與分佈式鎖實戰

Redis 是高併發架構中不可或缺的組件。本文將系統講解 Redis 緩存策略、緩存三大經典問題的解決方案、分佈式鎖的實現原理及生產環境最佳實踐。

一、Redis 緩存模式

1.1 Cache-Aside（旁路緩存）

最常用的緩存模式，應用程序先查緩存，未命中再查數據庫：

typescript

// Cache-Aside 模式實現
class CacheAsideService {
  constructor(
    private redis: RedisClient,
    private db: Database
  ) {}

  async getUser(id: string): Promise<User | null> {
    const cacheKey = `user:${id}`

    // 1. 先查緩存
    const cached = await this.redis.get(cacheKey)
    if (cached) {
      return JSON.parse(cached)
    }

    // 2. 緩存未命中，查數據庫
    const user = await this.db.user.findById(id)
    if (!user) {
      return null
    }

    // 3. 寫入緩存（設置過期時間）
    await this.redis.setex(cacheKey, 3600, JSON.stringify(user))

    return user
  }

  async updateUser(id: string, data: Partial<User>): Promise<void> {
    // 1. 更新數據庫
    await this.db.user.update(id, data)

    // 2. 刪除緩存（而非更新緩存）
    await this.redis.del(`user:${id}`)
  }
}

為什麼刪除緩存而不是更新緩存？

避免併發寫入導致數據不一致
部分場景更新緩存計算成本高
懶加載思想，下次讀取時再緩存

1.2 Write-Through（直寫模式）

寫入時同時更新緩存和數據庫，由緩存組件協調：

typescript

class WriteThroughService {
  constructor(
    private redis: RedisClient,
    private db: Database
  ) {}

  async writeUser(user: User): Promise<void> {
    // 同時寫入緩存和數據庫
    const pipeline = this.redis.pipeline()
    pipeline.set(`user:${user.id}`, JSON.stringify(user), 'EX', 3600)
    pipeline.exec()

    await this.db.user.upsert(user)
  }

  async readUser(id: string): Promise<User | null> {
    // 緩存中一定有數據（因為寫入時已同步）
    const cached = await this.redis.get(`user:${id}`)
    return cached ? JSON.parse(cached) : null
  }
}

1.3 Write-Behind（異步寫模式）

先寫緩存，異步批量寫入數據庫，性能最高但有一致性風險：

typescript

class WriteBehindService {
  private writeQueue: Map<string, any> = new Map()
  private flushTimer: NodeJS.Timeout

  constructor(
    private redis: RedisClient,
    private db: Database
  ) {
    // 每 5 秒批量刷入數據庫
    this.flushTimer = setInterval(() => this.flush(), 5000)
  }

  async writeUser(user: User): Promise<void> {
    // 1. 立即寫入緩存
    await this.redis.set(`user:${user.id}`, JSON.stringify(user), 'EX', 3600)

    // 2. 加入寫入隊列
    this.writeQueue.set(user.id, user)
  }

  private async flush(): Promise<void> {
    if (this.writeQueue.size === 0) return

    const batch = Array.from(this.writeQueue.values())
    this.writeQueue.clear()

    try {
      // 批量寫入數據庫
      await this.db.user.batchUpsert(batch)
      console.log(`Flushed ${batch.length} records to database`)
    } catch (err) {
      // 寫入失敗，重新加入隊列
      batch.forEach(item => this.writeQueue.set(item.id, item))
      console.error('Flush failed:', err)
    }
  }
}

1.4 緩存模式對比

模式	一致性	性能	複雜度	適用場景
Cache-Aside	最終一致	高	低	通用場景
Write-Through	強一致	中	中	一致性要求高
Write-Behind	弱一致	最高	高	日誌、計數器

二、緩存三大問題

2.1 緩存穿透

問題：查詢不存在的數據，緩存和數據庫都沒有，每次請求都打到數據庫。

用戶 → 緩存(未命中) → 數據庫(未命中) → 返回 null
                                                    ↓
惡意攻擊：大量請求不存在的 ID → 數據庫壓力驟增

解決方案 1：緩存空值

typescript

async getUser(id: string): Promise<User | null> {
  const cacheKey = `user:${id}`

  const cached = await this.redis.get(cacheKey)
  if (cached !== null) {
    if (cached === 'NULL') return null  // 空值標記
    return JSON.parse(cached)
  }

  const user = await this.db.user.findById(id)
  if (!user) {
    // 緩存空值，設置較短的過期時間（60秒）
    await this.redis.setex(cacheKey, 60, 'NULL')
    return null
  }

  await this.redis.setex(cacheKey, 3600, JSON.stringify(user))
  return user
}

解決方案 2：布隆過濾器

typescript

import BloomFilter from 'bloom-filter'

class BloomGuard {
  private filter: BloomFilter

  constructor(size: number = 1000000) {
    this.filter = BloomFilter.create(size, 0.01) // 1% 誤判率
  }

  // 初始化：加載所有存在的 ID
  async init(db: Database) {
    const ids = await db.user.getAllIds()
    ids.forEach(id => this.filter.add(id))
  }

  mightExist(id: string): boolean {
    return this.filter.contains(id)
  }
}

// 使用
async getUser(id: string): Promise<User | null> {
  // 布隆過濾器前置檢查
  if (!this.bloom.mightExist(id)) {
    return null  // 一定不存在
  }

  // 正常查緩存 → 數據庫流程
  return this.cacheAside.getUser(id)
}

2.2 緩存擊穿

問題：熱點數據過期瞬間，大量併發請求同時打到數據庫。

熱點 key 過期 → 1000 個併發請求 → 同時查數據庫 → 數據庫崩潰

解決方案 1：互斥鎖

typescript

async getHotData(key: string): Promise<any> {
  const cached = await this.redis.get(key)
  if (cached) return JSON.parse(cached)

  // 獲取互斥鎖
  const lockKey = `lock:${key}`
  const lockAcquired = await this.redis.set(
    lockKey, '1', 'NX', 'EX', 10  // 10秒自動過期
  )

  if (lockAcquired) {
    try {
      // 再次檢查緩存（可能已被其他請求填充）
      const cached = await this.redis.get(key)
      if (cached) return JSON.parse(cached)

      // 查詢數據庫
      const data = await this.db.query(key)
      await this.redis.setex(key, 3600, JSON.stringify(data))
      return data
    } finally {
      await this.redis.del(lockKey)
    }
  } else {
    // 等待 50ms 後重試
    await sleep(50)
    return this.getHotData(key)
  }
}

解決方案 2：邏輯過期

typescript

interface CachedData<T> {
  data: T
  expire: number  // 邏輯過期時間
}

async getWithLogicalExpiry(key: string): Promise<any> {
  const cached = await this.redis.get(key)
  if (!cached) return null

  const parsed: CachedData<any> = JSON.parse(cached)

  // 未邏輯過期，直接返回
  if (Date.now() < parsed.expire) {
    return parsed.data
  }

  // 邏輯過期，嘗試異步刷新
  const lockKey = `lock:${key}`
  const lockAcquired = await this.redis.set(lockKey, '1', 'NX', 'EX', 10)

  if (lockAcquired) {
    // 異步刷新緩存
    setImmediate(async () => {
      try {
        const data = await this.db.query(key)
        const newData: CachedData<any> = {
          data,
          expire: Date.now() + 3600 * 1000
        }
        await this.redis.set(key, JSON.stringify(newData))
      } finally {
        await this.redis.del(lockKey)
      }
    })
  }

  // 返回舊數據（不阻塞用戶）
  return parsed.data
}

2.3 緩存雪崩

問題：大量緩存同時過期，或 Redis 宕機，所有請求打到數據庫。

解決方案 1：隨機過期時間

typescript

async cacheData(key: string, data: any, ttl: number): Promise<void> {
  // 基礎 TTL + 隨機偏移（0~300秒）
  const randomTTL = ttl + Math.floor(Math.random() * 300)
  await this.redis.setex(key, randomTTL, JSON.stringify(data))
}

// 批量緩存時設置不同過期時間
async cacheUsers(users: User[]): Promise<void> {
  const pipeline = this.redis.pipeline()
  users.forEach((user, index) => {
    // 每個用戶的過期時間略有不同
    const ttl = 3600 + Math.floor(Math.random() * 600)
    pipeline.setex(`user:${user.id}`, ttl, JSON.stringify(user))
  })
  await pipeline.exec()
}

解決方案 2：多級緩存

typescript

class MultiLevelCache {
  // L1: 本地內存緩存（最快）
  private localCache = new Map<string, { data: any; expire: number }>()

  // L2: Redis 緩存
  constructor(private redis: RedisClient) {}

  async get(key: string): Promise<any> {
    // L1: 檢查本地緩存
    const local = this.localCache.get(key)
    if (local && local.expire > Date.now()) {
      return local.data
    }

    // L2: 檢查 Redis
    const cached = await this.redis.get(key)
    if (cached) {
      const data = JSON.parse(cached)
      // 回填 L1（本地緩存 60 秒）
      this.localCache.set(key, { data, expire: Date.now() + 60000 })
      return data
    }

    return null
  }

  async set(key: string, data: any, ttl: number): Promise<void> {
    // 寫入 L1
    this.localCache.set(key, { data, expire: Date.now() + 60000 })
    // 寫入 L2
    await this.redis.setex(key, ttl, JSON.stringify(data))
  }
}

三、分佈式鎖

3.1 基礎實現

typescript

class RedisLock {
  constructor(private redis: RedisClient) {}

  async acquire(key: string, ttl: number = 10): Promise<string | null> {
    // 生成唯一標識（防止誤解鎖）
    const lockId = crypto.randomUUID()

    // SET key value NX EX ttl
    const acquired = await this.redis.set(
      `lock:${key}`, lockId, 'NX', 'EX', ttl
    )

    return acquired ? lockId : null
  }

  async release(key: string, lockId: string): Promise<boolean> {
    // 使用 Lua 腳本保證原子性
    const script = `
      if redis.call("get", KEYS[1]) == ARGV[1] then
        return redis.call("del", KEYS[1])
      else
        return 0
      end
    `

    const result = await this.redis.eval(
      script, 1, `lock:${key}`, lockId
    )

    return result === 1
  }
}

3.2 帶自動續期的分佈式鎖

typescript

class AutoRenewLock {
  private renewTimers: Map<string, NodeJS.Timeout> = new Map()

  constructor(private redis: RedisClient) {}

  async acquire(
    key: string,
    ttl: number = 30,
    timeout: number = 5000
  ): Promise<string | null> {
    const lockId = crypto.randomUUID()
    const startTime = Date.now()

    // 嘗試獲取鎖，帶超時
    while (Date.now() - startTime < timeout) {
      const acquired = await this.redis.set(
        `lock:${key}`, lockId, 'NX', 'EX', ttl
      )

      if (acquired) {
        // 啟動自動續期
        this.startRenewal(key, lockId, ttl)
        return lockId
      }

      // 等待 100ms 後重試
      await sleep(100)
    }

    return null
  }

  private startRenewal(key: string, lockId: string, ttl: number): void {
    // 每 ttl/3 秒續期一次
    const interval = Math.floor(ttl * 1000 / 3)

    const timer = setInterval(async () => {
      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 this.redis.eval(
        script, 1, `lock:${key}`, lockId, String(ttl)
      )

      if (result === 0) {
        // 鎖已丟失，停止續期
        this.stopRenewal(key)
      }
    }, interval)

    this.renewTimers.set(key, timer)
  }

  private stopRenewal(key: string): void {
    const timer = this.renewTimers.get(key)
    if (timer) {
      clearInterval(timer)
      this.renewTimers.delete(key)
    }
  }

  async release(key: string, lockId: string): Promise<void> {
    this.stopRenewal(key)

    const script = `
      if redis.call("get", KEYS[1]) == ARGV[1] then
        return redis.call("del", KEYS[1])
      else
        return 0
      end
    `

    await this.redis.eval(script, 1, `lock:${key}`, lockId)
  }
}

3.3 使用示例

typescript

// 庫存扣減場景
async deductStock(productId: string, quantity: number): Promise<boolean> {
  const lock = new AutoRenewLock(this.redis)
  const lockKey = `stock:${productId}`
  const lockId = await lock.acquire(lockKey, 30, 5000)

  if (!lockId) {
    throw new Error('獲取鎖超時，請重試')
  }

  try {
    // 查詢庫存
    const stock = await this.db.product.getStock(productId)
    if (stock < quantity) {
      return false
    }

    // 扣減庫存
    await this.db.product.updateStock(productId, stock - quantity)
    return true
  } finally {
    await lock.release(lockKey, lockId)
  }
}

3.4 Redlock 算法

單點 Redis 鎖在主從切換時可能丟失鎖。Redlock 通過多個獨立 Redis 節點提高可靠性：

typescript

class Redlock {
  private servers: RedisClient[]

  constructor(servers: RedisClient[]) {
    this.servers = servers
  }

  async acquire(key: string, ttl: number = 10): Promise<string | null> {
    const lockId = crypto.randomUUID()
    const quorum = Math.floor(this.servers.length / 2) + 1
    const startTime = Date.now()

    // 向所有節點請求加鎖
    const results = await Promise.allSettled(
      this.servers.map(server =>
        server.set(`lock:${key}`, lockId, 'NX', 'EX', ttl)
      )
    )

    // 統計成功數量
    const acquired = results.filter(
      r => r.status === 'fulfilled' && r.value
    ).length

    // 計算耗時
    const elapsed = (Date.now() - startTime) / 1000

    if (acquired >= quorum && elapsed < ttl) {
      return lockId
    }

    // 加鎖失敗，釋放所有已獲取的鎖
    await this.release(key, lockId)
    return null
  }

  async release(key: string, lockId: string): Promise<void> {
    const script = `
      if redis.call("get", KEYS[1]) == ARGV[1] then
        return redis.call("del", KEYS[1])
      else
        return 0
      end
    `

    await Promise.allSettled(
      this.servers.map(server =>
        server.eval(script, 1, `lock:${key}`, lockId)
      )
    )
  }
}

四、Redis 數據結構選型

4.1 常用數據結構對比

結構	場景	時間複雜度	示例
String	緩存、計數器	O(1)	`SET key value`
Hash	對象存儲	O(1)	`HSET user:1 name "Alice"`
List	消息隊列	O(1)~O(N)	`LPUSH queue msg`
Set	去重、標籤	O(1)	`SADD tags "vue"`
ZSet	排行榜	O(logN)	`ZADD rank 100 "user1"`
Stream	消息流	O(1)	`XADD stream * key val`
Bitmap	簽到	O(1)	`SETBIT sign:uid 100 1`
HyperLogLog	UV 統計	O(1)	`PFADD uv user1`

4.2 排行榜實現

typescript

class LeaderboardService {
  constructor(private redis: RedisClient) {}

  // 添加分數
  async addScore(gameId: string, userId: string, score: number): Promise<void> {
    await this.redis.zadd(`leaderboard:${gameId}`, score, userId)
  }

  // 獲取 Top N
  async getTopN(gameId: string, n: number = 10): Promise<Array<{ userId: string; score: number }>> {
    const results = await this.redis.zrevrange(
      `leaderboard:${gameId}`, 0, n - 1, 'WITHSCORES'
    )

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

    return leaderboard
  }

  // 獲取用戶排名
  async getUserRank(gameId: string, userId: string): Promise<{ rank: number; score: number } | null> {
    const rank = await this.redis.zrevrank(`leaderboard:${gameId}`, userId)
    const score = await this.redis.zscore(`leaderboard:${gameId}`, userId)

    if (rank === null || score === null) return null

    return { rank: rank + 1, score: parseFloat(score) }
  }
}

4.3 限流器實現

typescript

class RateLimiter {
  constructor(private redis: RedisClient) {}

  // 滑動窗口限流
  async isAllowed(
    userId: string,
    action: string,
    limit: number,
    windowSec: number
  ): Promise<boolean> {
    const key = `rate:${action}:${userId}`
    const now = Date.now()
    const windowStart = now - windowSec * 1000

    const pipeline = this.redis.pipeline()

    // 移除窗口外的記錄
    pipeline.zremrangebyscore(key, 0, windowStart)
    // 添加當前請求
    pipeline.zadd(key, now, `${now}`)
    // 統計窗口內請求數
    pipeline.zcard(key)
    // 設置過期時間
    pipeline.expire(key, windowSec)

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

    return count <= limit
  }
}

// 使用：每分鐘最多 60 次請求
const allowed = await rateLimiter.isAllowed(userId, 'api_call', 60, 60)
if (!allowed) {
  throw new Error('請求過於頻繁')
}

五、最佳實踐

5.1 Key 命名規範

業務:對象:ID:字段

# 示例
user:1001:profile
order:2024:1001
cache:api:/users:page1
lock:stock:product-1001
rate:login:user-1001

5.2 Pipeline 批量操作

typescript

// 不好：多次往返
for (const id of userIds) {
  await redis.get(`user:${id}`)
}

// 好：Pipeline 批量
const pipeline = redis.pipeline()
userIds.forEach(id => pipeline.get(`user:${id}`))
const results = await pipeline.exec()

5.3 連接池配置

typescript

import Redis from 'ioredis'

const redis = new Redis({
  host: '127.0.0.1',
  port: 6379,
  maxRetriesPerRequest: 3,
  enableReadyCheck: true,
  retryStrategy: (times) => Math.min(times * 100, 3000),

  // 連接池
  family: 4,
  keepAlive: true,
  connectionTimeout: 5000,
  commandTimeout: 3000,
})

// 集群模式
const cluster = new Redis.Cluster(
  [
    { host: '10.0.0.1', port: 6379 },
    { host: '10.0.0.2', port: 6379 },
    { host: '10.0.0.3', port: 6379 },
  ],
  {
    scaleReads: 'slave',
    maxRedirections: 16,
    retryDelayOnFailover: 200,
    redisOptions: { password: process.env.REDIS_PASSWORD }
  }
)

六、總結

✅ 三種緩存模式（Cache-Aside、Write-Through、Write-Behind）
✅ 緩存穿透（空值緩存、布隆過濾器）
✅ 緩存擊穿（互斥鎖、邏輯過期）
✅ 緩存雪崩（隨機 TTL、多級緩存）
✅ 分佈式鎖（基礎鎖、自動續期鎖、Redlock 算法）
✅ 數據結構選型與實戰（排行榜、限流器）
✅ 最佳實踐（Key 命名、Pipeline、連接池）

Redis 是高併發系統的核心組件，掌握緩存策略和分佈式鎖是構建高可用架構的關鍵能力。

相關閱讀：

Redis 緩存策略與分佈式鎖實戰 2026 | 高併發架構核心組件 ​

一、Redis 緩存模式 ​

1.1 Cache-Aside（旁路緩存） ​

1.2 Write-Through（直寫模式） ​

1.3 Write-Behind（異步寫模式） ​

1.4 緩存模式對比 ​

二、緩存三大問題 ​

2.1 緩存穿透 ​

2.2 緩存擊穿 ​

2.3 緩存雪崩 ​

三、分佈式鎖 ​

3.1 基礎實現 ​

3.2 帶自動續期的分佈式鎖 ​

3.3 使用示例 ​

3.4 Redlock 算法 ​

四、Redis 數據結構選型 ​

4.1 常用數據結構對比 ​

4.2 排行榜實現 ​

4.3 限流器實現 ​

五、最佳實踐 ​

5.1 Key 命名規範 ​

5.2 Pipeline 批量操作 ​

5.3 連接池配置 ​

六、總結 ​

Redis 緩存策略與分佈式鎖實戰 2026 | 高併發架構核心組件

一、Redis 緩存模式

1.1 Cache-Aside（旁路緩存）

1.2 Write-Through（直寫模式）

1.3 Write-Behind（異步寫模式）

1.4 緩存模式對比

二、緩存三大問題

2.1 緩存穿透

2.2 緩存擊穿

2.3 緩存雪崩

三、分佈式鎖

3.1 基礎實現

3.2 帶自動續期的分佈式鎖

3.3 使用示例

3.4 Redlock 算法

四、Redis 數據結構選型

4.1 常用數據結構對比

4.2 排行榜實現

4.3 限流器實現

五、最佳實踐

5.1 Key 命名規範

5.2 Pipeline 批量操作

5.3 連接池配置

六、總結