系统设计面试框架

系统设计在面试中的重要性

# System design is not about a single correct answer.
# Interviewers look for:
eval_criteria = [
    'Ability to clarify requirements before designing',
    'Back-of-envelope capacity estimation',
    'High-level architecture with clear components',
    'Data modelling and storage choice',
    'Handling scalability (10x, 100x load)',
    'Trade-off discussion (consistency vs availability, etc.)',
    'Communication: talking through decisions as you make them',
]
for c in eval_criteria:
    print('-', c)

RADIO 框架概览

RADIO = {
    'R': 'Requirements — What must the system do? What scale?',
    'A': 'API         — Define endpoints/operations the system exposes',
    'D': 'Data Model  — Entities, schemas, storage types',
    'I': 'Infra       — High-level architecture: servers, queues, caches, CDN',
    'O': 'Optimise    — Identify and address bottlenecks, trade-offs',
}
for step, desc in RADIO.items():
    print(f'[{step}] {desc}')

print('\nTiming guide for a 45-min interview:')
print('  R: 5 min | A: 5 min | D: 10 min | I: 15 min | O: 10 min')

步骤 R：明确需求

# Requirements questions for URL Shortener:
functional = [
    'Shorten a given URL to a 7-character alias',
    'Redirect short URL to original URL',
    'Allow custom aliases (optional)',
    'URL expiry (optional)',
    'Analytics: click count per URL (optional)',
]
non_functional = [
    '100 million new URLs per day (write: ~1160/sec)',
    '10:1 read:write ratio => 11,600 redirects/sec',
    'Redirects must be < 100ms p99 latency',
    '99.99% availability (< 1 hr downtime/year)',
    'URLs must be globally accessible',
]
print('Functional:')
for f in functional: print(' -', f)
print('\nNon-functional:')
for nf in non_functional: print(' -', nf)

步骤 R：粗略容量估算

# Back-of-envelope for URL Shortener
writes_per_day = 100_000_000        # 100 million URLs/day
read_write_ratio = 100              # 100:1 read/write
reads_per_day = writes_per_day * read_write_ratio
bytes_per_url = 500                 # url string + metadata
years_to_store = 5

print('=== Capacity Estimation ===')
print(f'Writes/sec:  {writes_per_day / 86400:.0f}')
print(f'Reads/sec:   {reads_per_day / 86400:,.0f}')
print(f'Storage/day: {writes_per_day * bytes_per_url / 1e9:.1f} GB')
print(f'Storage total ({years_to_store}y): {writes_per_day * bytes_per_url * 365 * years_to_store / 1e12:.1f} TB')

cache_hit_rate = 0.80
hot_urls = reads_per_day * (1 - cache_hit_rate)
print(f'\n80% cache hit rate: {cache_hit_rate*100}% of reads from cache')
print(f'DB reads/sec: {hot_urls / 86400:,.0f}')

步骤 A：接口设计

# API Design for URL Shortener
apis = [
    {
        'method': 'POST',
        'path': '/api/v1/shorten',
        'request': '{"long_url": "https://...", "alias": "optional", "expires_at": "optional"}',
        'response': '201 Created: {"short_url": "https://short.ly/abc1234", "alias": "abc1234"}',
    },
    {
        'method': 'GET',
        'path': '/{alias}',
        'request': 'No body',
        'response': '301 Redirect to long_url (or 404 Not Found)',
    },
    {
        'method': 'GET',
        'path': '/api/v1/{alias}/stats',
        'request': 'Optional: date range query params',
        'response': '200 OK: {"clicks": 42000, "unique_visitors": 15000}',
    },
]
for api in apis:
    print(f'{api["method"]} {api["path"]}')
    print(f'  Request:  {api["request"]}')
    print(f'  Response: {api["response"]}')
    print()

步骤 D：数据模型

# Data model for URL Shortener

# Core table (PostgreSQL)
urls_schema = '''
CREATE TABLE urls (
    alias       VARCHAR(16) PRIMARY KEY,   -- e.g., 'abc1234'
    long_url    TEXT NOT NULL,
    user_id     UUID REFERENCES users(id),
    created_at  TIMESTAMP DEFAULT NOW(),
    expires_at  TIMESTAMP,
    click_count BIGINT DEFAULT 0
);
CREATE INDEX ON urls(user_id);
'''

# Cache layer (Redis) for hot reads
redis_schema = '''
alias  =>  long_url       # O(1) GET on cache hit
TTL = 24 hours (or until expiry)
Cache eviction: LRU
'''

