系统设计基础 System Design Basics
系统设计基础 System Design Basics
系统设计是软件工程中的重要组成部分,涉及架构设计、可扩展性、可靠性等多个方面。本指南涵盖系统设计的基本原则和常见实践。
设计原则 Design Principles
1. SOLID 原则
- Single Responsibility (单一职责)
- 一个类只负责一个功能领域中的相应职责
- Open-Closed (开闭原则)
- 对扩展开放,对修改关闭
- Liskov Substitution (里氏替换)
- 子类能够替换父类并出现在父类能够出现的任何地方
- Interface Segregation (接口隔离)
- 使用多个专门的接口比使用单一的总接口要好
- Dependency Inversion (依赖倒置)
- 高层模块不应该依赖低层模块,都应该依赖抽象
2. CAP 理论
- Consistency (一致性)
- 所有节点在同一时间具有相同的数据
- Availability (可用性)
- 保证每个请求不管成功或者失败都有响应
- Partition Tolerance (分区容错性)
- 系统中任意信息的丢失或失败不会影响系统的继续运作
3. 分布式系统原则
- 可扩展性 (Scalability)
- 可用性 (Availability)
- 可靠性 (Reliability)
- 可维护性 (Maintainability)
- 性能 (Performance)
常见组件 Common Components
1. 缓存 Cache
class Cache:
def __init__(self, capacity: int):
self.capacity = capacity
self.cache = {}
self.lru = []
def get(self, key: str) -> str:
if key in self.cache:
self.lru.remove(key)
self.lru.append(key)
return self.cache[key]
return None
def put(self, key: str, value: str) -> None:
if key in self.cache:
self.lru.remove(key)
elif len(self.cache) >= self.capacity:
oldest = self.lru.pop(0)
del self.cache[oldest]
self.cache[key] = value
self.lru.append(key)
缓存策略
- LRU (Least Recently Used)
- LFU (Least Frequently Used)
- FIFO (First In First Out)
- 随机替换 (Random Replacement)
2. 消息队列 Message Queue
class MessageQueue:
def __init__(self):
self.queue = []
self.consumers = []
def publish(self, message: str) -> None:
self.queue.append(message)
for consumer in self.consumers:
consumer.notify(message)
def subscribe(self, consumer) -> None:
self.consumers.append(consumer)
消息模式
- 发布/订阅 (Pub/Sub)
- 点对点 (Point-to-Point)
- 请求/响应 (Request/Reply)
3. 负载均衡 Load Balancer
class LoadBalancer:
def __init__(self):
self.servers = []
self.current = 0
def addServer(self, server: str) -> None:
self.servers.append(server)
def removeServer(self, server: str) -> None:
self.servers.remove(server)
def getServer(self) -> str:
if not self.servers:
return None
server = self.servers[self.current]
self.current = (self.current + 1) % len(self.servers)
return server
负载均衡策略
- 轮询 (Round Robin)
- 加权轮询 (Weighted Round Robin)
- 最少连接 (Least Connection)
- IP哈希 (IP Hash)
数据存储 Data Storage
1. SQL vs NoSQL
SQL 数据库
- 结构化数据
- ACID 事务
- 关系型查询
- 垂直扩展
CREATE TABLE users (
id INT PRIMARY KEY,
name VARCHAR(255),
email VARCHAR(255) UNIQUE
);
NoSQL 数据库
- 非结构化数据
- 高可用性
- 水平扩展
- 灵活模式
{
"id": "123",
"name": "John",
"email": "[email protected]",
"preferences": {
"theme": "dark",
"notifications": true
}
}
2. 分片 Sharding
class ShardedDatabase:
def __init__(self, shard_count: int):
self.shards = [{} for _ in range(shard_count)]
def getShard(self, key: str) -> int:
return hash(key) % len(self.shards)
def put(self, key: str, value: str) -> None:
shard = self.getShard(key)
self.shards[shard][key] = value
def get(self, key: str) -> str:
shard = self.getShard(key)
return self.shards[shard].get(key)
实际案例 Practical Examples
1. 短URL系统 URL Shortener
class URLShortener:
def __init__(self):
self.url_map = {}
self.counter = 0
self.chars = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
def encode(self, longUrl: str) -> str:
self.counter += 1
shortUrl = self._base62Encode(self.counter)
self.url_map[shortUrl] = longUrl
return "http://short.url/" + shortUrl
def decode(self, shortUrl: str) -> str:
code = shortUrl.split("/")[-1]
return self.url_map.get(code)
def _base62Encode(self, num: int) -> str:
if num == 0:
return self.chars[0]
result = []
while num:
num, rem = divmod(num, 62)
result.append(self.chars[rem])
return "".join(result[::-1])
2. 评论系统 Comment System
class Comment:
def __init__(self, id: str, content: str, user_id: str):
self.id = id
self.content = content
self.user_id = user_id
self.replies = []
self.timestamp = time.time()
class CommentSystem:
def __init__(self):
self.comments = {}
self.user_comments = {}
def addComment(self, content: str, user_id: str) -> str:
comment_id = str(uuid.uuid4())
comment = Comment(comment_id, content, user_id)
self.comments[comment_id] = comment
if user_id not in self.user_comments:
self.user_comments[user_id] = []
self.user_comments[user_id].append(comment_id)
return comment_id
def getComments(self, user_id: str) -> List[Comment]:
if user_id not in self.user_comments:
return []
return [self.comments[cid] for cid in self.user_comments[user_id]]
3. 限流器 Rate Limiter
class RateLimiter:
def __init__(self, capacity: int, time_window: int):
self.capacity = capacity # 最大请求数
self.time_window = time_window # 时间窗口(秒)
self.requests = {} # 用户请求历史
def isAllowed(self, user_id: str) -> bool:
current_time = time.time()
# 清理过期的请求记录
if user_id in self.requests:
self.requests[user_id] = [
req_time for req_time in self.requests[user_id]
if current_time - req_time <= self.time_window
]
else:
self.requests[user_id] = []
# 检查是否超过限制
if len(self.requests[user_id]) >= self.capacity:
return False
# 记录新请求
self.requests[user_id].append(current_time)
return True
设计考虑 Design Considerations
1. 可扩展性
- 水平扩展 vs 垂直扩展
- 无状态设计
- 数据分区
- 异步处理
2. 高可用性
- 冗余部署
- 故障转移
- 熔断机制
- 限流保护
3. 性能优化
- 缓存策略
- 数据预加载
- 并发处理
- 数据压缩
4. 安全性
- 认证和授权
- 数据加密
- 输入验证
- 审计日志
最佳实践 Best Practices
- 从简单开始,逐步优化
- 考虑系统的可扩展性
- 做好监控和日志
- 预留容错和降级方案
- 注重代码可维护性
常见陷阱 Common Pitfalls
- 过度设计
- 忽视可扩展性
- 单点故障
- 缺乏监控
- 安全性考虑不足