缓存设计

日调用量突破 10 万次

选了 302 重定向后，每次跳转都要查数据库。起初还好，但随着用户增多，问题开始显现。

某天早上，监控告警响了。数据库 CPU 使用率飙到 60%。

我打开日志，看到密密麻麻的查询：

SELECT long_url FROM url_mapping WHERE short_code = 'a1b2c3';
SELECT long_url FROM url_mapping WHERE short_code = 'x9y8z7';
SELECT long_url FROM url_mapping WHERE short_code = 'p4q5r6';
... 成千上万条

短链接服务的访问特点，我总结了一下：

读取 vs 写入比例：1000:1

典型的访问模式：
- 创建短链接：1 次
- 重定向跳转：1000 次+

一个短链接被创建后，可能被访问成千上万次。如果每次都查数据库，数据库很快就扛不住。

我知道，不能再每次都查数据库了。

第一次接触 Redis

作为前端出身的开发者，这是我第一次接触 NoSQL。

我花了一天学习 Redis。它和 MySQL 完全不同：

MySQL（关系型数据库）：

• 数据存在磁盘上
• 查询需要解析 SQL、优化执行计划
• 单次查询耗时：10-50ms
• 适合复杂查询、事务

Redis（内存数据库）：

• 数据存在内存中
• 简单的 Key-Value 操作
• 单次查询耗时：0.1-2ms
• 适合高速读写

我在本地启动 Redis，尝试基本操作：

import redis

# 连接 Redis
r = redis.Redis(host='localhost', port=6379, db=0)

# 设置值
r.set('url:a1b2c3', 'https://example.com/very/long/url')
# 设置值并过期时间（秒）
r.setex('url:x9y8z7', 3600, 'https://another.com/path')

# 获取值
long_url = r.get('url:a1b2c3')
print(long_url)  # b'https://example.com/very/long/url'

# 删除
r.delete('url:a1b2c3')

# 检查是否存在
exists = r.exists('url:a1b2c3')  # 0

简单，直接，快速。

这就是我需要的。

缓存架构设计

最简单的方案：查询前先查 Redis，命中则直接返回，未命中则查数据库并回填缓存。

架构图：

                    用户请求
                       ↓
                ┌──────────────┐
                │  API 服务    │
                └──────────────┘
                       ↓
           ┌───────────┴───────────┐
           ↓                       ↓
    ┌──────────────┐        ┌──────────────┐
    │  Redis 缓存  │  ←───  │   MySQL      │
    │  (热数据)    │  回写   │  (全量数据)  │
    └──────────────┘        └──────────────┘

这种模式叫 Cache-Aside（旁路缓存）：

1. 先读缓存，命中直接返回
2. 未命中读数据库
3. 写入缓存，设置过期时间

我用 Python 实现：

def get_long_url(short_code):
    cache_key = f"url:{short_code}"

    # 1. 先读缓存
    long_url = redis.get(cache_key)
    if long_url:
        return long_url.decode()

    # 2. 缓存未命中，读数据库
    long_url = db.query(
        "SELECT long_url FROM url_mapping WHERE short_code = %s",
        (short_code,)
    )

    if long_url:
        # 3. 写入缓存
        redis.setex(cache_key, 3600, long_url)
        return long_url

    return None

为什么设置过期时间？

内存有限，不能无限存储。过期时间让不常用的数据自动清理。

# 配置
CACHE_TTL = {
    'url_mapping': 3600,      # 映射关系：1 小时
    'url_reverse': 86400,     # 反向映射：24 小时
    'click_count': 0,         # 点击计数：永不过期
}

缓存 Key 的设计：

# 短链接码 → 长 URL（核心缓存）
"url:{short_code}"
例如：url:a1b2c3

# 长 URL → 短链接码（去重缓存）
"url:reverse:{md5_prefix}"
例如：url:reverse:abc123def456

# 点击计数器
"click:{short_code}"
例如：click:a1b2c3

# 今日点击数
"click:today:{short_code}:{date}"
例如：click:today:a1b2c3:20260331

