01 - 系统设计原则¶

目标: 掌握系统设计的基本原则和方法论，能够设计可扩展、高可用的系统

时间: 2-3周

核心原则: 没有最好的设计，只有最合适的设计

🎯 为什么要学系统设计？¶

从程序员到架构师的必经之路¶

Text Only

初级程序员: "这个功能怎么实现？"
中级程序员: "这个模块怎么设计？"
高级程序员: "这个系统怎么架构？"
架构师: "这个业务怎么规划技术方案？"

实际场景¶

Text Only

场景1: 设计一个电商系统
- 需要支持多少并发？
- 数据库如何设计？
- 缓存策略是什么？
- 如何保证数据一致性？
- 如何水平扩展？

场景2: 设计一个即时通讯系统
- 消息如何实时推送？
- 消息存储方案？
- 如何处理离线消息？
- 如何保证消息顺序？

📚 核心设计原则¶

1. SOLID 原则¶

S - 单一职责原则 (Single Responsibility Principle)¶

Python

# ❌ 违反单一职责
class UserManager:
    def create_user(self, user_data):
        # 创建用户
        pass

    def send_email(self, email, message):
        # 发送邮件
        pass

    def generate_report(self):
        # 生成报表
        pass

# ✅ 遵循单一职责
class UserService:
    def create_user(self, user_data):
        pass

class EmailService:
    def send_email(self, email, message):
        pass

class ReportService:
    def generate_report(self):
        pass

O - 开闭原则 (Open/Closed Principle)¶

Python

# ❌ 违反开闭原则
class PaymentProcessor:
    def process(self, payment_type, amount):
        if payment_type == "alipay":
            # 支付宝支付逻辑
            pass
        elif payment_type == "wechat":
            # 微信支付逻辑
            pass
        elif payment_type == "credit_card":
            # 信用卡支付逻辑
            pass

# ✅ 遵循开闭原则
from abc import ABC, abstractmethod

class PaymentMethod(ABC):
    @abstractmethod
    def pay(self, amount):
        pass

class AlipayPayment(PaymentMethod):
    def pay(self, amount):
        print(f"支付宝支付: {amount}")

class WechatPayment(PaymentMethod):
    def pay(self, amount):
        print(f"微信支付: {amount}")

class PaymentProcessor:
    def __init__(self, payment_method: PaymentMethod):
        self.payment_method = payment_method

    def process(self, amount):
        self.payment_method.pay(amount)

# 新增支付方式无需修改原有代码
class BitcoinPayment(PaymentMethod):
    def pay(self, amount):
        print(f"比特币支付: {amount}")

L - 里氏替换原则 (Liskov Substitution Principle)¶

Python

# ❌ 违反里氏替换原则
class Rectangle:
    def __init__(self, width, height):
        self._width = width
        self._height = height

    @property  # @property将方法变为属性访问
    def width(self):
        return self._width

    @width.setter
    def width(self, value):
        self._width = value

    @property
    def height(self):
        return self._height

    @height.setter
    def height(self, value):
        self._height = value

    def area(self):
        return self._width * self._height

class Square(Rectangle):
    """正方形是长方形的子类？"""
    def __init__(self, side):
        super().__init__(side, side)

    @Rectangle.width.setter
    def width(self, value):
        self._width = value
        self._height = value  # 违反里氏替换！

    @Rectangle.height.setter
    def height(self, value):
        self._width = value
        self._height = value

# 使用方期望设置width不影响height
def test_rectangle(rect):
    rect.width = 5
    rect.height = 10
    assert rect.area() == 50  # Square会失败！

# ✅ 正确的设计
class Shape(ABC):
    @abstractmethod
    def area(self):
        pass

class Rectangle(Shape):
    def __init__(self, width, height):
        self.width = width
        self.height = height

    def area(self):
        return self.width * self.height

class Square(Shape):
    def __init__(self, side):
        self.side = side

    def area(self):
        return self.side ** 2

I - 接口隔离原则 (Interface Segregation Principle)¶

Python

