问题出现

那天凌晨两点，手机开始疯狂震动

小王把他的短链接分享到了微博。

一个科技博主的转发，让他的 SaaS 产品页面一夜之间涌入了上万人。每一个访客都要经过我的短链接服务跳转——我的那台 1 核 2G 的小服务器，突然开始承载它从未见过的流量。

我盯着监控面板，看着数字一路飙升：

时间线：
- 第1周：23 个用户，156 个短链接，日访问量 43 次
- 第2周：89 个用户，1,245 个短链接，日访问量 1,234 次
- 第3周：234 个用户，5,678 个短链接，日访问量 6,789 次
- 第4周：567 个用户，12,345 个短链接，日访问量 15,678 次

增长速度：
- 用户增长：25 倍
- 短链接增长：79 倍
- 日访问量：43 → 15,678（365 倍）

一个月前，我还觉得这个项目没人用。现在，567 个用户每天都在依赖它。

这本该让我开心。但手机不停震动的通知告诉我——好戏才刚刚开始。

第一个告警：SQLite 扛不住了 💥

凌晨 2:17，第一条告警弹了出来：

[2024-02-15 02:17:33] ERROR: Database connection pool exhausted
[2024-02-15 02:17:33] ERROR: Unable to acquire database connection within 30s
[2024-02-15 02:17:34] ERROR: Request timeout: GET /d7f3b2
[2024-02-15 02:17:35] ERROR: Request timeout: GET /a3f8c2
[2024-02-15 02:17:35] ERROR: Request timeout: GET /b7d9e1
...

我立刻打开代码，心跳加速——

# 当时的数据库连接方式
import sqlite3

# 🔴 问题：每次请求都创建新连接
def get_db_connection():
    return sqlite3.connect('urls.db')

@app.route('/<short_code>')
def redirect_to_original(short_code):
    conn = get_db_connection()
    cursor = conn.cursor()
    cursor.execute(
        "SELECT long_url FROM urls WHERE short_code = ?",
        (short_code,)
    )
    result = cursor.fetchone()
    conn.close()
    if result:
        return redirect(result[0], code=302)
    else:
        return "Not found", 404

问题很清楚：SQLite 默认只允许 1 个写入连接。当并发请求达到上百个时，大量请求在排队等锁：

# SQLite 的致命限制
MAX_CONNECTIONS = 1  # 写入锁只有一个

# 当 100 个请求同时进来
concurrent_requests = 100
available_connections = 1
waiting_requests = concurrent_requests - available_connections
# 结果：99 个请求在等，最终超时

我没有犹豫。凌晨 2:30，我开始做一件从来没在生产环境做过的事——数据库迁移。

SQLite → PostgreSQL。手心全是汗。

import psycopg2
from psycopg2 import pool

# 创建连接池
connection_pool = psycopg2.pool.SimpleConnectionPool(
    minconn=5,
    maxconn=50,
    host='localhost',
    database='urls',
    user='user',
    password='password'
)

def get_db_connection():
    """从连接池获取连接"""
    try:
        return connection_pool.getconn()
    except Exception as e:
        print(f"获取连接失败：{e}")
        raise

def release_db_connection(conn):
    """释放连接回连接池"""
    connection_pool.putconn(conn)

@app.route('/<short_code>')
def redirect_to_original(short_code):
    conn = None
    try:
        conn = get_db_connection()
        cursor = conn.cursor()
        cursor.execute(
            "SELECT long_url FROM urls WHERE short_code = %s",
            (short_code,)
        )
        result = cursor.fetchone()
        if result:
            return redirect(result[0], code=302)
        else:
            return "Not found", 404
    finally:
        if conn:
            release_db_connection(conn)

凌晨 4 点，迁移完成。监控面板上的错误日志停止了滚动。

我瘫在椅子上，看着监控曲线恢复平稳。那一刻我明白了一件事：SQLite 是个好数据库，但它不是为这个场景设计的。

第二个告警：响应时间从 45ms 飙到 345ms 🐌

刚搞定数据库，新的问题又来了。

用户反馈”短链接打开变慢了”。我查了监控数据，整个人都不好了：

# 响应时间监控
response_time_monitor = {
    "第一周": {
        "avg": 45,      # 平均 45ms
        "p50": 42,
        "p95": 67,
        "p99": 123
    },
    "第二周": {
        "avg": 78,
        "p50": 65,
        "p95": 156,
        "p99": 345
    },
    "第三周": {
        "avg": 156,
        "p50": 123,
        "p95": 456,
        "p99": 1234
    },
    "第四周": {
        "avg": 345,
        "p50": 234,
        "p95": 1234,
        "p99": 3456  # 99 分位达到 3.5 秒！
    }
}

P99 从 123ms 涨到 3.5 秒。在用户看来，就是”点了没反应”。