# 用户短链接列表
"user:urls:{user_id}"
例如：user:urls:12345

效果验证

上线后，我盯了一天监控数据。

结果超出预期：

92% 的请求从缓存返回，只有 8% 需要查数据库。

我写了个性能测试脚本验证：

import time
import statistics

def benchmark(shortener, short_code, iterations=1000):
    """性能测试"""
    times = []

    for _ in range(iterations):
        start = time.perf_counter()
        shortener.get(short_code)
        times.append(time.perf_counter() - start)

    print(f"平均响应：{statistics.mean(times)*1000:.2f}ms")
    print(f"P95 响应：{statistics.quantiles(times, n=20)[18]*1000:.2f}ms")
    print(f"P99 响应：{statistics.quantiles(times, n=100)[98]*1000:.2f}ms")

# 测试结果
# 无缓存：平均 50ms
# 有缓存：平均 5ms
# 性能提升：10 倍

数据库压力骤降，系统稳定了。

缓存一致性：新问题出现

但缓存带来了新问题。

某天用户反馈：“我修改了原始 URL，但跳转还是旧的地址。”

我检查数据库，确实是新地址。但 Redis 里还是旧的。

问题根源：用户修改了数据，但缓存没有同步更新。

时间线：
1. 短链接 a1b2c3 指向 https://old-url.com
2. 缓存存储：url:a1b2c3 = https://old-url.com（1 小时过期）
3. 用户修改：a1b2c3 改为指向 https://new-url.com
4. 数据库已更新，但缓存还是旧的
5. 1 小时内，所有请求都跳转到旧地址

这是典型的缓存一致性问题。

解决方案 1：主动失效

修改数据时，主动删除缓存：

def update_url(short_code, new_long_url):
    # 1. 更新数据库
    db.execute(
        "UPDATE url_mapping SET long_url = %s WHERE short_code = %s",
        (new_long_url, short_code)
    )

    # 2. 删除缓存（下次查询时会重新加载）
    redis.delete(f"url:{short_code}")

下次查询时，缓存未命中，从数据库加载最新数据。

解决方案 2：双写

修改数据时，同时更新缓存：

def update_url(short_code, new_long_url):
    # 1. 更新数据库
    db.execute(
        "UPDATE url_mapping SET long_url = %s WHERE short_code = %s",
        (new_long_url, short_code)
    )

    # 2. 更新缓存
    redis.setex(f"url:{short_code}", 3600, new_long_url)

我选择主动失效

原因：

1. 更简单：不用考虑新旧值
2. 更安全：数据库是唯一真相
3. 更灵活：修改后可能不会立即访问，删除缓存节省内存

删除操作也有个问题：如果删除失败怎么办？

我加了重试：

def update_url(short_code, new_long_url):
    # 更新数据库
    db.execute(
        "UPDATE url_mapping SET long_url = %s WHERE short_code = %s",
        (new_long_url, short_code)
    )

    # 删除缓存（带重试）
    for attempt in range(3):
        try:
            redis.delete(f"url:{short_code}")
            break
        except Exception as e:
            print(f"删除缓存失败，重试 {attempt + 1}/3: {e}")
            time.sleep(0.1)

缓存穿透：恶意攻击

解决了缓存一致性，又出现了新问题。

监控显示大量短链接码不存在：

访问不存在的短链接：
- url:xyz123 → 缓存没有 → 查数据库 → 也没有
- url:abc999 → 缓存没有 → 查数据库 → 也没有
- url:random → 缓存没有 → 查数据库 → 也没有

有人用随机短链接码恶意攻击。

每次都穿透缓存，打到数据库。这就是缓存穿透。

解决方案：缓存空值

def get_long_url(short_code):
    cache_key = f"url:{short_code}"

    # 查缓存
    long_url = redis.get(cache_key)
    if long_url is not None:
        if long_url == b'__NULL__':
            return None  # 缓存的空值
        return long_url.decode()

    # 查数据库
    long_url = db.get_long_url(short_code)

    if long_url:
        redis.setex(cache_key, 3600, long_url)
    else:
        # 缓存空值，设置较短过期时间
        redis.setex(cache_key, 300, '__NULL__')

    return long_url

