软件工程原则是指导软件开发实践的基本准则和哲学理念,它们源于数十年来软件行业的经验教训,帮助开发者构建高质量、可维护、可扩展的软件系统。这些原则不仅适用于代码编写,还贯穿于需求分析、架构设计、测试部署和运维管理的整个软件生命周期。
分离关注点原则由荷兰计算机科学家艾兹格·迪科斯彻(Edsger Dijkstra)在1974年首次提出,其核心思想是:将复杂系统分解为多个独立的关注点(Concern),每个关注点处理系统的一个特定方面,关注点之间通过明确定义的接口进行交互。
关注点是指软件系统中任何引起开发者兴趣或需要处理的概念、功能或责任。例如:
降低复杂度:人类大脑处理复杂信息的能力有限。当一个模块同时处理多个关注点时,开发者需要同时理解多个领域的知识,认知负荷会急剧增加。通过分离关注点,每个模块只负责一个明确的责任,大大降低了理解和维护的难度。
提高可维护性:当需求变化时,分离关注点可以确保变更只影响相关的模块。例如,如果数据库从MySQL迁移到PostgreSQL,只需修改数据持久化层,业务逻辑层无需改动。
促进代码复用:独立的关注点可以更容易地在不同项目或系统的不同部分之间复用。例如,一个通用的日志模块可以被多个业务模块共享。
支持并行开发:不同的团队成员可以独立开发不同的关注点,只要遵循约定的接口,就可以并行工作而不会相互阻塞。
将系统划分为高内聚、低耦合的模块是实现分离关注点的基本手段。每个模块应该有明确的职责边界,只暴露必要的接口。
# 关注点分离示例:业务逻辑与数据访问分离
# 数据访问层(Data Access Layer)
class UserRepository:
"""负责用户数据的持久化操作"""
def __init__(self, db_connection):
self.db = db_connection
def find_by_id(self, user_id: int) -> User:
"""根据ID查询用户"""
query = "SELECT * FROM users WHERE id = %s"
result = self.db.execute(query, (user_id,))
return User(**result) if result else None
def save(self, user: User) -> None:
"""保存用户信息"""
query = """
INSERT INTO users (id, name, email)
VALUES (%s, %s, %s)
ON CONFLICT (id) DO UPDATE
SET name = EXCLUDED.name, email = EXCLUDED.email
"""
self.db.execute(query, (user.id, user.name, user.email))
# 业务逻辑层(Business Logic Layer)
class UserService:
"""负责用户相关的业务逻辑"""
def __init__(self, user_repository: UserRepository):
self.repo = user_repository
def activate_user(self, user_id: int) -> bool:
"""激活用户账户"""
user = self.repo.find_by_id(user_id)
if not user:
raise UserNotFoundException(f"User {user_id} not found")
if user.is_active:
return False # 已经激活
user.activate()
self.repo.save(user)
# 发送激活通知(另一个关注点)
self._send_activation_notification(user)
return True
def _send_activation_notification(self, user: User) -> None:
"""发送激活通知(可以进一步分离到通知服务)"""
pass
分层架构是分离关注点的经典实现模式,常见的三层架构包括:
更复杂的系统可能采用更多层次,如引入应用服务层、领域层、基础设施层等。
AOP提供了一种在运行时动态地将代码织入到现有类和方法中的机制,特别适用于横切关注点(Cross-cutting Concerns)的分离。
横切关注点是那些跨越多个模块的关注点,如日志记录、事务管理、安全控制等。使用AOP,可以将这些关注点从业务逻辑中剥离出来。
# AOP风格示例:使用装饰器分离日志关注点
import functools
import logging
# 日志关注点
def log_execution(logger: logging.Logger):
"""记录方法执行的装饰器"""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
logger.info(f"Executing {func.__name__} with args={args}, kwargs={kwargs}")
try:
result = func(*args, **kwargs)
logger.info(f"{func.__name__} completed successfully")
return result
except Exception as e:
logger.error(f"{func.__name__} failed: {e}")
raise
return wrapper
return decorator
# 业务逻辑(不包含日志代码)
class OrderService:
def __init__(self):
self.logger = logging.getLogger(__name__)
@log_execution(self.logger)
def process_order(self, order_id: str) -> OrderResult:
"""处理订单(只关注业务逻辑)"""
# 纯业务逻辑,没有日志代码
order = self._load_order(order_id)
self._validate_order(order)
self._calculate_total(order)
self._save_order(order)
return OrderResult(success=True, order=order)
过度分离:将系统划分为过多的微小模块会导致模块间通信开销增加,反而降低系统可维护性。应该在"单一职责"和"合理粒度"之间找到平衡。
错误的关注点边界:如果关注点划分不合理,可能导致变更时需要修改多个模块。例如,将用户认证逻辑分散在多个服务中,而不是集中管理。
忽视性能影响:某些关注点分离技术(如AOP、代理模式)可能引入运行时开销。在性能敏感的场景需要权衡。
单一职责原则由罗伯特·C·马丁(Robert C. Martin,又称Uncle Bob)提出,是SOLID原则中的第一个原则。其定义是:
一个类应该只有一个引起它变化的原因。
这里的"原因"指的是需求变化的来源。如果一个类承担了多个职责,当其中一个职责的需求发生变化时,可能会影响到其他职责的实现,导致代码脆弱、难以维护。
识别类的职责需要理解变化轴线(Axis of Change)。变化轴线是指系统中可能独立变化的不同维度。例如:
如果两个变化轴线独立变化,它们应该由不同的类来负责。
class Employee:
"""员工类 - 违反了SRP"""
def __init__(self, name: str, salary: float):
self.name = name
self.salary = salary
# 职责1:员工数据管理
def get_name(self) -> str:
return self.name
def set_name(self, name: str) -> None:
self.name = name
# 职责2:薪资计算
def calculate_pay(self) -> float:
"""计算月薪"""
return self.salary * 0.8 # 扣除20%税费
# 职责3:数据持久化
def save_to_database(self) -> None:
"""保存到数据库"""
import sqlite3
conn = sqlite3.connect('employees.db')
cursor = conn.cursor()
cursor.execute(
"INSERT INTO employees (name, salary) VALUES (?, ?)",
(self.name, self.salary)
)
conn.commit()
conn.close()
# 职责4:报表生成
def generate_report(self) -> str:
"""生成员工报表"""
return f"Employee: {self.name}, Salary: {self.salary}"
这个Employee类有四个职责:
如果数据库 schema 发生变化,需要修改save_to_database方法;如果报表格式要求变化,需要修改generate_report方法。这两个变化互不影响,应该分离到不同的类中。
# 职责1:员工实体(纯数据)
class Employee:
"""员工实体类 - 只负责数据封装"""
def __init__(self, id: int, name: str, base_salary: float):
self.id = id
self.name = name
self.base_salary = base_salary
# 职责2:薪资计算
class SalaryCalculator:
"""薪资计算器 - 负责薪资计算逻辑"""
def __init__(self, tax_rate: float = 0.2):
self.tax_rate = tax_rate
def calculate_net_salary(self, employee: Employee) -> float:
"""计算税后薪资"""
return employee.base_salary * (1 - self.tax_rate)
def calculate_bonus(self, employee: Employee, performance_score: float) -> float:
"""根据绩效计算奖金"""
return employee.base_salary * performance_score * 0.1
# 职责3:数据持久化
class EmployeeRepository:
"""员工数据仓库 - 负责数据持久化"""
def __init__(self, db_connection):
self.db = db_connection
def save(self, employee: Employee) -> None:
"""保存员工"""
self.db.execute(
"INSERT INTO employees (id, name, base_salary) VALUES (?, ?, ?)",
(employee.id, employee.name, employee.base_salary)
)
def find_by_id(self, employee_id: int) -> Optional[Employee]:
"""根据ID查询员工"""
result = self.db.execute(
"SELECT * FROM employees WHERE id = ?", (employee_id,)
)
return Employee(**result) if result else None
# 职责4:报表生成
class EmployeeReportGenerator:
"""员工报表生成器 - 负责报表生成"""
def generate_summary_report(self, employees: List[Employee]) -> str:
"""生成员工汇总报表"""
total_salary = sum(e.base_salary for e in employees)
report_lines = [
"Employee Summary Report",
"=" * 30,
f"Total Employees: {len(employees)}",
f"Total Salary: ${total_salary:,.2f}",
"-" * 30,
]
for emp in employees:
report_lines.append(f" - {emp.name}: ${emp.base_salary:,.2f}")
return "\n".join(report_lines)
def generate_detail_report(self, employee: Employee,
calculator: SalaryCalculator) -> str:
"""生成员工详细报表"""
net_salary = calculator.calculate_net_salary(employee)
return f"""
Employee Detail Report
======================
ID: {employee.id}
Name: {employee.name}
Base Salary: ${employee.base_salary:,.2f}
Net Salary: ${net_salary:,.2f}
======================
"""
SRP不是要求每个类只做"一件事"(那会导致类爆炸),而是要求每个类只有一个变化原因。判断标准:
开闭原则由伯特兰·迈耶(Bertrand Meyer)在1988年提出,是面向对象设计的核心原则之一:
软件实体(类、模块、函数等)应该对扩展开放,对修改关闭。
这意味着当系统需要新功能时,应该通过添加新代码来实现,而不是修改已有代码。
降低风险:修改已有代码可能引入新的bug,特别是在大型系统中,一处修改可能产生连锁反应。通过扩展而非修改,可以保持现有代码的稳定性。
提高可测试性:新功能通过新类实现,可以独立测试,不会影响已有功能的测试用例。
支持演进:系统可以随着需求变化而演进,而不会变得越来越脆弱。
抽象是实现OCP的基础。通过定义抽象接口,具体实现可以独立变化。
from abc import ABC, abstractmethod
from typing import List
# 抽象接口(稳定)
class PaymentProcessor(ABC):
"""支付处理器抽象"""
@abstractmethod
def process_payment(self, amount: float, currency: str) -> PaymentResult:
"""处理支付"""
pass
@abstractmethod
def refund(self, transaction_id: str) -> RefundResult:
"""处理退款"""
pass
# 具体实现1:信用卡支付(扩展)
class CreditCardProcessor(PaymentProcessor):
"""信用卡支付处理器"""
def __init__(self, api_key: str):
self.api_key = api_key
def process_payment(self, amount: float, currency: str) -> PaymentResult:
# 调用信用卡支付API
return PaymentResult(success=True, transaction_id="cc_123")
def refund(self, transaction_id: str) -> RefundResult:
# 调用信用卡退款API
return RefundResult(success=True)
# 具体实现2:支付宝支付(扩展)
class AlipayProcessor(PaymentProcessor):
"""支付宝支付处理器"""
def __init__(self, app_id: str, private_key: str):
self.app_id = app_id
self.private_key = private_key
def process_payment(self, amount: float, currency: str) -> PaymentResult:
# 调用支付宝API
return PaymentResult(success=True, transaction_id="alipay_456")
def refund(self, transaction_id: str) -> RefundResult:
# 调用支付宝退款API
return RefundResult(success=True)
# 具体实现3:微信支付(扩展)
class WeChatPayProcessor(PaymentProcessor):
"""微信支付处理器"""
def __init__(self, mch_id: str, app_id: str):
self.mch_id = mch_id
self.app_id = app_id
def process_payment(self, amount: float, currency: str) -> PaymentResult:
# 调用微信支付API
return PaymentResult(success=True, transaction_id="wx_789")
def refund(self, transaction_id: str) -> RefundResult:
# 调用微信退款API
return RefundResult(success=True)
# 使用方(稳定)
class PaymentService:
"""支付服务 - 不需要修改就能支持新的支付方式"""
def __init__(self):
self.processors: dict[str, PaymentProcessor] = {}
def register_processor(self, name: str, processor: PaymentProcessor) -> None:
"""注册支付处理器"""
self.processors[name] = processor
def pay(self, method: str, amount: float, currency: str) -> PaymentResult:
"""执行支付"""
processor = self.processors.get(method)
if not processor:
raise ValueError(f"Unsupported payment method: {method}")
return processor.process_payment(amount, currency)
当需要添加新的支付方式(如PayPal)时,只需创建新的类实现PaymentProcessor接口,无需修改PaymentService或已有支付处理器。
策略模式是实现OCP的经典设计模式,它将算法族分别封装起来,让它们可以互相替换。
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import List
@dataclass
class DiscountContext:
"""折扣计算上下文"""
original_price: float
user_type: str
membership_level: int
purchase_history: List[dict]
# 策略接口
class DiscountStrategy(ABC):
"""折扣策略接口"""
@abstractmethod
def calculate_discount(self, context: DiscountContext) -> float:
"""计算折扣金额"""
pass
@abstractmethod
def is_applicable(self, context: DiscountContext) -> bool:
"""判断该策略是否适用于当前上下文"""
pass
# 具体策略1:新用户折扣
class NewUserDiscountStrategy(DiscountStrategy):
"""新用户首单8折"""
def is_applicable(self, context: DiscountContext) -> bool:
return len(context.purchase_history) == 0
def calculate_discount(self, context: DiscountContext) -> float:
return context.original_price * 0.2
# 具体策略2:会员等级折扣
class MembershipDiscountStrategy(DiscountStrategy):
"""根据会员等级打折"""
DISCOUNT_RATES = {1: 0.05, 2: 0.1, 3: 0.15, 4: 0.2, 5: 0.25}
def is_applicable(self, context: DiscountContext) -> bool:
return context.membership_level > 0
def calculate_discount(self, context: DiscountContext) -> float:
rate = self.DISCOUNT_RATES.get(context.membership_level, 0)
return context.original_price * rate
# 具体策略3:满减折扣
class ThresholdDiscountStrategy(DiscountStrategy):
"""满100减20,满200减50"""
THRESHOLDS = [(200, 50), (100, 20)]
def is_applicable(self, context: DiscountContext) -> bool:
return context.original_price >= 100
def calculate_discount(self, context: DiscountContext) -> float:
for threshold, discount in self.THRESHOLDS:
if context.original_price >= threshold:
return discount
return 0
# 上下文类
class DiscountCalculator:
"""折扣计算器 - 支持动态添加策略"""
def __init__(self):
self.strategies: List[DiscountStrategy] = []
def add_strategy(self, strategy: DiscountStrategy) -> None:
"""添加折扣策略(扩展点)"""
self.strategies.append(strategy)
def calculate_final_price(self, context: DiscountContext) -> tuple[float, List[str]]:
"""
计算最终价格
返回: (最终价格, 应用的折扣列表)
"""
total_discount = 0
applied_discounts = []
for strategy in self.strategies:
if strategy.is_applicable(context):
discount = strategy.calculate_discount(context)
total_discount += discount
applied_discounts.append(f"{strategy.__class__.__name__}: -${discount:.2f}")
final_price = max(0, context.original_price - total_discount)
return final_price, applied_discounts
预测变化:识别系统中可能变化的部分,设计抽象来隔离这些变化。但不要过度设计,只为确实可能变化的部分引入抽象。
依赖抽象:高层模块不应该依赖低层模块,两者都应该依赖抽象。
使用组合而非继承:继承是强耦合的,组合更加灵活。通过组合,可以在运行时动态地改变行为。
里氏替换原则由芭芭拉·利斯科夫(Barbara Liskov)在1987年提出:
如果S是T的子类型,那么程序中所有使用T类型对象的地方,都可以替换为S类型对象,而不会改变程序的正确性。
简单来说,子类应该能够完全替代父类,而不需要修改使用父类的代码。
子类重写了父类的方法,但增加了更强的前置条件。
# 反例:违反LSP
class Rectangle:
"""矩形"""
def __init__(self, width: float, height: float):
self._width = width
self._height = height
@property
def width(self) -> float:
return self._width
@width.setter
def width(self, value: float) -> None:
self._width = value
@property
def height(self) -> float:
return self._height
@height.setter
def height(self, value: float) -> None:
self._height = value
def area(self) -> float:
return self._width * self._height
class Square(Rectangle):
"""正方形 - 错误地继承自Rectangle"""
def __init__(self, side: float):
super().__init__(side, side)
@Rectangle.width.setter
def width(self, value: float) -> None:
"""违反LSP:修改width时同时修改height"""
self._width = value
self._height = value # 这是问题所在!
@Rectangle.height.setter
def height(self, value: float) -> None:
"""违反LSP:修改height时同时修改width"""
self._width = value
self._height = value
# 使用代码(期望Rectangle的行为)
def resize_rectangle(rect: Rectangle, new_width: float) -> None:
"""调整矩形宽度,期望高度保持不变"""
original_height = rect.height
rect.width = new_width
# 期望: rect.height == original_height
assert rect.height == original_height, "Height should not change!"
# 测试
rect = Rectangle(4, 5)
resize_rectangle(rect, 6) # 通过
square = Square(5)
resize_rectangle(square, 6) # 失败!AssertionError
正方形虽然是几何意义上的矩形,但在面向对象设计中,它不应该继承自矩形,因为它们的行为契约不同。
子类承诺的比父类少,导致调用者期望的某些保证不再成立。
# 反例:后置条件弱化
class Account:
"""账户基类"""
def __init__(self, balance: float):
self._balance = balance
def withdraw(self, amount: float) -> float:
"""
取款
后置条件:返回的余额 >= 0
"""
if amount > self._balance:
raise InsufficientFundsException()
self._balance -= amount
return self._balance # 保证 >= 0
class OverdraftAccount(Account):
"""透支账户 - 允许负余额"""
def __init__(self, balance: float, overdraft_limit: float):
super().__init__(balance)
self._overdraft_limit = overdraft_limit
def withdraw(self, amount: float) -> float:
"""违反LSP:允许余额为负"""
if amount > self._balance + self._overdraft_limit:
raise InsufficientFundsException()
self._balance -= amount
return self._balance # 可能 < 0,违反父类后置条件
正确的做法是识别真正的继承关系,或者使用组合替代继承。
from abc import ABC, abstractmethod
from dataclasses import dataclass
# 抽象接口
@dataclass
class Shape(ABC):
"""形状抽象"""
@abstractmethod
def area(self) -> float:
pass
@abstractmethod
def perimeter(self) -> float:
pass
@dataclass
class Rectangle(Shape):
"""矩形"""
width: float
height: float
def area(self) -> float:
return self.width * self.height
def perimeter(self) -> float:
return 2 * (self.width + self.height)
@dataclass
class Square(Shape):
"""正方形 - 正确的设计,不继承Rectangle"""
side: float
def area(self) -> float:
return self.side * self.side
def perimeter(self) -> float:
return 4 * self.side
# 工厂方法创建
class ShapeFactory:
@staticmethod
def create_rectangle(width: float, height: float) -> Rectangle:
return Rectangle(width, height)
@staticmethod
def create_square(side: float) -> Square:
return Square(side)
契约式设计(Design by Contract)是实现LSP的有效方法。每个方法都有:
子类可以:
子类不可以:
接口隔离原则由罗伯特·C·马丁提出:
客户端不应该被迫依赖它们不使用的方法。
换句话说,应该将大接口拆分为多个小接口,每个接口只包含相关的方法。
当一个接口包含太多方法时,实现类被迫实现所有方法,即使某些方法对它没有意义。
# 反例:胖接口
class MultiFunctionDevice(ABC):
"""多功能设备接口 - 包含太多方法"""
@abstractmethod
def print(self, document: str) -> None:
pass
@abstractmethod
def scan(self, document: str) -> str:
pass
@abstractmethod
def fax(self, document: str, number: str) -> None:
pass
@abstractmethod
def copy(self, document: str) -> str:
pass
# 简单打印机被迫实现所有方法
class SimplePrinter(MultiFunctionDevice):
"""简单打印机 - 只能打印"""
def print(self, document: str) -> None:
print(f"Printing: {document}")
def scan(self, document: str) -> str:
raise NotImplementedError("Simple printer cannot scan")
def fax(self, document: str, number: str) -> None:
raise NotImplementedError("Simple printer cannot fax")
def copy(self, document: str) -> str:
raise NotImplementedError("Simple printer cannot copy")
将胖接口拆分为多个小接口:
from abc import ABC, abstractmethod
# 小接口1:打印
class Printer(ABC):
@abstractmethod
def print(self, document: str) -> None:
pass
# 小接口2:扫描
class Scanner(ABC):
@abstractmethod
def scan(self, document: str) -> str:
pass
# 小接口3:传真
class Fax(ABC):
@abstractmethod
def fax(self, document: str, number: str) -> None:
pass
# 小接口4:复印
class Copier(ABC):
@abstractmethod
def copy(self, document: str) -> str:
pass
# 简单打印机只实现Printer接口
class SimplePrinter(Printer):
def print(self, document: str) -> None:
print(f"Printing: {document}")
# 多功能打印机实现多个接口
class MultiFunctionPrinter(Printer, Scanner, Fax, Copier):
def print(self, document: str) -> None:
print(f"Printing: {document}")
def scan(self, document: str) -> str:
print(f"Scanning: {document}")
return f"scanned_{document}"
def fax(self, document: str, number: str) -> None:
print(f"Faxing {document} to {number}")
def copy(self, document: str) -> str:
print(f"Copying: {document}")
return f"copy_of_{document}"
接口隔离原则与依赖注入结合使用效果更佳:
from typing import Protocol
class Printable(Protocol):
"""可打印协议"""
def print(self, document: str) -> None: ...
class Scannable(Protocol):
"""可扫描协议"""
def scan(self, document: str) -> str: ...
class DocumentProcessor:
"""文档处理器 - 只依赖需要的接口"""
def __init__(self, printer: Printable, scanner: Scannable):
self._printer = printer
self._scanner = scanner
def copy_document(self, document: str) -> None:
"""复印文档"""
scanned = self._scanner.scan(document)
self._printer.print(scanned)
依赖倒置原则是SOLID原则中的最后一个,也是最重要的一个:
高层模块不应该依赖低层模块,两者都应该依赖抽象。
抽象不应该依赖细节,细节应该依赖抽象。
在没有依赖倒置的传统设计中,高层模块直接依赖低层模块:
# 反例:高层依赖低层
class MySQLDatabase:
"""MySQL数据库实现"""
def connect(self): pass
def query(self, sql: str): pass
def close(self): pass
class UserRepository:
"""用户仓库 - 直接依赖具体实现"""
def __init__(self):
# 硬编码依赖MySQLDatabase
self._db = MySQLDatabase()
def get_user(self, user_id: int):
self._db.connect()
result = self._db.query(f"SELECT * FROM users WHERE id = {user_id}")
self._db.close()
return result
这种设计的问题:
UserRepository无法独立于MySQLDatabase进行单元测试UserRepositoryUserRepository的初始化与MySQLDatabase的创建耦合from abc import ABC, abstractmethod
from typing import Optional
# 抽象(稳定)
class DatabaseConnection(ABC):
"""数据库连接抽象"""
@abstractmethod
def connect(self) -> None:
pass
@abstractmethod
def query(self, sql: str, params: tuple = ()) -> list:
pass
@abstractmethod
def execute(self, sql: str, params: tuple = ()) -> int:
pass
@abstractmethod
def close(self) -> None:
pass
# 具体实现(细节)
class MySQLConnection(DatabaseConnection):
"""MySQL连接实现"""
def __init__(self, host: str, port: int, database: str,
user: str, password: str):
self._config = {
'host': host, 'port': port, 'database': database,
'user': user, 'password': password
}
self._connection = None
def connect(self) -> None:
import mysql.connector
self._connection = mysql.connector.connect(**self._config)
def query(self, sql: str, params: tuple = ()) -> list:
cursor = self._connection.cursor(dictionary=True)
cursor.execute(sql, params)
return cursor.fetchall()
def execute(self, sql: str, params: tuple = ()) -> int:
cursor = self._connection.cursor()
cursor.execute(sql, params)
self._connection.commit()
return cursor.rowcount
def close(self) -> None:
if self._connection:
self._connection.close()
class PostgreSQLConnection(DatabaseConnection):
"""PostgreSQL连接实现"""
# 类似实现...
pass
# 高层模块依赖抽象
class UserRepository:
"""用户仓库 - 依赖抽象,不依赖具体实现"""
def __init__(self, db: DatabaseConnection):
self._db = db # 依赖注入
def get_user(self, user_id: int) -> Optional[dict]:
self._db.connect()
try:
results = self._db.query(
"SELECT * FROM users WHERE id = %s",
(user_id,)
)
return results[0] if results else None
finally:
self._db.close()
def save_user(self, user: dict) -> bool:
self._db.connect()
try:
if user.get('id'):
rows = self._db.execute(
"""UPDATE users SET name = %s, email = %s
WHERE id = %s""",
(user['name'], user['email'], user['id'])
)
else:
rows = self._db.execute(
"""INSERT INTO users (name, email) VALUES (%s, %s)""",
(user['name'], user['email'])
)
return rows > 0
finally:
self._db.close()
# 使用
mysql_db = MySQLConnection("localhost", 3306, "mydb", "user", "pass")
user_repo = UserRepository(mysql_db)
# 可以轻松切换数据库
postgres_db = PostgreSQLConnection("localhost", 5432, "mydb", "user", "pass")
user_repo_pg = UserRepository(postgres_db) # 无需修改UserRepository
在大型应用中,手动管理依赖关系会变得复杂,可以使用依赖注入容器:
class DIContainer:
"""简单的依赖注入容器"""
def __init__(self):
self._registrations = {}
self._singletons = {}
def register(self, interface: type, implementation: type,
lifetime: str = "transient") -> None:
"""注册服务"""
self._registrations[interface] = {
'implementation': implementation,
'lifetime': lifetime
}
def register_instance(self, interface: type, instance: object) -> None:
"""注册单例实例"""
self._singletons[interface] = instance
def resolve(self, interface: type) -> object:
"""解析服务"""
# 检查单例
if interface in self._singletons:
return self._singletons[interface]
registration = self._registrations.get(interface)
if not registration:
raise KeyError(f"No registration for {interface}")
impl_class = registration['implementation']
# 解析构造函数参数
import inspect
sig = inspect.signature(impl_class.__init__)
params = list(sig.parameters.items())[1:] # 跳过self
args = []
for param_name, param in params:
if param.annotation != inspect.Parameter.empty:
# 递归解析依赖
args.append(self.resolve(param.annotation))
else:
raise ValueError(f"Cannot resolve parameter {param_name}")
instance = impl_class(*args)
# 如果是单例,保存实例
if registration['lifetime'] == 'singleton':
self._singletons[interface] = instance
return instance
# 配置容器
container = DIContainer()
container.register(DatabaseConnection, MySQLConnection)
container.register(UserRepository, UserRepository)
# 解析
user_repo = container.resolve(UserRepository)
# 自动注入DatabaseConnection
最小惊讶原则是一个用户体验设计原则,同样适用于软件设计:
组件的行为应该符合大多数用户对它的预期。如果某种行为会让用户感到惊讶,那么应该重新设计。
方法名应该准确描述其行为:
# 反例:命名具有误导性
class ShoppingCart:
def add_item(self, item: str) -> None:
"""这个方法实际上会清空购物车再添加"""
self._items.clear() # 令人惊讶!
self._items.append(item)
# 正例:命名准确
class ShoppingCart:
def add_item(self, item: str) -> None:
"""添加商品到购物车"""
self._items.append(item)
def set_item(self, item: str) -> None:
"""设置购物车唯一商品(清空后添加)"""
self._items.clear()
self._items.append(item)
# 反例:相似方法行为不一致
class StringUtils:
@staticmethod
def is_empty(s: str) -> bool:
"""检查字符串是否为空或None"""
return s is None or len(s) == 0
@staticmethod
def is_blank(s: str) -> bool:
"""检查字符串是否为空白 - 但抛出NPE如果s为None"""
return s.strip() == "" # 令人惊讶:不一致的空值处理
# 正例:行为一致
class StringUtils:
@staticmethod
def is_empty(s: Optional[str]) -> bool:
"""检查字符串是否为空或None"""
return s is None or len(s) == 0
@staticmethod
def is_blank(s: Optional[str]) -> bool:
"""检查字符串是否为空白或None"""
return s is None or s.strip() == ""
相同的操作执行多次应该产生相同的结果:
# 反例:非幂等
class Counter:
def increment(self) -> int:
"""增加计数器 - 但连续调用结果不同"""
self._count += random.randint(1, 10) # 令人惊讶!
return self._count
# 正例:幂等(在相同状态下)
class Counter:
def increment(self, step: int = 1) -> int:
"""按指定步长增加计数器"""
self._count += step
return self._count
# 反例:异常行为不一致
class FileProcessor:
def read_file(self, path: str) -> str:
"""读取文件 - 文件不存在时返回空字符串"""
if not os.path.exists(path):
return "" # 令人惊讶:应该抛出异常
with open(path, 'r') as f:
return f.read()
def write_file(self, path: str, content: str) -> None:
"""写入文件 - 目录不存在时抛出异常"""
with open(path, 'w') as f: # 令人惊讶:与read行为不一致
f.write(content)
# 正例:行为一致
class FileProcessor:
def read_file(self, path: str) -> str:
"""读取文件 - 文件不存在时抛出FileNotFoundError"""
with open(path, 'r') as f:
return f.read()
def write_file(self, path: str, content: str,
create_dirs: bool = False) -> None:
"""写入文件"""
if create_dirs:
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, 'w') as f:
f.write(content)
DRY原则由安迪·亨特(Andy Hunt)和戴夫·托马斯(Dave Thomas)在《程序员修炼之道》中提出:
系统中的每一个知识片段都应该有单一、明确、权威的表示。
# 反例:重复代码
class OrderService:
def process_credit_card_order(self, order: Order) -> None:
# 验证订单
if not order.is_valid():
raise InvalidOrderException()
# 检查库存
if not self.check_inventory(order):
raise OutOfStockException()
# 处理支付
self.charge_credit_card(order)
# 更新库存
self.update_inventory(order)
# 发送确认
self.send_confirmation(order)
def process_paypal_order(self, order: Order) -> None:
# 验证订单 - 重复!
if not order.is_valid():
raise InvalidOrderException()
# 检查库存 - 重复!
if not self.check_inventory(order):
raise OutOfStockException()
# 处理支付
self.charge_paypal(order)
# 更新库存 - 重复!
self.update_inventory(order)
# 发送确认 - 重复!
self.send_confirmation(order)
# 反例:数据重复
class Customer:
def __init__(self, name: str, street: str, city: str,
zip_code: str, country: str):
self.name = name
self.street = street # 地址信息重复
self.city = city
self.zip_code = zip_code
self.country = country
class Order:
def __init__(self, customer_name: str, street: str, city: str,
zip_code: str, country: str):
self.customer_name = customer_name
self.shipping_street = street # 同样的地址信息
self.shipping_city = city
self.shipping_zip = zip_code
self.shipping_country = country
# 正例:提取地址对象
@dataclass
class Address:
street: str
city: str
zip_code: str
country: str
@dataclass
class Customer:
name: str
address: Address
@dataclass
class Order:
customer: Customer
shipping_address: Address # 可以复用Address
# 反例:注释重复代码
class Calculator:
def add(self, a: int, b: int) -> int:
"""
这个方法接受两个整数参数a和b,
然后返回它们的和。
"""
return a + b # 注释只是重复了代码
# 正例:注释解释"为什么"而非"是什么"
class Calculator:
def add(self, a: int, b: int) -> int:
"""
使用饱和加法防止整数溢出。
在金融计算中,溢出应该返回最大值而非抛出异常。
"""
result = a + b
return result if result >= a else sys.maxsize # 饱和处理
class OrderService:
def process_order(self, order: Order, payment_method: str) -> None:
"""统一的订单处理流程"""
self._validate_order(order)
self._reserve_inventory(order)
# 支付方式特定的处理
if payment_method == "credit_card":
self._charge_credit_card(order)
elif payment_method == "paypal":
self._charge_paypal(order)
elif payment_method == "alipay":
self._charge_alipay(order)
self._confirm_inventory(order)
self._send_confirmation(order)
def _validate_order(self, order: Order) -> None:
"""验证订单"""
if not order.is_valid():
raise InvalidOrderException()
def _reserve_inventory(self, order: Order) -> None:
"""预留库存"""
if not self.check_inventory(order):
raise OutOfStockException()
self.hold_inventory(order)
from abc import ABC, abstractmethod
class PaymentProcessor(ABC):
"""支付处理器模板"""
def process(self, order: Order) -> None:
"""模板方法 - 定义算法骨架"""
self._validate_order(order)
self._reserve_inventory(order)
self._process_payment(order) # 子类实现
self._confirm_inventory(order)
self._send_confirmation(order)
def _validate_order(self, order: Order) -> None:
if not order.is_valid():
raise InvalidOrderException()
def _reserve_inventory(self, order: Order) -> None:
if not self.check_inventory(order):
raise OutOfStockException()
self.hold_inventory(order)
@abstractmethod
def _process_payment(self, order: Order) -> None:
"""子类实现具体的支付方式"""
pass
def _confirm_inventory(self, order: Order) -> None:
self.finalize_inventory(order)
def _send_confirmation(self, order: Order) -> None:
self.send_email(order.customer_email, "Order Confirmed")
class CreditCardProcessor(PaymentProcessor):
def _process_payment(self, order: Order) -> None:
self.charge_credit_card(order)
class PayPalProcessor(PaymentProcessor):
def _process_payment(self, order: Order) -> None:
self.charge_paypal(order)
DRY不是绝对原则,过度追求DRY可能导致:
过早抽象:在两个相似但不完全相同的代码片段上强制抽象,导致抽象层难以理解和维护。
错误的抽象:将偶然相似的代码抽象在一起,当它们后续朝不同方向演化时,抽象层成为阻碍。
性能问题:某些情况下,重复代码可能比复杂的抽象更高效。
经验法则:
KISS原则强调简单性:
大多数系统如果保持简单而非复杂,会工作得更好。简单应该是设计的关键目标,应该避免不必要的复杂性。
# 反例:过度工程
class AbstractFactoryFactory:
"""工厂工厂 - 创建工厂的工厂"""
def create_factory(self, factory_type: str) -> AbstractFactory:
if factory_type == "widget":
return WidgetFactory()
elif factory_type == "gadget":
return GadgetFactory()
class AbstractFactory(ABC):
@abstractmethod
def create_product(self) -> AbstractProduct:
pass
class WidgetFactory(AbstractFactory):
def create_product(self) -> AbstractProduct:
return Widget()
# 实际上只需要:
def create_widget() -> Widget:
return Widget()
# 反例:过早优化
class OptimizedCache:
"""复杂的缓存实现,但实际上数据量很小"""
def __init__(self):
self._l1_cache = {}
self._l2_cache = {}
self._eviction_queue = deque()
self._lock = threading.RLock()
def get(self, key: str):
with self._lock:
# 复杂的缓存策略...
pass
# 正例:简单实现
class SimpleCache:
"""简单字典缓存"""
def __init__(self):
self._cache = {}
def get(self, key: str):
return self._cache.get(key)
def put(self, key: str, value):
self._cache[key] = value
# 需求:读取配置文件
# 复杂方案
class ConfigurationManager:
def __init__(self):
self._providers = []
self._cache = {}
self._observers = []
def add_provider(self, provider: ConfigProvider):
self._providers.append(provider)
def get(self, key: str):
if key not in self._cache:
for provider in self._providers:
value = provider.get(key)
if value is not None:
self._cache[key] = value
break
return self._cache.get(key)
# 简单方案
import json
config = json.load(open('config.json'))
db_host = config['database']['host']
# 反例:不必要的接口
class IStringValidator(ABC):
@abstractmethod
def validate(self, s: str) -> bool:
pass
class NonEmptyValidator(IStringValidator):
def validate(self, s: str) -> bool:
return len(s) > 0
# 正例:简单的函数
def is_non_empty(s: str) -> bool:
return len(s) > 0
YAGNI是极限编程(XP)的核心原则:
除非确实需要,否则不要实现某个功能。
# 反例:预测性开发
class UserService:
def __init__(self):
self._cache = RedisCache() # 现在不需要缓存
self._message_queue = KafkaQueue() # 现在不需要队列
self._search_engine = Elasticsearch() # 现在不需要搜索
def create_user(self, user: User) -> None:
self._db.save(user)
self._cache.invalidate(f"user:{user.id}") # 不需要
self._message_queue.publish("user.created", user) # 不需要
self._search_engine.index(user) # 不需要
# 正例:只实现当前需要的
class UserService:
def __init__(self, db: Database):
self._db = db
def create_user(self, user: User) -> None:
self._db.save(user)
# 反例:过度配置
class Logger:
def __init__(self):
self._config = self._load_config()
self._format = self._config.get('format', 'default')
self._color_scheme = self._config.get('color_scheme', 'dark')
self._timestamp_format = self._config.get('timestamp_format', 'ISO')
self._max_line_length = self._config.get('max_line_length', 80)
# ... 20 more config options
# 正例:合理的默认值
class Logger:
def __init__(self, level: str = "INFO", output: str = "stdout"):
self._level = level
self._output = output
YAGNI不意味着完全不考虑未来,而是:
好莱坞原则源于电影行业的"不要打电话给我们,我们会打电话给你":
不要调用我们,我们会调用你。
在软件设计中,这意味着高层组件决定何时、如何调用低层组件,而不是低层组件主动调用高层组件。
# 传统方式:低层调用高层
class LowLevelModule:
def do_something(self):
# 直接依赖高层
HighLevelModule().callback()
# 好莱坞原则:高层控制低层
class HighLevelModule:
def __init__(self):
self._low_level = LowLevelModule(self) # 注入自己
def run(self):
self._low_level.do_something()
def callback(self):
print("High level callback")
class LowLevelModule:
def __init__(self, callback_target):
self._target = callback_target
def do_something(self):
# 通过回调与高层通信
self._target.callback()
class EventBus:
"""事件总线 - 实现好莱坞原则"""
def __init__(self):
self._listeners = defaultdict(list)
def subscribe(self, event_type: str, listener: callable):
"""订阅事件 - 低层注册自己"""
self._listeners[event_type].append(listener)
def publish(self, event_type: str, data: any):
"""发布事件 - 高层决定何时调用"""
for listener in self._listeners[event_type]:
listener(data)
# 使用
bus = EventBus()
# 低层模块订阅事件
class EmailService:
def __init__(self, bus: EventBus):
bus.subscribe("user.registered", self.send_welcome_email)
def send_welcome_email(self, user_data: dict):
print(f"Sending welcome email to {user_data['email']}")
# 高层模块控制流程
class UserRegistration:
def __init__(self, bus: EventBus):
self._bus = bus
def register(self, user_data: dict):
# 保存用户...
# 然后通知所有订阅者
self._bus.publish("user.registered", user_data)
继承是白盒复用,子类了解父类的内部实现细节,导致紧耦合。
# 继承的问题
class Bird:
def fly(self):
print("Flying")
class Penguin(Bird):
"""企鹅是鸟,但不会飞"""
def fly(self):
raise NotImplementedError("Penguins can't fly")
组合是黑盒复用,对象之间通过接口交互,彼此不了解内部实现。
from typing import Protocol
class Flyable(Protocol):
def fly(self) -> None:
pass
class Swimmable(Protocol):
def swim(self) -> None:
pass
class FlyingBehavior:
def fly(self) -> None:
print("Flying with wings")
class SwimmingBehavior:
def swim(self) -> None:
print("Swimming")
class CannotFly:
def fly(self) -> None:
print("Cannot fly")
class Duck:
"""鸭子 - 组合飞行和游泳行为"""
def __init__(self):
self._flying = FlyingBehavior()
self._swimming = SwimmingBehavior()
def fly(self) -> None:
self._flying.fly()
def swim(self) -> None:
self._swimming.swim()
class Penguin:
"""企鹅 - 只组合游泳行为"""
def __init__(self):
self._flying = CannotFly()
self._swimming = SwimmingBehavior()
def fly(self) -> None:
self._flying.fly()
def swim(self) -> None:
self._swimming.swim()
继承适合表达"是一个"(is-a)关系,且满足LSP时:
防御性编程假设程序会面临各种错误情况,主动预防和处理这些问题:
代码应该不信任任何输入,包括自己的内部状态。
from typing import Optional
import re
class UserRegistration:
def register(self, email: str, password: str, age: int) -> None:
# 验证所有输入
if not email or not isinstance(email, str):
raise ValueError("Email is required")
if not self._is_valid_email(email):
raise ValueError(f"Invalid email format: {email}")
if not password or len(password) < 8:
raise ValueError("Password must be at least 8 characters")
if not isinstance(age, int) or age < 13 or age > 120:
raise ValueError("Age must be between 13 and 120")
# 验证通过后才处理
self._create_user(email, password, age)
def _is_valid_email(self, email: str) -> bool:
pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
return re.match(pattern, email) is not None
def calculate_discount(price: float, discount_rate: float) -> float:
"""
计算折扣后价格
前置条件:
- price >= 0
- 0 <= discount_rate <= 1
后置条件:
- 返回值 >= 0
- 返回值 <= price
"""
# 前置条件检查
assert price >= 0, f"Price must be non-negative, got {price}"
assert 0 <= discount_rate <= 1, f"Discount rate must be in [0, 1], got {discount_rate}"
result = price * (1 - discount_rate)
# 后置条件检查
assert result >= 0, f"Result must be non-negative, got {result}"
assert result <= price, f"Result must be <= price, got {result}"
return result
from abc import ABC, abstractmethod
class Logger(ABC):
@abstractmethod
def log(self, message: str) -> None:
pass
class ConsoleLogger(Logger):
def log(self, message: str) -> None:
print(f"[LOG] {message}")
class NullLogger(Logger):
"""空对象 - 避免到处检查 None"""
def log(self, message: str) -> None:
pass # 什么都不做
class Service:
def __init__(self, logger: Optional[Logger] = None):
# 使用NullLogger作为默认值,避免None检查
self._logger = logger or NullLogger()
def do_something(self) -> None:
self._logger.log("Starting operation")
# 业务逻辑...
self._logger.log("Operation completed")
失败快速原则主张问题应该尽早暴露:
如果程序会失败,应该尽快失败,并提供清晰的错误信息。
# 反例:延迟失败
def process_order(order_data: dict) -> None:
# 保存到数据库
order_id = db.save(order_data)
# 发送通知
send_notification(order_data['email'])
# 更新库存 - 可能在这里失败,但前面已经执行了副作用
update_inventory(order_data['items'])
# 最后验证 - 太晚了!
if not order_data.get('items'):
raise ValueError("Order must have items")
# 正例:尽早验证
def process_order(order_data: dict) -> None:
# 首先验证
if not order_data.get('items'):
raise ValueError("Order must have items")
if not order_data.get('email') or '@' not in order_data['email']:
raise ValueError("Valid email is required")
# 验证通过后才执行有副作用的操作
order_id = db.save(order_data)
send_notification(order_data['email'])
update_inventory(order_data['items'])
class Money:
"""货币值对象"""
def __init__(self, amount: float, currency: str):
# 在构造函数中验证,确保对象始终有效
if not isinstance(amount, (int, float)):
raise TypeError(f"Amount must be numeric, got {type(amount)}")
if amount < 0:
raise ValueError(f"Amount cannot be negative, got {amount}")
if not currency or len(currency) != 3:
raise ValueError(f"Currency must be 3-letter code, got {currency}")
self._amount = amount
self._currency = currency.upper()
def add(self, other: 'Money') -> 'Money':
# 业务规则验证
if self._currency != other._currency:
raise ValueError(
f"Cannot add {self._currency} and {other._currency}"
)
return Money(self._amount + other._amount, self._currency)
这些软件工程原则不是孤立的,它们相互关联、相互支持:
不是所有原则都同等重要,建议的优先级:
高优先级:SRP、KISS、DRY、Fail Fast
中优先级:OCP、DIP、LSP、ISP
情境性原则:YAGNI、好莱坞原则、防御性编程
原则是指南,不是法律:
掌握这些原则的建议路径:
软件工程原则提供了一套经过验证的指导方针,帮助开发者构建高质量、可维护的软件系统。但原则的应用需要结合具体场景,在实践中不断体会和调整,最终形成自己的设计判断力。