我给每个重定向请求加了计时分析：

import time
import logging

def redirect_with_timing(short_code):
    start_time = time.time()

    # 步骤1：数据库查询
    db_start = time.time()
    result = db.query(
        "SELECT long_url FROM urls WHERE short_code = %s",
        (short_code,)
    )
    db_time = (time.time() - db_start) * 1000

    # 步骤2：日志记录
    log_start = time.time()
    log_click(short_code, request.remote_addr)
    log_time = (time.time() - log_start) * 1000

    total_time = (time.time() - start_time) * 1000
    logging.info(f"Database: {db_time:.2f}ms, Logging: {log_time:.2f}ms, Total: {total_time:.2f}ms")
    return redirect(result[0], code=302)

# 日志分析结果（第四周）
"""
时间分布统计：
- 数据库查询：312ms（平均）← 占 90%
- 日志记录：28ms（平均）
- 其他开销：5ms（平均）
- 总计：345ms
"""

90% 的时间花在数据库查询上。数据量从几百条涨到了一万多条，每次查询都在做全表扫描。

我打开数据库检查索引：

-- 检查当前索引
PRAGMA index_list(urls);

-- 🔴 发现：short_code 字段没有索引！

一个 CREATE INDEX 就能解决的事：

-- 创建索引
CREATE INDEX idx_short_code ON urls(short_code);

-- 验证效果
EXPLAIN QUERY PLAN
SELECT long_url FROM urls WHERE short_code = 'a3f8c2';
-- 优化前：全表扫描（SCAN TABLE）
-- 优化后：索引扫描（SEARCH TABLE USING INDEX）

跑了一遍性能测试，看到结果的那一刻，我长出一口气：

# 性能测试：10,000 次查询
# 无索引：45.67 秒
# 有索引：2.34 秒
# 性能提升：19.5 倍

一个索引，19.5 倍提升。这个教训我记一辈子——任何经常出现在 WHERE 子句里的字段，都必须建索引。

第三个告警：磁盘 87% 了 💾

我刚准备去补个觉，运维同学（其实就是我自己）发来消息：

磁盘空间告警：
- 总容量：100GB
- 已使用：87GB
- 可用空间：13GB
- 使用率：87%

预计：3 天后磁盘将满！

我写了个脚本分析磁盘占用：

import os

def analyze_disk_usage():
    db_size = os.path.getsize('urls.db') / (1024**3)
    print(f"数据库大小：{db_size:.2f} GB")

    log_size = os.path.getsize('access.log') / (1024**3)
    print(f"日志文件大小：{log_size:.2f} GB")

analyze_disk_usage()

# 输出：
# 数据库大小：12.34 GB
# 日志文件大小：45.67 GB  ← 日志占了将近一半！

日志 45 GB。每次短链接访问我都写一条详细日志，每天 15,678 次访问 × 约 200 字节 ≈ 3 MB/天。积少成多，一个月下来就爆了。

解决方案很直接——日志轮转 + 压缩：

# 方案1：日志轮转——单文件不超过 100MB，保留 10 个备份
import logging
from logging.handlers import RotatingFileHandler

handler = RotatingFileHandler(
    'access.log',
    maxBytes=100*1024*1024,  # 100MB
    backupCount=10
)
logger = logging.getLogger()
logger.addHandler(handler)

# 方案2：压缩旧日志
import gzip
import shutil

def compress_old_logs():
    old_log = 'access.log.7'
    with open(old_log, 'rb') as f_in:
        with gzip.open(f'{old_log}.gz', 'wb') as f_out:
            shutil.copyfileobj(f_in, f_out)
    os.remove(old_log)  # 删除原文件，节省约 90% 空间

# 方案3：精简日志格式
def log_click_minimal(short_code, ip):
    """只记录必要信息：短代码、IP、时间戳（约 50 字节）"""
    logger.info(f"{short_code},{ip},{int(time.time())}")
# 从 200 字节 → 50 字节，减少 75%

磁盘使用率从 87% 降到了 34%。这个问题的教训是：日志不是免费的，它是隐形成本最高的存储消耗。

最可怕的问题：短链接指向了别人的网站 🎲

如果说前面三个问题只是”系统不好用”，那这个就是”系统不可信”。

一个用户发来了愤怒的邮件：

我创建的短链接和别人的重复了！

我的短链接：https://short.url/a3f8c2 指向：https://mywebsite.com/page1

但打开后跳转到：https://otherwebsite.com/page2

请尽快处理！

我的心一下子提到了嗓子眼。短链接指向错误页面——这比宕机还严重。用户信任一旦崩塌，就再也回不来了。

我马上检查短码生成逻辑：

# 当时的短码生成
import hashlib

def generate_short_code(url):
    """使用 MD5 生成短码"""
    md5_hash = hashlib.md5(url.encode()).hexdigest()
    return md5_hash[:6]  # 只取前 6 位