效果：

空值缓存过期时间设置为 5 分钟：既能防护，又不会太久影响正常创建。

缓存雪崩：集体过期

还有一个隐患：大量缓存同时过期。

场景：
- 10:00 批量创建 1000 个短链接，缓存过期时间都是 1 小时
- 11:00 这 1000 个缓存同时过期
- 11:00:01 突然 1000 个请求打到数据库
- 数据库崩溃

这就是缓存雪崩。

解决方案：随机过期时间

import random

def cache_url(short_code, long_url):
    # 基础过期时间 + 随机波动
    base_ttl = 3600
    random_ttl = random.randint(-600, 600)  # ±10 分钟
    ttl = base_ttl + random_ttl

    redis.setex(f"url:{short_code}", ttl, long_url)

每个缓存的过期时间都有差异，避免同时失效。

完整实现

我把所有方案整合到一起：

import redis
import hashlib
import random
from typing import Optional
from dataclasses import dataclass

@dataclass
class URLRecord:
    short_code: str
    long_url: str
    click_count: int
    status: int

class CachedURLShortener:
    def __init__(self, db_conn, redis_conn):
        self.db = db_conn
        self.redis = redis_conn

        # 缓存配置
        self.URL_CACHE_TTL = 3600       # URL 映射：1 小时
        self.REVERSE_CACHE_TTL = 86400  # 反向映射：24 小时
        self.NULL_CACHE_TTL = 300       # 空值缓存：5 分钟

    def get(self, short_code: str) -> Optional[URLRecord]:
        """获取长 URL（重定向使用）"""
        cache_key = f"url:{short_code}"

        # 1. 查缓存
        cached = self.redis.get(cache_key)
        if cached is not None:
            if cached == b'__NULL__':
                return None  # 缓存的空值

            data = cached.decode()
            # 异步更新点击数
            self._record_click(short_code)
            return URLRecord(
                short_code=short_code,
                long_url=data,
                click_count=0,  # 缓存中不存储点击数
                status=1
            )

        # 2. 查数据库
        record = self.db.get_record(short_code)
        if record:
            # 3. 写入缓存（带随机过期）
            ttl = self.URL_CACHE_TTL + random.randint(-600, 600)
            self.redis.setex(
                cache_key,
                ttl,
                record['long_url']
            )
            # 异步更新点击数
            self._record_click(short_code)
            return record

        # 4. 缓存空值（防穿透）
        self.redis.setex(cache_key, self.NULL_CACHE_TTL, '__NULL__')
        return None

    def find_by_long_url(self, long_url: str) -> Optional[str]:
        """根据长 URL 查找短链接码（去重使用）"""
        url_hash = self._hash_url(long_url)
        cache_key = f"url:reverse:{url_hash}"

        # 1. 查缓存
        cached = self.redis.get(cache_key)
        if cached:
            return cached.decode()

        # 2. 查数据库
        short_code = self.db.find_short_code(long_url)
        if short_code:
            # 3. 写入缓存
            self.redis.setex(
                cache_key,
                self.REVERSE_CACHE_TTL,
                short_code
            )
            return short_code

        return None

    def set(self, short_code: str, long_url: str):
        """创建短链接"""
        url_hash = self._hash_url(long_url)

        # 事务：同时写入数据库和缓存
        pipe = self.redis.pipeline()

        # URL 映射缓存（带随机过期）
        ttl = self.URL_CACHE_TTL + random.randint(-600, 600)
        pipe.setex(f"url:{short_code}", ttl, long_url)

        # 反向映射缓存
        pipe.setex(f"url:reverse:{url_hash}", self.REVERSE_CACHE_TTL, short_code)

        pipe.execute()

        # 写入数据库
        self.db.insert(short_code, long_url)

    def update(self, short_code: str, new_long_url: str):
        """更新短链接"""
        # 1. 更新数据库
        self.db.update(short_code, new_long_url)

        # 2. 删除缓存（主动失效）
        for attempt in range(3):
            try:
                self.redis.delete(f"url:{short_code}")
                break
            except Exception as e:
                print(f"删除缓存失败，重试 {attempt + 1}/3: {e}")

    def delete(self, short_code: str):
        """删除短链接"""
        # 删除缓存
        self.redis.delete(f"url:{short_code}")

        # 删除数据库
        self.db.delete(short_code)

    def _record_click(self, short_code: str):
        """异步记录点击"""
        pipe = self.redis.pipeline()
        pipe.incr(f"click:{short_code}")
        pipe.incr(f"click:today:{short_code}:{self._today()}")
        pipe.execute()

    def _hash_url(self, url: str) -> str:
        return hashlib.md5(url.encode()).hexdigest()[:12]

    def _today(self) -> str:
        from datetime import datetime
        return datetime.now().strftime('%Y%m%d')