# ❌ 违反接口隔离原则
class Worker(ABC):
    @abstractmethod
    def work(self):
        pass

    @abstractmethod
    def eat(self):
        pass

class Human(Worker):
    def work(self):
        print("Human working")

    def eat(self):
        print("Human eating")

class Robot(Worker):
    def work(self):
        print("Robot working")

    def eat(self):
        raise NotImplementedError("Robot doesn't eat")  # 被迫实现不需要的方法

# ✅ 遵循接口隔离原则
class Workable(ABC):
    @abstractmethod
    def work(self):
        pass

class Eatable(ABC):
    @abstractmethod
    def eat(self):
        pass

class Human(Workable, Eatable):
    def work(self):
        print("Human working")

    def eat(self):
        print("Human eating")

class Robot(Workable):
    def work(self):
        print("Robot working")

D - 依赖倒置原则 (Dependency Inversion Principle)¶

Python

# ❌ 违反依赖倒置原则
class MySQLDatabase:
    def connect(self):
        print("Connecting to MySQL")

    def query(self, sql):
        print(f"Executing: {sql}")

class UserRepository:
    def __init__(self):
        self.db = MySQLDatabase()  # 直接依赖具体实现

    def get_user(self, user_id):
        self.db.connect()
        return self.db.query(f"SELECT * FROM users WHERE id = {user_id}")

# ✅ 遵循依赖倒置原则
class Database(ABC):
    @abstractmethod
    def connect(self):
        pass

    @abstractmethod
    def query(self, sql):
        pass

class MySQLDatabase(Database):
    def connect(self):
        print("Connecting to MySQL")

    def query(self, sql):
        print(f"MySQL executing: {sql}")

class PostgreSQLDatabase(Database):
    def connect(self):
        print("Connecting to PostgreSQL")

    def query(self, sql):
        print(f"PostgreSQL executing: {sql}")

class UserRepository:
    def __init__(self, db: Database):  # 依赖抽象
        self.db = db

    def get_user(self, user_id):
        self.db.connect()
        return self.db.query(f"SELECT * FROM users WHERE id = {user_id}")

# 使用
mysql_repo = UserRepository(MySQLDatabase())
pg_repo = UserRepository(PostgreSQLDatabase())

2. CAP 定理¶

Text Only

CAP定理: 分布式系统无法同时满足一致性、可用性、分区容错性

C - Consistency (一致性): 所有节点数据一致
A - Availability (可用性): 每个请求都能得到响应
P - Partition Tolerance (分区容错性): 网络分区时系统仍能运行

在分布式系统中，P是必须的，所以只能在C和A之间选择

CP系统: 优先保证一致性
- 银行系统
- 库存系统

AP系统: 优先保证可用性
- 社交网络
- 电商网站

3. BASE 理论¶

Text Only

BASE是对CAP中AP的延伸，强调最终一致性

Basically Available (基本可用): 系统出现故障时，允许损失部分可用性
Soft State (软状态): 允许数据存在中间状态
Eventually Consistent (最终一致性): 不保证实时一致，但最终会一致

实现方式:
- 消息队列异步处理
- 读写分离
- 缓存同步

🏗️ 系统设计方法论¶

4S 分析法¶

Text Only

1. Scenario (场景)
   - 功能需求: 需要实现什么功能？
   - 非功能需求: 并发量？数据量？延迟要求？

2. Service (服务)
   - 拆分服务: 按业务领域拆分
   - 服务边界: 明确每个服务的职责

3. Storage (存储)
   - 数据模型: 关系型还是非关系型？
   - 存储方案: 主从？分片？

4. Scale (扩展)
   - 水平扩展: 如何加机器？
   - 负载均衡: 如何分配流量？
   - 容错处理: 故障如何恢复？

实战：设计一个短链接系统¶

Text Only

1. Scenario (场景分析)

   功能需求:
   - 生成短链接
   - 短链接跳转
   - 查看统计

   非功能需求:
   - 日活用户: 1000万
   - 日生成短链: 100万
   - 跳转请求: 1亿/天 ≈ 1160/秒
   - 读多写少 (100:1)