问题出在这里：MD5 前 6 位只有 16^6 = 1677 万种组合。在 12,345 个链接的规模下，冲突已经出现了。

我跑了个测试：

def test_collision():
    urls = [f"https://example.com/page{i}" for i in range(100000)]
    short_codes = {}
    collisions = []

    for url in urls:
        code = generate_short_code(url)
        if code in short_codes:
            collisions.append({
                'code': code,
                'url1': short_codes[code],
                'url2': url
            })
        else:
            short_codes[code] = url

    print(f"测试 URL 数：{len(urls)}")
    print(f"冲突数量：{len(collisions)}")
    print(f"冲突率：{len(collisions)/len(urls)*100:.4f}%")

test_collision()
# 输出：
# 测试 URL 数：100,000
# 冲突数量：23
# 冲突率：0.0230%

万分之二的冲突率，听起来不高。但对于一个短链接服务来说，一次冲突就是一次事故。

我花了一整天重写短码生成方案，最终选择了自增 ID + Base62：

BASE62 = "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"

def generate_short_code_from_id():
    """使用自增 ID 生成短码——从根源上杜绝冲突"""
    # 插入记录，获取自增 ID
    cursor = db.execute(
        "INSERT INTO urls (long_url) VALUES (%s)",
        (url,)
    )
    last_id = cursor.lastrowid
    # 将 ID 转换为 Base62
    return encode_base62(last_id)

def encode_base62(num):
    """将数字转换为 Base62"""
    if num == 0:
        return BASE62[0]
    base62_str = ""
    while num > 0:
        base62_str = BASE62[num % 62] + base62_str
        num //= 62
    return base62_str

自增 ID 保证唯一，Base62 保证短码紧凑。6 位 Base62 可以表示 568 亿个链接，足够了。

优点一目了然：

1. 完全唯一，不可能有冲突
2. 短码更短，6 位就够了
3. 生成速度极快，不需要计算哈希

那天晚上，我给受影响的用户逐一发了道歉邮件。这是关系到用户信任的问题，不能出半点差错。

第五个告警：OOM Killer 把我的进程干掉了 💾

周四下午，服务突然完全无响应。

不是慢，是完全挂了。

我 SSH 到服务器，查看系统日志：

[2024-02-15 16:45:23] ERROR: Out of memory
[2024-02-15 16:45:23] ERROR: Killed (OOM Killer)

OOM Killer——Linux 内核在内存耗尽时强制杀掉进程的自救机制。我成了被杀掉的那个。

罪魁祸首很快找到了：

# 我之前为了"优化性能"加的内存缓存
class ProblematicCache:
    def __init__(self):
        self.cache = {}  # 🔴 无限制增长的字典

    def set(self, key, value):
        self.cache[key] = value  # 只进不出，迟早爆

    def get(self, key):
        return self.cache.get(key)

算了笔账：

cache_size = 1000000      # 100 万个 URL
memory_per_item = 1000    # 每个 URL 约 1KB
total_memory = cache_size * memory_per_item  # = 1 GB

# 而我服务器总共只有 2GB 内存
# 系统 + PostgreSQL + Flask + 日志 ≈ 1.2GB
# 缓存吃掉 1GB → 内存爆了

解决方案很明确：用专业的缓存服务替代 Python 字典。

我引入了 Redis，带 LRU（最近最少使用）淘汰策略：

import redis

redis_client = redis.Redis(
    host='localhost',
    port=6379,
    db=0,
    maxmemory='2gb',              # 最大 2GB
    maxmemory_policy='allkeys-lru' # 内存满了自动淘汰最少使用的
)

def cache_with_redis(short_code, long_url):
    """使用 Redis 缓存，1 小时过期"""
    redis_client.setex(
        f"url:{short_code}",
        3600,
        long_url
    )

def get_from_cache(short_code):
    """从 Redis 获取"""
    cached = redis_client.get(f"url:{short_code}")
    if cached:
        return cached.decode()
    return None

Redis 的好处是：内存用满时自动淘汰最冷的数据，不会无限制增长。同时缓存命中时响应时间从毫秒级降到微秒级。

重启服务后，我盯着内存监控看了整整一个小时，确认曲线稳住了才敢移开眼睛。

经历了这一周的连环打击，我决定重新设计架构 🏗️

周五晚上，我坐在电脑前，把这周经历的所有问题列了一遍：

1. SQLite 扛不住并发 → 换了 PostgreSQL
2. 数据库查询太慢 → 加了索引
3. 日志吃掉磁盘 → 做了日志轮转
4. 短码冲突导致数据错误 → 重写了短码生成
5. 内存溢出搞挂服务 → 引入了 Redis

每个问题都是临时救火，但系统已经变成了一堆补丁。我需要重新梳理一下整体架构：

