系统设计面试指南：从方法论到经典案例

1. 需求澄清（5-10 分钟）
   ├─ 功能需求：系统要做什么？
   ├─ 非功能需求：QPS、延迟、可用性要求？
   └─ 边界条件：用户量级、数据规模、读写比例？

2. 高层设计（10-15 分钟）
   ├─ 画出核心组件图
   ├─ 确定数据流方向
   └─ 选择通信协议（HTTP、WebSocket、gRPC、消息队列）

3. 深入细节（15-20 分钟）
   ├─ 数据模型设计
   ├─ 关键算法选型
   ├─ 扩展策略（分片、缓存、复制）
   └─ 瓶颈分析与优化

4. 总结回顾（5 分钟）
   ├─ 系统容量估算
   ├─ 可能的故障点
   └─ 未来演进方向

长 URL → MD5/SHA256 → 取前 7 位 Base62 → 短码

优点：实现简单，同 URL 生成同短码
缺点：Hash 碰撞需要额外处理

import hashlib

BASE62 = "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"

def generate_short_code(url: str) -> str:
    hash_hex = hashlib.md5(url.encode()).hexdigest()
    # 取前 8 个十六进制字符转为整数
    num = int(hash_hex[:8], 16)
    
    # 转为 Base62
    chars = []
    for _ in range(7):
        chars.append(BASE62[num % 62])
        num //= 62
    
    return ''.join(reversed(chars))

Snowflake ID → Base62 编码 → 短码

优点：无碰撞，可排序，包含时间戳信息
缺点：短码不固定长度

# 简易 Snowflake 实现
import time

class SnowflakeID:
    def __init__(self, worker_id: int, datacenter_id: int):
        self.worker_id = worker_id
        self.datacenter_id = datacenter_id
        self.sequence = 0
        self.last_timestamp = -1
        self.epoch = 1700000000000  # 自定义起始时间

    def next_id(self) -> int:
        timestamp = int(time.time() * 1000)
        
        if timestamp < self.last_timestamp:
            raise Exception('时钟回拨')
        
        if timestamp == self.last_timestamp:
            self.sequence = (self.sequence + 1) & 0xFFF
            if self.sequence == 0:
                while timestamp <= self.last_timestamp:
                    timestamp = int(time.time() * 1000)
        else:
            self.sequence = 0
        
        self.last_timestamp = timestamp
        
        return (
            ((timestamp - self.epoch) << 22) |
            (self.datacenter_id << 17) |
            (self.worker_id << 12) |
            self.sequence
        )

CREATE TABLE short_urls (
    id BIGINT PRIMARY KEY,
    short_code VARCHAR(10) UNIQUE NOT NULL,
    original_url TEXT NOT NULL,
    user_id BIGINT,
    expires_at TIMESTAMP,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    click_count BIGINT DEFAULT 0,
    INDEX idx_short_code (short_code),
    INDEX idx_expires_at (expires_at)
);

                    ┌──────────────┐
                    │   CDN/DNS    │
                    └──────┬───────┘
                           │
                    ┌──────▼───────┐
                    │ Load Balancer│
                    └──────┬───────┘
                           │
              ┌────────────┼────────────┐
              │            │            │
     ┌────────▼───┐ ┌─────▼────┐ ┌────▼──────┐
     │  API Server │ │ API Srv  │ │ API Server │
     └────────┬───┘ └─────┬────┘ └────┬──────┘
              │            │            │
              └────────────┼────────────┘
                           │
              ┌────────────┼────────────┐
              │            │            │
     ┌────────▼───┐ ┌─────▼─────┐ ┌───▼────────┐
     │   Redis     │ │  MySQL    │ │  Kafka      │
     │  (缓存热点)  │ │ (持久化)   │ │ (异步统计)   │
     └────────────┘ └───────────┘ └────────────┘

import redis
from functools import lru_cache

cache = redis.Redis(host='localhost', port=6379, decode_responses=True)

async def resolve_short_url(short_code: str) -> str | None:
    # 1. 先查本地缓存（L1）
    # 2. 再查 Redis（L2）
    
    original_url = cache.get(f"url:{short_code}")
    if original_url:
        return original_url
    
    # 3. 查数据库
    original_url = await db.fetch_one(
        "SELECT original_url FROM short_urls WHERE short_code = ?",
        short_code
    )
    
    if original_url:
        # 写入 Redis，TTL 根据过期时间或默认 7 天
        cache.setex(f"url:{short_code}", 604800, original_url)
        return original_url
    
    return None

┌─────────┐                         ┌─────────────┐
│  Client  │──── WebSocket ────────►│   Gateway    │
│   A      │◄─── connection ───────│   Server     │
└─────────┘                         └──────┬───────┘
                                            │
┌─────────┐                         ┌──────▼───────┐
│  Client  │──── WebSocket ────────►│   Gateway    │
│   B      │◄─── connection ───────│   Server     │
└─────────┘                         └──────┬───────┘
                                            │
                              ┌─────────────┼─────────────┐
                              │             │             │
                     ┌────────▼───┐  ┌──────▼────┐ ┌─────▼──────┐
                     │   Kafka    │  │   Redis   │ │ PostgreSQL │
                     │  (消息队列)  │  │ (在线状态)  │ │ (持久化存储) │
                     └────────────┘  └───────────┘ └────────────┘

// 客户端消息确认机制
interface ChatMessage {
  id: string;           // 客户端生成的唯一 ID
  senderId: string;
  receiverId: string;
  content: string;
  timestamp: number;
  sequenceNo: number;   // 会话级递增序列号
}

