系统设计原则
核心原则
经过三年的实战，我总结出这些系统设计原则。
1. 渐进式演进
原则
不要过度设计，从简单开始，逐步演进。实践
# ❌ 错误：一开始就这样设计
class OverEngineeredSystem:
def __init__(self):
self.load_balancer = LoadBalancer()
self.cache_cluster = CacheCluster()
self.database_sharding = DatabaseSharding()
self.message_queue = MessageQueue()
# ... 复杂的架构
# ✅ 正确：从简单开始
class SimpleSystem:
def __init__(self):
self.database = Database()
# 遇到问题再加复杂度时机
何时引入复杂度：
- 简单方案无法满足需求时
- 性能成为瓶颈时
- 可靠性要求提高时
- 团队规模扩大时2. 服务降级
原则
保证核心功能，必要时牺牲非核心功能。实践
class DegradationManager:
"""降级管理器"""
def handle_with_degradation(self, feature_name):
"""带降级的处理"""
try:
# 尝试正常处理
return self.normal_handler(feature_name)
except Exception as e:
logging.warning(f'{feature_name} failed: {e}')
# 检查是否需要降级
if self.should_degrade(feature_name):
return self.degraded_handler(feature_name)
# 返回友好错误
return {
'error': 'Service temporarily unavailable',
'message': 'Please try again later'
}
def should_degrade(self, feature_name):
"""判断是否应该降级"""
# 检查系统负载
if get_system_load() > 0.9:
return True
# 检查错误率
if get_error_rate() > 0.1:
return True
return False
# 使用示例
@app.route('/api/news')
def handle_news():
result = degradation_manager.handle_with_degradation('news')
if 'error' in result:
return jsonify(result), 503
return jsonify(result)降级策略
功能优先级：
P0：核心功能（用户认证、API 调用）
P1：重要功能（查询、统计）
P2：次要功能（推荐、分析）
P3：可选功能（报表、导出）
降级顺序：
P3 → P2 → P1 → P03. 幂等性设计
原则
同一个操作执行多次，结果与执行一次相同。实践
def create_subscription(user_id, plan_id):
"""创建订阅（幂等）"""
# 生成幂等键
idempotency_key = f'subscription:{user_id}:{plan_id}'
# 检查是否已处理
existing = redis_client.get(idempotency_key)
if existing:
return json.loads(existing)
# 创建订阅
subscription = {
'user_id': user_id,
'plan_id': plan_id,
'status': 'active',
'created_at': datetime.now().isoformat()
}
# 保存到数据库
db.execute(
'INSERT INTO subscriptions (...) VALUES (...)',
subscription
)
# 保存到缓存（24 小时）
redis_client.setex(
idempotency_key,
86400,
json.dumps(subscription)
)
return subscription
# 即使重复调用，结果也相同4. 故障隔离
原则
单点故障不影响整体。实践
class IsolatedServiceCaller:
"""隔离的服务调用器"""
def __init__(self):
self.executors = {
'service_a': ThreadPoolExecutor(max_workers=5),
'service_b': ThreadPoolExecutor(max_workers=5),
'service_c': ThreadPoolExecutor(max_workers=5)
}
def call_service(self, service_name, func, *args, **kwargs):
"""调用服务（隔离）"""
executor = self.executors.get(service_name)
if not executor:
return func(*args, **kwargs)
try:
future = executor.submit(func, *args, **kwargs)
result = future.result(timeout=5)
return result
except TimeoutError:
logging.error(f'Service {service_name} timeout')
return {'error': 'Service timeout'}
except Exception as e:
logging.error(f'Service {service_name} error: {e}')
return {'error': str(e)}
# 使用
caller = IsolatedServiceCaller()
# service_a 失败不影响 service_b
result_a = caller.call_service('service_a', function_a)
result_b = caller.call_service('service_b', function_b)5. 监控先行
原则
没有监控的系统是盲目的。必须监控的指标
class SystemMetrics:
"""系统指标"""
def collect_all_metrics(self):
"""收集所有指标"""
return {
# 性能指标
'performance': {
'qps': self.get_qps(),
'latency_p50': self.get_latency_percentile(50),
'latency_p99': self.get_latency_percentile(99),
'error_rate': self.get_error_rate()
},
# 资源指标
'resources': {
'cpu_usage': psutil.cpu_percent(),
'memory_usage': psutil.virtual_memory().percent,
'disk_usage': psutil.disk_usage('/').percent,
'network_io': self.get_network_io()
},
# 业务指标
'business': {
'active_users': self.get_active_users(),
'api_calls_today': self.get_api_calls_today(),
'revenue_today': self.get_revenue_today()
},
# 依赖指标
'dependencies': {
'database': self.check_database_health(),
'redis': self.check_redis_health(),
'external_apis': self.check_external_apis_health()
}
}
def alert_if_needed(self, metrics):
"""根据指标决定是否告警"""
# QPS 异常
if metrics['performance']['qps'] > 10000:
send_alert(f'High QPS: {metrics["performance"]["qps"]}')
# 错误率过高
if metrics['performance']['error_rate'] > 0.05:
send_alert(f'High error rate: {metrics["performance"]["error_rate"]}')
# CPU 过高
if metrics['resources']['cpu_usage'] > 90:
send_alert(f'High CPU: {metrics["resources"]["cpu_usage"]}%')
# 数据库不健康
if not metrics['dependencies']['database']:
send_alert('Database unhealthy')6. 简单性优先
原则
能用简单方案解决的，就不要用复杂的。对比
# ❌ 复杂：使用分布式锁
def complex_approach():
lock = distributed_lock.acquire('resource')
try:
# 业务逻辑
do_something()
finally:
lock.release()
# ✅ 简单：使用数据库唯一约束
def simple_approach():
try:
# 直接插入，唯一约束保证不重复
db.execute('INSERT INTO ...')
return True
except IntegrityError:
return False # 已存在7. 容错设计
原则
凡事都要考虑失败的情况。实践
def call_external_api_with_retry(url, max_retries=3):
"""调用外部 API（带重试）"""
for attempt in range(max_retries):
try:
response = requests.get(url, timeout=3)
response.raise_for_status()
return response.json()
except requests.Timeout:
if attempt == max_retries - 1:
raise
time.sleep(2 ** attempt) # 指数退避
except requests.RequestException as e:
if attempt == max_retries - 1:
raise
time.sleep(1)
def call_with_fallback(url):
"""调用 API（带回退）"""
try:
return call_external_api_with_retry(url)
except Exception as e:
logging.error(f'API call failed: {e}')
# 返回缓存数据
cached = redis_client.get(f'cache:{url}')
if cached:
return json.loads(cached)
# 返回默认值
return {'error': 'Service unavailable', 'data': None}8. 数据一致性
原则
根据业务需求选择合适的一致性级别。一致性级别
强一致性：
- 适用：金融交易、库存扣减
- 代价：性能、可用性
- 实现：分布式事务、2PC
最终一致性：
- 适用：社交网络、推荐系统
- 代价：可能读到旧数据
- 实现：异步复制、事件溯源
弱一致性：
- 适用：统计、分析
- 代价：数据可能不准确
- 实现：定期同步总结
系统设计 checklist
功能性：
✓ 核心功能完整
✓ 边界情况处理
✓ 错误处理
性能：
✓ 响应时间可接受
✓ 并发处理能力
✓ 缓存策略
可靠性：
✓ 故障自动恢复
✓ 数据备份
✓ 容灾方案
可维护性：
✓ 代码清晰
✓ 文档完善
✓ 监控告警
可扩展性：
✓ 模块化设计
✓ 接口规范
✓ 易于修改最后的话
系统设计是一个持续学习和实践的过程。
记住：
- 没有完美的架构，只有合适的架构
- 技术服务于业务
- 简单往往最好
- 监控和日志很重要
继续学习：
- 阅读大厂技术博客
- 研究开源项目
- 参与系统设计讨论
- 实践是最好的老师
🎉 恭喜完成整个课程！
我现在已经掌握了构建一个完整 API 平台所需的系统设计知识。继续学习和实践，我会成为一名优秀的系统设计师！