这套方案覆盖了：

1. Cache-Aside 模式：先查缓存，未命中查数据库
2. 缓存空值：防止穿透
3. 随机过期：防止雪崩
4. 主动失效：保证一致性

小结

核心要点：

1. Cache-Aside 模式最常用
2. 缓存空值防止穿透
3. 随机过期时间防止雪崩
4. 修改数据时主动删除缓存

缓存解决了大部分性能问题。数据库 CPU 从 60% 降到 15%，响应时间从 50ms 降到 8ms。

但 301 和 302 的选择，在有了缓存后，又有了新的考量……

练习题

练习 1

为什么缓存空值可以防止缓存穿透？空值缓存的过期时间应该设置多久？

攻击场景：
- 攻击者访问不存在的 short_code: "xyz123"
- 缓存中没有 → 查数据库 → 也没有
- 每次都打到数据库

防御方案：
- 第 1 次查询后，缓存空值：redis.setex("url:xyz123", 300, "__NULL__")
- 第 2-100 次查询：直接从缓存返回 NULL
- 300 秒后空值过期，下次查询才会再次访问数据库

# 配置建议
NULL_CACHE_TTL = 300  # 5 分钟

练习 2

设计一个支持热点数据自动识别的缓存策略。

class SmartCacheShortener:
    def __init__(self, db, redis):
        self.db = db
        self.redis = redis

        # 分级缓存配置
        self.TTL_COLD = 300       # 冷数据：5 分钟
        self.TTL_WARM = 3600      # 温数据：1 小时
        self.TTL_HOT = 86400 * 7  # 热数据：7 天

        # 阈值配置
        self.HOT_THRESHOLD = 1000    # 日访问>1000 为热数据
        self.WARM_THRESHOLD = 100    # 日访问>100 为温数据

    def get(self, short_code):
        cache_key = f"url:{short_code}"

        # 查缓存
        cached = self.redis.get(cache_key)
        if cached:
            record = self._parse_cached(cached)

            # 智能续期：热数据自动延长 TTL
            click_count = self._get_click_count(short_code)
            if click_count > self.HOT_THRESHOLD:
                self.redis.expire(cache_key, self.TTL_HOT)
            elif click_count > self.WARM_THRESHOLD:
                self.redis.expire(cache_key, self.TTL_WARM)

            return record

        # 查数据库
        record = self.db.get_record(short_code)
        if record:
            # 根据热度设置 TTL
            ttl = self._calculate_ttl(record['click_count'])
            self.redis.setex(cache_key, ttl, record['long_url'])
            return record

        # 缓存空值
        self.redis.setex(cache_key, 300, '__NULL__')
        return None

    def _calculate_ttl(self, click_count):
        """根据点击数计算 TTL"""
        if click_count > self.HOT_THRESHOLD:
            return self.TTL_HOT
        elif click_count > self.WARM_THRESHOLD:
            return self.TTL_WARM
        else:
            return self.TTL_COLD

    def _get_click_count(self, short_code):
        """获取今日点击数"""
        today = datetime.now().strftime('%Y%m%d')
        count = self.redis.get(f"click:today:{short_code}:{today}")
        return int(count) if count else 0