class ChatClient {
  private pendingMessages: Map<string, ChatMessage> = new Map();
  private lastAckedSeq: Map<string, number> = new Map();

  async sendMessage(message: ChatMessage): Promise<void> {
    // 1. 乐观地先显示消息
    this.displayMessage(message, 'sending');

    // 2. 存入待确认队列
    this.pendingMessages.set(message.id, message);

    // 3. 通过 WebSocket 发送
    this.ws.send(JSON.stringify({ type: 'message', payload: message }));

    // 4. 设置超时重发（指数退避）
    this.scheduleRetry(message.id, 0);
  }

  private scheduleRetry(messageId: string, attempt: number): void {
    const timeout = Math.min(1000 * Math.pow(2, attempt), 30000);
    setTimeout(() => {
      if (this.pendingMessages.has(messageId)) {
        const msg = this.pendingMessages.get(messageId)!;
        this.ws.send(JSON.stringify({ type: 'message', payload: msg }));
        this.scheduleRetry(messageId, attempt + 1);
      }
    }, timeout);
  }

  handleAck(messageId: string): void {
    const message = this.pendingMessages.get(messageId);
    if (message) {
      this.displayMessage(message, 'sent');
      this.pendingMessages.delete(messageId);
    }
  }
}

消息写入路径（保证不丢）：

Client → Gateway → Kafka → Message Service → PostgreSQL + 推送
                │                                      │
                └── 持久化到 Kafka 即返回 ACK ◄────────┘
                    异步写入 DB + 推送给接收方

优势：
- Kafka 保证消息持久化和顺序
- 解耦网关与存储层
- 支持消息回放和多个消费者

import hashlib

class ConsistentHash:
    def __init__(self, servers: list[str], virtual_nodes: int = 150):
        self.virtual_nodes = virtual_nodes
        self.ring: dict[int, str] = {}
        self.sorted_keys: list[int] = []
        
        for server in servers:
            self._add_server(server)
    
    def _hash(self, key: str) -> int:
        return int(hashlib.md5(key.encode()).hexdigest(), 16)
    
    def _add_server(self, server: str):
        for i in range(self.virtual_nodes):
            node_key = f"{server}:vnode:{i}"
            hash_val = self._hash(node_key)
            self.ring[hash_val] = server
            self.sorted_keys.append(hash_val)
        self.sorted_keys.sort()
    
    def get_server(self, user_id: str) -> str:
        """查找用户连接应该路由到哪个服务器"""
        hash_val = self._hash(user_id)
        
        for key in self.sorted_keys:
            if hash_val <= key:
                return self.ring[key]
        
        return self.ring[self.sorted_keys[0]]

import time
import asyncio
from dataclasses import dataclass

@dataclass
class TokenBucket:
    capacity: int          # 桶容量
    refill_rate: float     # 每秒填充令牌数
    tokens: float = 0.0
    last_refill: float = 0.0
    
    def __post_init__(self):
        self.tokens = float(self.capacity)
        self.last_refill = time.monotonic()
    
    def _refill(self):
        now = time.monotonic()
        elapsed = now - self.last_refill
        self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate)
        self.last_refill = now
    
    def allow_request(self, tokens: int = 1) -> bool:
        self._refill()
        if self.tokens >= tokens:
            self.tokens -= tokens
            return True
        return False

-- rate_limiter.lua
-- KEYS[1]: 令牌桶 key
-- ARGV[1]: 桶容量
-- ARGV[2]: 每秒填充速率
-- ARGV[3]: 本次请求需要的令牌数

local bucket_key = KEYS[1]
local capacity = tonumber(ARGV[1])
local refill_rate = tonumber(ARGV[2])
local requested = tonumber(ARGV[3])

local now = redis.call('TIME')[1]  -- 秒级时间戳

local bucket = redis.call('HMGET', bucket_key, 'tokens', 'last_refill')
local tokens = tonumber(bucket[1]) or capacity
local last_refill = tonumber(bucket[2]) or now

-- 补充令牌
local elapsed = now - last_refill
if elapsed > 0 then
    tokens = math.min(capacity, tokens + elapsed * refill_rate)
end

-- 判断是否可以放行
if tokens >= requested then
    tokens = tokens - requested
    redis.call('HMSET', bucket_key, 'tokens', tokens, 'last_refill', now)
    redis.call('EXPIRE', bucket_key, 60)
    return 1  -- 放行
else
    redis.call('HMSET', bucket_key, 'tokens', tokens, 'last_refill', now)
    redis.call('EXPIRE', bucket_key, 60)
    return 0  -- 限流
end

import redis.asyncio as redis

class DistributedRateLimiter:
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
        
        # 加载 Lua 脚本
        with open('rate_limiter.lua') as f:
            self.script = self.redis.register_script(f.read())
    
    async def is_allowed(
        self,
        user_id: str,
        endpoint: str,
        capacity: int = 100,
        refill_rate: float = 10.0,
        tokens: int = 1,
    ) -> bool:
        key = f"rate_limit:{user_id}:{endpoint}"
        result = await self.script(
            keys=[key],
            args=[capacity, refill_rate, tokens],
        )
        return result == 1

                  ┌──────────────────┐
                  │   API Gateway    │
                  │  (全局限流: IP级)  │
                  └────────┬─────────┘
                           │
                  ┌────────▼─────────┐
                  │  Service Mesh    │
                  │  (服务限流: 接口级) │
                  └────────┬─────────┘
                           │
                  ┌────────▼─────────┐
                  │  Application     │
                  │  (业务限流: 用户级) │
                  └──────────────────┘