print('PostgreSQL schema:')
print(urls_schema)
print('Redis cache:')
print(redis_schema)
print('Storage split: Redis for hot reads (~80%), PostgreSQL for writes and cold reads')

步骤 I：高层次基础设施

# ASCII architecture sketch
architecture = '''
         Clients
            |
       [Load Balancer]
       /             \\
  [Write API]    [Read/Redirect API]
      |                  |  |
  [Alias Gen]       [Redis Cache]
      |                  |
  [PostgreSQL]---->[Read Replicas]

Alias Generation:
- Base62 encoding of auto-incrementing ID
- 62^7 = 3.5 trillion unique URLs (enough for 5 years at 100M/day)
- OR random 7-char Base62 with collision check
'''
print(architecture)
print('Key design decisions:')
print('  - Read/Write split: separate services for scalability')
print('  - Cache-aside pattern: read from Redis, fallback to DB')
print('  - 301 vs 302 redirect: 301 cached by browser (less load), 302 always hits server (analytics)')

步骤 O：优化并处理瓶颈

# Key optimisation discussion points
optimisations = {
    'Read latency': [
        'Redis cluster in multiple regions (CDN-edge caching)',
        '301 redirect for non-tracked URLs (client caches)',
        '302 redirect when click analytics are needed',
    ],
    'Write throughput': [
        'Async writes: accept request, enqueue to Kafka, batch-commit to DB',
        'Ticket server (centralised ID generator) to avoid UUID collision',
        'Or: hash(user_id + timestamp + random) with retry on collision',
    ],
    'Availability': [
        'Multi-AZ PostgreSQL with automatic failover',
        'Redis Sentinel or Redis Cluster for HA cache',
        'Health checks + circuit breaker on each service',
    ],
    'Storage': [
        'Partition urls table by hash(alias) for horizontal scaling',
        'Archive expired URLs to cold storage (S3)',
    ],
}
for area, points in optimisations.items():
    print(f'{area}:')
    for p in points: print(f'  - {p}')
    print()

别名生成：62 进制编码

import string

BASE62_CHARS = string.digits + string.ascii_lowercase + string.ascii_uppercase
BASE = 62

def encode_base62(num):
    if num == 0:
        return BASE62_CHARS[0]
    result = ''
    while num:
        result = BASE62_CHARS[num % BASE] + result
        num //= BASE
    return result

def decode_base62(s):
    result = 0
    for c in s:
        result = result * BASE + BASE62_CHARS.index(c)
    return result

# Generate aliases for IDs 1, 100, 1000, 10^9
for id_ in [1, 100, 1000, 1_000_000, 10**9, 3_521_614_606207]:
    alias = encode_base62(id_)
    decoded = decode_base62(alias)
    print(f'ID {id_:20,} => alias "{alias}" (len={len(alias)}) => decoded={decoded}')

将 RADIO 应用于推特信息流设计

# Fan-out on Write vs Fan-out on Read trade-off
fan_out_strategies = {
    'Fan-out on Write (Push)': {
        'How': 'When user A posts, immediately write to all followers feeds',
        'Pro': 'O(1) feed read — feed is precomputed',
        'Con': 'Celebrities with 10M followers => 10M writes per post; very slow write',
        'Best for': 'Users with few followers (regular users)',
    },
    'Fan-out on Read (Pull)': {
        'How': 'When user reads feed, query followees tweets and merge',
        'Pro': 'Writes are fast (one DB write per tweet)',
        'Con': 'Feed read is slow: must query all N followees',
        'Best for': 'Celebrity accounts (few reads per post)',
    },
    'Hybrid': {
        'How': 'Push for regular users, pull for celebrities',
        'Pro': 'Balances read and write cost',
        'Con': 'More complex implementation',
        'Best for': 'Production systems like Twitter',
    },
}
for strategy, details in fan_out_strategies.items():
    print(f'{strategy}:')
    for k, v in details.items(): print(f'  {k}: {v}')
    print()

系统设计面试中的常见错误

# Checklist: before you stop talking, verify you covered:
checklist = [
    '[ ] Asked clarifying questions about scale and constraints',
    '[ ] Made capacity estimates (RPS, storage, bandwidth)',
    '[ ] Defined the API surface clearly',
    '[ ] Described the data model and storage choices',
    '[ ] Drew a high-level architecture with named components',
    '[ ] Identified the main bottleneck and proposed a solution',
    '[ ] Discussed at least one trade-off explicitly',
    '[ ] Verified the design meets the stated requirements',
]
for item in checklist:
    print(item)

快速测验

课程回顾