2. Service (服务设计)

   API设计:
   POST /api/v1/shorten
     Request: { "url": "https://example.com/very/long/url" }
     Response: { "short_url": "https://short.ly/Ab3x9z" }

   GET /:short_code
     302 Redirect to original URL

   GET /api/v1/stats/:short_code
     Response: { "clicks": 1000, "countries": {...} }

3. Storage (存储设计)

   数据量估算:
   - 短码: 6字符 (62^6 ≈ 568亿)
   - 每条记录: 约500字节
   - 5年数据: 100万 * 365 * 5 * 500B ≈ 912GB

   数据库选择:
   - 写: MySQL/PostgreSQL (主从复制)
   - 读: Redis (缓存热点数据)
   - 统计: ClickHouse/Elasticsearch

   表结构:
   CREATE TABLE short_urls (
       id BIGINT PRIMARY KEY AUTO_INCREMENT,
       short_code VARCHAR(10) UNIQUE NOT NULL,
       original_url VARCHAR(2048) NOT NULL,
       created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
       expires_at TIMESTAMP NULL,
       user_id BIGINT,
       INDEX idx_short_code (short_code),
       INDEX idx_created_at (created_at)
   );

4. Scale (扩展设计)

   短码生成策略:
   - 方案1: 自增ID转62进制 (简单但可预测)
   - 方案2: 哈希算法 (可能有冲突)
   - 方案3: 预生成 + 发号器 (推荐)

   架构图:

   User → CDN → Load Balancer → API Gateway
                                     ↓
                              ┌──────┴──────┐
                              ↓             ↓
                        Shorten Service   Redirect Service
                              ↓             ↓
                        ID Generator    Redis Cache
                              ↓             ↓
                        MySQL Master    MySQL Slave

🎯 设计模式¶

常用设计模式¶

Python

# 1. 单例模式 (Singleton)
class Singleton:
    _instance = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)  # super()调用父类方法
        return cls._instance

# 2. 工厂模式 (Factory)
class Animal(ABC):  # ABC抽象基类；abstractmethod强制子类实现
    @abstractmethod
    def speak(self):
        pass

class Dog(Animal):
    def speak(self):
        return "Woof!"

class Cat(Animal):
    def speak(self):
        return "Meow!"

class AnimalFactory:
    @staticmethod  # @staticmethod静态方法，不需要实例
    def create(animal_type):
        if animal_type == "dog":
            return Dog()
        elif animal_type == "cat":
            return Cat()
        raise ValueError(f"Unknown animal type: {animal_type}")

# 3. 观察者模式 (Observer)
class Subject:
    def __init__(self):
        self._observers = []

    def attach(self, observer):
        self._observers.append(observer)

    def notify(self, message):
        for observer in self._observers:
            observer.update(message)

class Observer:
    def update(self, message):
        print(f"Received: {message}")

# 4. 策略模式 (Strategy)
class PaymentStrategy(ABC):
    @abstractmethod
    def pay(self, amount):
        pass

class CreditCardPayment(PaymentStrategy):
    def pay(self, amount):
        print(f"Paying {amount} using Credit Card")

class PayPalPayment(PaymentStrategy):
    def pay(self, amount):
        print(f"Paying {amount} using PayPal")

class ShoppingCart:
    def __init__(self, strategy: PaymentStrategy):
        self.strategy = strategy

    def checkout(self, amount):
        self.strategy.pay(amount)

✅ 学习检查点¶

理解并能应用SOLID原则
理解CAP定理和BASE理论
掌握4S系统设计方法
能设计简单的分布式系统
熟悉常用设计模式

📚 推荐资源¶

书籍¶

《设计模式：可复用面向对象软件的基础》（GoF）
《领域驱动设计》
《软件架构设计》

在线资源¶

System Design Primer (GitHub)
系统设计面试指南
High Scalability博客

记住：好的架构是演进出来的，不是设计出来的！ 🏗️