┌─────────────────────────────────────────────────────────┐
│                      负载均衡器                          │
│                    (Nginx/HAProxy)                      │
└───────────────────────┬─────────────────────────────────┘
                        │
        ┌───────────────┼───────────────┐
        │               │               │
┌───────▼───────┐ ┌────▼────┐ ┌───────▼───────┐
│  Web 服务器1  │ │Web 服务器2│ │  Web 服务器3  │
│  (Flask/Gunicorn)         │ │               │
└───────┬───────┘ └────┬────┘ └───────┬───────┘
        │               │               │
        └───────────────┼───────────────┘
                        │
        ┌───────────────┼───────────────┐
        │               │               │
┌───────▼───────┐ ┌────▼────┐ ┌───────▼───────┐
│  Redis 缓存   │ │Redis 缓存│ │  Redis 缓存   │
│  (主从复制)   │ │(从节点) │ │  (从节点)     │
└───────┬───────┘ └────┬────┘ └───────┬───────┘
        │               │               │
        └───────────────┼───────────────┘
                        │
        ┌───────────────┴───────────────┐
        │                               │
┌───────▼────────┐            ┌────────▼────────┐
│  PostgreSQL    │            │   消息队列       │
│  (主从复制)    │            │  (日志收集)     │
└────────────────┘            └─────────────────┘

技术栈全面升级：

这一月我瘦了 3 斤，但系统终于稳住了 📊

周末，我跑了一轮完整的性能测试，对比优化前后的数据：

# 1000 个并发请求测试结果

# 优化前（SQLite + Python dict 缓存）
#   总耗时：125.67 秒
#   成功率：78.5%
#   平均响应时间：2345.67ms
#   QPS：7.96

# 优化后（PostgreSQL + Redis + 索引 + 连接池）
#   总耗时：2.34 秒
#   成功率：99.9%
#   平均响应时间：45.23ms
#   QPS：427.35

总结一下这一个月踩过的坑和填过的坑：

系统的演进路线也清晰了：

阶段1（单机）：
  Flask + SQLite + 本地缓存

阶段2（优化）：
  Flask + PostgreSQL + Redis + 连接池

阶段3（分布式）：
  负载均衡 + 多台服务器 + 主从复制

这一周我瘦了 3 斤，但系统终于稳住了。

关键经验

这一个月的救火经历，让我学到了几件事：

1. 提前规划：不要等问题出现再处理。如果我从一开始就用 PostgreSQL，第一个告警根本不会发生。
2. 监控是生命线：没有监控，我根本不知道系统出了什么问题。告警让我能在用户投诉前发现问题。
3. 逐步优化：一次解决一个问题。同时改太多东西，出了 bug 都不知道是哪个改动引起的。
4. 性能测试不能省：如果不是跑了对比测试，我不会有底气说”系统稳住了”。

但事情还没有结束。流量还在涨，用户开始抱怨短链接打开速度不够快。我需要更快的跳转速度……

想一想

问题1

如果你的服务突然被某个热门新闻引用，流量瞬间增长 100 倍，你会怎么做？

问题2

如何设计一个高可用的短链接系统？单机房扛不住时怎么办？

练习题

练习1

为什么从 SQLite 切换到 PostgreSQL 能大幅提升并发性能？我凌晨两点做这个决策时，心里其实也没底。

练习2

Redis 缓存为什么能提升性能？缓存穿透、缓存击穿、缓存雪崩分别是什么？

# 解决方案：缓存空值
def get_url_with_cache(short_code):
    cached = redis.get(f"url:{short_code}")
    if cached == "NULL":
        return None  # 命中了缓存的空值

    if cached:
        return cached

    result = db.query("SELECT long_url FROM urls WHERE short_code = %s", (short_code,))

    if result:
        redis.setex(f"url:{short_code}", 3600, result[0])
    else:
        # 缓存空值，防止穿透
        redis.setex(f"url:{short_code}", 300, "NULL")

    return result

# 解决方案：互斥锁
def get_url_with_lock(short_code):
    cached = redis.get(f"url:{short_code}")
    if cached:
        return cached

    lock_key = f"lock:url:{short_code}"
    lock = redis.set(lock_key, "1", nx=True, ex=10)

    if lock:
        try:
            result = db.query("SELECT long_url FROM urls WHERE short_code = %s", (short_code,))
            redis.setex(f"url:{short_code}", 3600, result[0])
            return result
        finally:
            redis.delete(lock_key)
    else:
        time.sleep(0.1)
        return get_url_with_lock(short_code)

# 解决方案：过期时间加随机值
import random

def set_cache_with_random_expire(key, value, base_expire=3600):
    expire = base_expire + random.randint(0, 300)  # 基础时间 + 随机 0-300 秒
    redis.setex(key, expire, value)

（下一节：自定义短链接）