欢迎来到"Python教程从零基础到实战"系列的第二十二期！
经过了前面二十一期，你已经掌握了 Python 的基础语法、数据结构、面向对象编程、爬虫、数据分析、Web 开发、部署运维和性能优化。但你有没有遇到过这样的代码——
```python
# 一个"功能完善"但结构混乱的配置管理模块
defget_config(key):
"""获取配置"""
# 先读环境变量
val = os.environ.get(key)
if val:
return val
# 再读配置文件
# ...一堆 if/else...
# 再读数据库...
# ...又一堆逻辑...
return default
defset_config(key, value):
# 又要写环境变量、写文件、写数据库...
# 复制粘贴了无数行 get_config 的逻辑
pass
```
这段代码能跑，但维护起来非常痛苦。改一个逻辑要动到处函数，加一个新配置来源要改每一处。问题出在哪？**没有好的架构设计**。
这一期，我们来讲——**设计模式**。不照搬 textbook，我们用真实的 Python 项目来说明，每种模式怎么解决实际问题。
---
## 一、为什么需要设计模式？
### 1.1 设计模式的本质
设计模式不是"必须遵守的规则"，而是**前辈们总结的、经过验证的解决方案模板**。它们回答的问题是：
> "当 A 情况发生时，用 B 方式来解决，能避免 C 问题。"
比如：**多个类需要共享同一个全局状态**——用**单例模式**，比在每次调用时手动判断"我是不是第一个创建的实例"要优雅得多。
### 1.2 Python 的特殊性
和其他语言不同，Python 有几个独特的优势：
-**一切皆对象**：函数也是对象，可以传递、返回、存储
-**动态特性**：可以在运行时修改类的行为
-**装饰器**：天然支持"在不修改源码的前提下增强功能"
这意味着很多在其他语言里需要用复杂套路实现的模式，在 Python 里有更简单的写法。但这不意味着设计模式没用——它们提供的**结构化思路**依然不可替代。
---
## 二、创建型模式：如何优雅地创建对象
### 2.1 单例模式（Singleton）
**场景**：配置管理器、日志器、数据库连接池——整个程序只需要一个实例。
#### 方案一：模块本身就是单例
```python
# config.py — Python 模块导入天然就是单例！
import os
classConfigManager:
"""配置管理器：确保全局只有一个实例"""
_instance = None
def__new__(cls):
ifcls._instance isNone:
print("[Config] 创建新实例")
cls._instance = super().__new__(cls)
cls._instance._initialized = False
returncls._instance
def__init__(self):
ifself._initialized:
return# 已经初始化过了，跳过
self._initialized = True
self._config = {}
self._load_from_env()
self._load_from_file()
def_load_from_env(self):
"""从环境变量加载"""
env_vars = {
"DATABASE_URL": os.getenv("DATABASE_URL"),
"API_KEY": os.getenv("API_KEY"),
"DEBUG": os.getenv("DEBUG", "false").lower() == "true",
}
for key, value in env_vars.items():
if value isnotNone:
self._config[key] = value
def_load_from_file(self):
"""从 YAML 配置文件加载"""
try:
import yaml
withopen("config.yaml", "r", encoding="utf-8") as f:
file_config = yaml.safe_load(f)
if file_config:
self._config.update(file_config)
exceptFileNotFoundError:
pass# 没有配置文件就跳过
exceptImportError:
# yaml 库不存在时使用 JSON
import json
try:
withopen("config.json", "r", encoding="utf-8") as f:
self._config.update(json.load(f))
exceptFileNotFoundError:
pass
defget(self, key, default=None):
returnself._config.get(key, default)
defset(self, key, value):
self._config[key] = value
def__repr__(self):
returnf"ConfigManager(config={dict(self._config)})"
# 使用
cfg1 = ConfigManager()
cfg2 = ConfigManager()
print(cfg1 is cfg2) # True! 它们是同一个实例
cfg1.set("APP_NAME", "MyApp")
print(cfg2.get("APP_NAME")) # "MyApp" — 通过另一个实例也能访问
```
#### 方案二：用装饰器实现
```python
from functools import wraps
defsingleton(cls):
"""装饰器版本：适用于任意类"""
instances = {}
@wraps(cls)
defget_instance(*args, **kwargs):
ifclsnotin instances:
instances[cls] = cls(*args, **kwargs)
return instances[cls]
return get_instance
@singleton
classDatabaseConnection:
def__init__(self, url: str):
self.url = url
print(f"[DB] 连接到 {url}")
defquery(self, sql: str) -> list:
print(f"[DB] 执行 SQL: {sql}")
return [{"id": 1, "name": "Alice"}]
db1 = DatabaseConnection("postgresql://localhost/mydb")
db2 = DatabaseConnection("postgresql://localhost/otherdb") # ← 被忽略！
print(db1.url) # "postgresql://localhost/mydb" — 参数只接受第一次的
```
#### 方案三：Pythonic 的做法（推荐）
```python
# 实际上，对于纯配置场景，最 Pythonic 的方式根本不需要单例：
# 直接用模块级别的变量就行
# settings.py
DATABASE_URL = "postgresql://localhost/mydb"
API_KEY = "sk-xxxxx"
DEBUG = False
# 在其他文件中直接导入
from settings importDATABASE_URL, DEBUG
# 或者作为命名空间
import settings
print(settings.DATABASE_URL)
```
**经验法则**：Python 中，如果功能能用模块级别变量实现，就别用单例。单例的真正战场在——需要"状态"的场景，比如日志器、指标采集器。
---
### 2.2 工厂模式（Factory）
**场景**：需要根据用户输入或配置，动态选择创建哪种类型的对象。
```python
from abc importABC, abstractmethod
import json
import yaml
classFormatter(ABC):
"""输出格式化器的抽象基类"""
@abstractmethod
defformat(self, data: dict) -> str:
pass
classJSONFormatter(Formatter):
defformat(self, data: dict) -> str:
return json.dumps(data, ensure_ascii=False, indent=2)
classCSVFormatter(Formatter):
defformat(self, data: dict) -> str:
ifnot data:
return""
headers = list(data[0].keys())
lines = [",".join(headers)]
for row in data:
lines.append(",".join(str(row[h]) for h in headers))
return"\n".join(lines)
classMarkdownTableFormatter(Formatter):
defformat(self, data: dict) -> str:
ifnot data:
return""
headers = list(data[0].keys())
lines = ["| " + " | ".join(headers) + " |"]
lines.append("| " + " | ".join("---"for _ in headers) + " |")
for row in data:
lines.append("| " + " | ".join(str(row[h]) for h in headers) + " |")
return"\n".join(lines)
# ===== 工厂：负责创建正确的格式化器 =====
classFormatterFactory:
"""集中管理格式化器的创建"""
# 注册表：格式名 -> 格式化器类
_formatters: dict[str, type] = {
"json": JSONFormatter,
"csv": CSVFormatter,
"markdown": MarkdownTableFormatter,
}
@classmethod
defregister(cls, name: str, formatter_class: type):
"""允许外部注册新的格式化器"""
cls._formatters[name] = formatter_class
@classmethod
defcreate(cls, fmt: str) -> Formatter:
"""根据名字创建格式化器"""
if fmt notincls._formatters:
raiseValueError(
f"未知的格式 '{fmt}'，支持的格式: {list(cls._formatters.keys())}"
)
returncls._formatters[fmt]()
# 使用示例
users = [
{"id": 1, "name": "张三", "role": "管理员"},
{"id": 2, "name": "李四", "role": "普通用户"},
]
# 创建不同格式的输出
for fmt_type in ["json", "csv", "markdown"]:
formatter = FormatterFactory.create(fmt_type)
print(f"\n=== {fmt_type.upper()} ===")
print(formatter.format(users))
# 扩展：用户可以注册自己的格式化器
classXMLFormatter(Formatter):
defformat(self, data: dict) -> str:
lines = ["<?xml version='1.0' encoding='utf-8'?>", "<records>"]
for record in data:
line = " <record>"
for key, value in record.items():
line += f"<{key}>{value}</{key}>"
line += "</record>"
lines.append(line)
lines.append("</records>")
return"\n".join(lines)
FormatterFactory.register("xml", XMLFormatter)
xml_formatter = FormatterFactory.create("xml")
print(f"\n=== xml ===\n{xml_formatter.format(users)}")
```
---
### 2.3 建造者模式（Builder）
**场景**：构建一个复杂的 HTTP 请求、构建一份包含多个选项的报告、构建一个数据库查询。
```python
classHttpRequestBuilder:
"""链式调用的 HTTP 请求构建器"""
def__init__(self, method: str, url: str):
self.method = method
self.url = url
self._headers: dict[str, str] = {}
self._params: dict[str, str] = {}
self._body = None
defheader(self, key: str, value: str) -> "HttpRequestBuilder":
"""添加请求头——返回 self 以支持链式调用"""
self._headers[key] = value
returnself# ← 关键：返回自身
defparams(self, **kwargs) -> "HttpRequestBuilder":
"""添加 URL 查询参数"""
self._params.update(kwargs)
returnself
defjson_body(self, data: dict) -> "HttpRequestBuilder":
"""设置 JSON body"""
import json
self._body = json.dumps(data).encode("utf-8")
self.header("Content-Type", "application/json")
returnself
deftext_body(self, text: str) -> "HttpRequestBuilder":
"""设置纯文本 body"""
self._body = text.encode("utf-8")
self.header("Content-Type", "text/plain")
returnself
defbuild(self) -> dict:
"""构建最终请求"""
from urllib.parse import urlencode
request = {
"method": self.method,
"url": self.url,
"headers": self._headers,
"body": self._body,
}
ifself._params:
query_string = urlencode(self._params)
request["url"] += f"?{query_string}"
return request
# ===== 使用示例 =====
# 链式调用构建 GET 请求
get_request = HttpRequestBuilder("GET", "https://api.example.com/users").params(
page=1, per_page=10
).build()
print("GET 请求:", get_request)
# 链式调用构建 POST 请求
post_request = HttpRequestBuilder("POST", "https://api.example.com/users").header(
"Authorization", "Bearer my-token"
).json_body({"name": "王五", "email": "wang@example.com"}).build()
print("\nPOST 请求:", post_request)
# 复杂请求
complex_request = (
HttpRequestBuilder("PUT", "https://api.example.com/users/42")
.header("Authorization", "Bearer my-token")
.header("If-Match", '"v1"')
.json_body({"name": "王五 updated"})
.build()
)
print("\nPUT 请求:", complex_request)
```
**什么时候用建造者？** 当你需要构建的对象有很多可选参数时，建造者比一堆可选参数的构造函数要清晰得多。
---
## 三、结构型模式：如何优雅地组合对象
### 3.1 适配器模式（Adapter）
**场景**：你的代码想用一个 API，但供应商提供了不同的接口。适配器帮你"翻译"。
```python
from abc importABC, abstractmethod
import smtplib
from email.mime.text import MIMEText
classEmailService(ABC):
"""我们的代码所期望的统一邮件服务接口"""
@abstractmethod
defsend(self, to: str, subject: str, body: str) -> bool:
pass
classSMTPAdapter(EmailService):
"""适配器：将统一接口适配到 SMTP 协议"""
def__init__(self, host: str, port: int, username: str, password: str):
self.host = host
self.port = port
self.username = username
self.password = password
defsend(self, to: str, subject: str, body: str) -> bool:
msg = MIMEText(body, "html", "utf-8")
msg["Subject"] = subject
msg["From"] = self.username
msg["To"] = to
try:
server = smtplib.SMTP(self.host, self.port)
server.starttls()
server.login(self.username, self.password)
server.sendmail(self.username, [to], msg.as_string())
server.quit()
returnTrue
exceptExceptionas e:
print(f"SMTP 发送失败: {e}")
returnFalse
classSendGridAdapter(EmailService):
"""适配器：将统一接口适配到 SendGrid API"""
def__init__(self, api_key: str):
self.api_key = api_key
defsend(self, to: str, subject: str, body: str) -> bool:
import requests
data = {
"personalizations": [{"to": [{"email": to}]}],
"from": {"email": "noreply@example.com"},
"subject": subject,
"content": [{"type": "text/html", "value": body}],
}
try:
resp = requests.post(
"https://api.sendgrid.com/v3/mail/send",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
},
json=data,
timeout=10,
)
return resp.status_code == 202
exceptExceptionas e:
print(f"SendGrid 发送失败: {e}")
returnFalse
classAWSSESAdapter(EmailService):
"""适配器：适配 AWS SES"""
def__init__(self, region: str, access_key: str, secret_key: str):
self.region = region
self.access_key = access_key
self.secret_key = secret_key
defsend(self, to: str, subject: str, body: str) -> bool:
try:
import boto3
ses = boto3.client("ses", region_name=self.region)
ses.send_email(
Source=self.access_key + "@example.com",
Destination={"ToAddresses": [to]},
Message={
"Subject": {"Data": subject},
"Body": {"Html": {"Data": body}},
},
)
returnTrue
exceptExceptionas e:
print(f"AWS SES 发送失败: {e}")
returnFalse
# ===== 使用示例：上层业务代码完全不用关心底层实现 =====
defnotify_user(email_service: EmailService, user_email: str, message: str):
"""通知用户——只依赖抽象接口"""
success = email_service.send(user_email, "系统通知", message)
if success:
print(f" ✓ 通知已发送至 {user_email}")
else:
print(f" ✗ 通知发送失败: {user_email}")
# 可以随时切换邮件服务，不需要修改 notify_user 的代码！
smtp_service = SMTPAdapter("smtp.gmail.com", 587, "user@gmail.com", "password")
notify_user(smtp_service, "admin@example.com", "<h1>服务器健康</h1><p>一切正常</p>")
# 切换到 SendGrid
sendgrid_service = SendGridAdapter("SG.xxxxx")
notify_user(sendgrid_service, "admin@example.com", "<h1>服务器健康</h1><p>一切正常</p>")
```
---
### 3.2 装饰器模式（Decorator）
你已经用过 Python 的 `@` 装饰器语法了，但那是**函数装饰器**——它修改的是函数本身。装饰器模式修饰的是**对象的行为**，两者概念不同但思路相通。
```python
from abc importABC, abstractmethod
classCoffee(ABC):
"""咖啡抽象类"""
@abstractmethod
defcost(self) -> float:
pass
@abstractmethod
defdescription(self) -> str:
pass
classSimpleCoffee(Coffee):
"""基础咖啡"""
defcost(self) -> float:
return10.0
defdescription(self) -> str:
return"简易咖啡"
# ===== 装饰器：给咖啡加料 =====
classCoffeeDecorator(Coffee):
"""咖啡装饰器基类"""
def__init__(self, coffee: Coffee):
self._coffee = coffee
defcost(self) -> float:
returnself._coffee.cost()
defdescription(self) -> str:
returnself._coffee.description()
classMilk(CoffeeDecorator):
"""牛奶装饰器"""
defcost(self) -> float:
returnself._coffee.cost() + 3.0
defdescription(self) -> str:
returnf"{self._coffee.description()} + 牛奶"
classWhippedCream(CoffeeDecorator):
"""奶油装饰器"""
defcost(self) -> float:
returnself._coffee.cost() + 2.0
defdescription(self) -> str:
returnf"{self._coffee.description()} + 奶油"
classSugar(CoffeeDecorator):
"""糖装饰器"""
defcost(self) -> float:
returnself._coffee.cost() + 1.0
defdescription(self) -> str:
returnf"{self._coffee.description()} + 糖"
# ===== 使用示例 =====
my_coffee = SimpleCoffee()
print(f"{my_coffee.description()}: ¥{my_coffee.cost():.1f}")
# 简易咖啡: ¥10.0
my_coffee = Milk(SimpleCoffee())
print(f"{my_coffee.description()}: ¥{my_coffee.cost():.1f}")
# 简易咖啡 + 牛奶: ¥13.0
my_coffee = WhippedCream(Milk(SimpleCoffee()))
print(f"{my_coffee.description()}: ¥{my_coffee.cost():.1f}")
# 简易咖啡 + 牛奶 + 奶油: ¥15.0
my_coffee = Sugar(WhippedCream(Milk(SimpleCoffee())))
print(f"{my_coffee.description()}: ¥{my_coffee.cost():.1f}")
# 简易咖啡 + 牛奶 + 奶油 + 糖: ¥16.0
```
Python 的函数装饰器和对象装饰器模式可以结合使用：
```python
from functools import wraps
import time
deftimer(func):
"""计时装饰器——测量函数执行时间"""
@wraps(func)
defwrapper(*args, **kwargs):
start = time.perf_counter()
result = func(*args, **kwargs)
elapsed = time.perf_counter() - start
print(f" [{func.__name__}] 耗时 {elapsed:.4f}s")
return result
return wrapper
defretry(max_retries: int = 3, delay: float = 1.0):
"""重试装饰器——失败时自动重试"""
defdecorator(func):
@wraps(func)
defwrapper(*args, **kwargs):
for attempt inrange(1, max_retries + 1):
try:
return func(*args, **kwargs)
exceptExceptionas e:
if attempt == max_retries:
raise
print(f" ⚠ 第 {attempt} 次尝试失败: {e}，{delay}s 后重试")
time.sleep(delay)
return wrapper
return decorator
# 组合使用
@timer
@retry(max_retries=3, delay=0.1)
defunstable_api_call(request_id: int) -> dict:
"""模拟一个偶尔会失败的 API 调用"""
import random
if random.random() < 0.3:
raiseConnectionError("网络连接不稳定")
return {"request_id": request_id, "data": "success"}
result = unstable_api_call(42)
print(f" 结果: {result}")
```
---
### 3.3 外观模式（Facade）
**场景**：子系统太复杂，提供一个简化的统一入口。
```python
import subprocess
import os
from datetime import datetime
classDockerOperations:
"""Docker 操作门面——把复杂的 docker 命令封装成简单方法"""
def__init__(self, project_dir: str = "."):
self.project_dir = project_dir
defbuild_image(self, image_name: str, tag: str = "latest") -> bool:
"""构建镜像"""
cmd = f'docker build -t "{image_name}:{tag}" .'
print(f" 🔨 构建镜像: {cmd}")
result = subprocess.run(cmd, shell=True, capture_output=True, text=True,
cwd=self.project_dir)
return result.returncode == 0
defrun_container(self, image_name: str, container_name: str,
ports: dict = None, detach: bool = True) -> bool:
"""运行容器"""
cmd = f'docker run'
if detach:
cmd += " -d"
if container_name:
cmd += f' --name "{container_name}"'
if ports:
for host, container in ports.items():
cmd += f' -p {host}:{container}'
cmd += f' "{image_name}:latest"'
print(f" ▶ 启动容器: {cmd}")
result = subprocess.run(cmd, shell=True, capture_output=True, text=True,
cwd=self.project_dir)
return result.returncode == 0
defstop_container(self, container_name: str) -> bool:
"""停止容器"""
cmd = f'docker stop "{container_name}"'
print(f" ⏸ 停止容器: {cmd}")
result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
return result.returncode == 0
defremove_container(self, container_name: str) -> bool:
"""删除容器"""
cmd = f'docker rm "{container_name}"'
print(f" 🗑 删除容器: {cmd}")
result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
return result.returncode == 0
defhealth_check(self, container_name: str) -> dict:
"""检查容器状态"""
import json
cmd = f'docker inspect --format="{{{{json .State}}}}" "{container_name}"'
result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
if result.returncode != 0:
return {"status": "not_found"}
state = json.loads(result.stdout)
return {
"running": state.get("Running", False),
"status": state.get("Status", "unknown"),
"exit_code": state.get("ExitCode", -1),
"started_at": state.get("StartedAt", ""),
}
# ===== 外观：一行搞定整个部署流程 =====
classDeploymentFacade:
"""部署外观——对外只暴露一个 deploy() 方法"""
def__init__(self, app_name: str, project_dir: str = "."):
self.app_name = app_name
self.docker = DockerOperations(project_dir)
self.tag = datetime.now().strftime("%Y%m%d-%H%M%S")
defdeploy(self, restart_if_exists: bool = True) -> bool:
"""一键部署：停旧容器 → 删旧容器 → 构建 → 启动"""
container_name = f"{self.app_name}-app"
# 第一步：清理旧容器
if restart_if_exists:
try:
health = self.docker.health_check(container_name)
if health["running"]:
print(f"\n 📦 发现运行的容器，先停止...")
self.docker.stop_container(container_name)
self.docker.remove_container(container_name)
exceptException:
pass# 容器不存在也没关系
# 第二步：构建新镜像
image_name = f"{self.app_name}"
print(f"\n 🏷 标记新版本: {self.tag}")
ifnotself.docker.build_image(image_name, self.tag):
print(" ❌ 镜像构建失败！")
returnFalse
# 第三步：启动容器
print(f"\n 🚀 启动新版本容器...")
ifnotself.docker.run_container(
image_name, container_name,
ports={8080: 8080, 8081: 8081}
):
print(" ❌ 容器启动失败！")
returnFalse
# 第四步：确认健康
health = self.docker.health_check(container_name)
if health.get("running"):
print(f"\n ✅ 部署成功！容器正在运行")
print(f" 访问 http://localhost:8080")
returnTrue
else:
print(f"\n ⚠️ 容器启动但可能异常: {health}")
returnFalse
# 使用：一行完成部署
facade = DeploymentFacade("my-web-app")
facade.deploy()
```
---
## 四、行为型模式：对象之间如何协作
### 4.1 策略模式（Strategy）
**场景**：同一个操作有多种算法实现，运行时可以选择。比如排序算法、支付渠道、数据导出格式。
```python
from abc importABC, abstractmethod
import csv
import json
import os
classExportStrategy(ABC):
"""导出策略抽象类"""
@abstractmethod
defexport(self, data: list[dict], filepath: str) -> str:
"""导出数据到指定文件"""
pass
@abstractmethod
defextension(self) -> str:
"""返回文件扩展名"""
pass
classJSONExportStrategy(ExportStrategy):
defexport(self, data: list[dict], filepath: str) -> str:
withopen(filepath, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
return filepath
defextension(self) -> str:
return".json"
classCSVExportStrategy(ExportStrategy):
defexport(self, data: list[dict], filepath: str) -> str:
ifnot data:
withopen(filepath, "w", encoding="utf-8") as f:
f.write("")
return filepath
fieldnames = list(data[0].keys())
withopen(filepath, "w", encoding="utf-8", newline="") as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(data)
return filepath
defextension(self) -> str:
return".csv"
classReportGenerator:
"""报告生成器：依赖策略，而非硬编码"""
def__init__(self, strategy: ExportStrategy):
self._strategy = strategy
defchange_strategy(self, strategy: ExportStrategy):
"""运行时切换策略"""
self._strategy = strategy
defgenerate_report(self, data: list[dict], output_dir: str = ".") -> str:
"""生成报告"""
filepath = os.path.join(
output_dir, f"report{self._strategy.extension()}"
)
exported_path = self._strategy.export(data, filepath)
return exported_path
# 使用
sales_data = [
{"product": "iPhone 15", "quantity": 120, "revenue": 899000},
{"product": "MacBook Pro", "quantity": 45, "revenue": 2245000},
{"product": "AirPods Pro", "quantity": 300, "revenue": 359700},
]
# 默认导出为 JSON
generator = ReportGenerator(JSONExportStrategy())
path = generator.generate_report(sales_data)
print(f"JSON 报告: {path}")
# 切换为 CSV
generator.change_strategy(CSVExportStrategy())
path = generator.generate_report(sales_data)
print(f"CSV 报告: {path}")
```
**核心要点**：`ReportGenerator` 完全不关心导出的具体实现。新增一种导出格式只需要创建一个新策略类，不需要改动任何已有代码——符合**开闭原则**（对扩展开放，对修改封闭）。
---
### 4.2 观察者模式（Observer）
**场景**：事件总线、消息推送、UI 刷新、传感器数据订阅。一个对象状态变化时，所有关注者都要收到通知。
```python
classEventBus:
"""简单的事件总线"""
def__init__(self):
self._listeners: dict[str, list] = {}
defsubscribe(self, event_type: str, callback):
"""订阅某个事件"""
if event_type notinself._listeners:
self._listeners[event_type] = []
self._listeners[event_type].append(callback)
print(f" 👂 订阅: {event_type}")
defunsubscribe(self, event_type: str, callback):
"""取消订阅"""
if event_type inself._listeners:
self._listeners[event_type].remove(callback)
defpublish(self, event_type: str, data: dict = None):
"""发布事件"""
if data isNone:
data = {}
if event_type inself._listeners:
for callback inself._listeners[event_type]:
callback(event_type, data)
# ===== 实际场景：电商系统中的事件通知 =====
classOrderSystem:
"""订单系统——发布事件"""
def__init__(self, bus: EventBus):
self.bus = bus
self._orders = {}
self._order_counter = 0
defcreate_order(self, customer: str, items: list, amount: float) -> str:
"""创建订单并发布事件"""
self._order_counter += 1
order_id = f"ORD-{self._order_counter:06d}"
self._orders[order_id] = {
"customer": customer,
"items": items,
"amount": amount,
"status": "pending",
}
# 发布事件
self.bus.publish("order.created", {
"order_id": order_id,
"customer": customer,
"amount": amount,
})
return order_id
classNotificationService:
"""通知服务——监听事件"""
def__init__(self, bus: EventBus):
bus.subscribe("order.created", self.on_order_created)
bus.subscribe("order.paid", self.on_order_paid)
defon_order_created(self, event_type: str, data: dict):
print(f" 📧 [通知服务] 新订单: {data['order_id']} ({data['customer']}, ¥{data['amount']})")
defon_order_paid(self, event_type: str, data: dict):
print(f" 💰 [通知服务] 订单支付成功: {data['order_id']}")
classInventoryService:
"""库存服务——监听事件"""
def__init__(self, bus: EventBus):
bus.subscribe("order.created", self.on_order_created)
self.stock = {"iPhone 15": 100, "MacBook Pro": 50, "AirPods Pro": 200}
defon_order_created(self, event_type: str, data: dict):
total_items = len(data["items"])
print(f" 📦 [库存服务] 新订单占用库存: {total_items} 件商品")
classAnalyticsService:
"""分析服务——监听事件"""
def__init__(self, bus: EventBus):
bus.subscribe("order.created", self.on_order_created)
bus.subscribe("order.paid", self.on_order_paid)
self.stats = {"total_orders": 0, "total_revenue": 0.0}
defon_order_created(self, event_type: str, data: dict):
self.stats["total_orders"] += 1
print(f" 📊 [分析服务] 订单计数更新: 总订单={self.stats['total_orders']}")
defon_order_paid(self, event_type: str, data: dict):
self.stats["total_revenue"] += data.get("amount", 0)
print(f" 📈 [分析服务] 收入统计更新: 总收入=¥{self.stats['total_revenue']:,.2f}")
# ===== 运行 =====
bus = EventBus()
order_system = OrderSystem(bus)
NotificationService(bus)
InventoryService(bus)
AnalyticsService(bus)
print("\n--- 创建订单 ---")
order_system.create_order(
customer="张三",
items=["iPhone 15", "AirPods Pro"],
amount=8590.00,
)
print("\n--- 模拟支付 ---")
bus.publish("order.paid", {"order_id": "ORD-000001", "amount": 8590.00})
```
---
### 4.3 责任链模式（Chain of Responsibility）
**场景**：请求需要经过多道审批/处理步骤。每条链路独立，可以灵活增删。
```python
from abc importABC, abstractmethod
from typing import Optional
classHandler(ABC):
"""处理器抽象类"""
def__init__(self):
self._next_handler: Optional[Handler] = None
defset_next(self, handler: "Handler") -> "Handler":
"""设置下一个处理器，返回它以便链式调用"""
self._next_handler = handler
return handler # 返回下一个处理器本身
defhandle(self, request: dict) -> Optional[dict]:
"""处理请求"""
ifself._can_handle(request):
result = self._process(request)
if result andself._next_handler:
returnself._next_handler.handle(result)
return result
elifself._next_handler:
returnself._next_handler.handle(request)
returnNone
def_can_handle(self, request: dict) -> bool:
"""判断能否处理此请求——子类覆盖"""
returnTrue
@abstractmethod
def_process(self, request: dict) -> Optional[dict]:
"""实际处理逻辑——子类覆盖"""
pass
classAuthHandler(Handler):
"""认证处理器"""
def_can_handle(self, request: dict) -> bool:
return"token"notin request
def_process(self, request: dict) -> Optional[dict]:
token = request.get("token")
if token and token.startswith("valid_"):
request["user"] = "authenticated_user"
print(" 🔐 认证成功")
return request
request["user"] = "anonymous"
print(" 🔐 无有效 Token，标记为匿名用户")
return request
classPermissionHandler(Handler):
"""权限处理器"""
REQUIRED_PERMISSIONS = {
"/api/admin": "admin",
"/api/orders": "reader",
"/api/reports": "analyst",
}
def_can_handle(self, request: dict) -> bool:
return"user"in request
def_process(self, request: dict) -> Optional[dict]:
path = request.get("path", "")
required = self.REQUIRED_PERMISSIONS.get(path)
if required:
print(f" 🔑 路径 {path} 需要权限: {required}")
request["permission"] = required
return request
classRateLimitHandler(Handler):
"""速率限制处理器"""
def__init__(self, max_requests: int = 100):
super().__init__()
self.max_requests = max_requests
self._request_count = 0
def_can_handle(self, request: dict) -> bool:
return"permission"in request
def_process(self, request: dict) -> Optional[dict]:
self._request_count += 1
ifself._request_count > self.max_requests:
print(f" ⏱ 速率限制触发！已超过 {self.max_requests} 次请求")
request["rate_limited"] = True
return request
print(f" ⏱ 请求计数: {self._request_count}/{self.max_requests}")
return request
classLoggingHandler(Handler):
"""日志处理器（放在链条最后）"""
def_process(self, request: dict) -> Optional[dict]:
path = request.get("path", "/")
status = "blocked"if request.get("rate_limited") else"allowed"
user = request.get("user", "unknown")
print(f" 📝 日志: {user} 访问 {path} → {status}")
return request
# ===== 组装责任链 =====
chain = AuthHandler() \
.set_next(PermissionHandler()) \
.set_next(RateLimitHandler(max_requests=3)) \
.set_next(LoggingHandler())
# 测试请求
response = chain.handle({
"path": "/api/orders",
"token": "invalid_token_xyz",
})
print(f"响应: {response}\n")
# 再来几次请求触发速率限制
for i inrange(4):
chain.handle({"path": "/api/orders", "token": "valid_user123"})
```
---
## 五、实战重构：从一个"面条代码"到优雅架构
### 5.1 重构前：一团糟
```python
# utils/bad_analyzer.py — 这是一个典型的"什么都在一个文件里"的反面教材
import os
import json
import sqlite3
from datetime import datetime
defprocess_data(input_file, output_file, config_file="config.json"):
"""这个函数什么都干"""
# 读配置
withopen(config_file, "r") as f:
config = json.load(f)
threshold = config.get("anomaly_threshold", 0.5)
db_path = config.get("database", "analysis.db")
# 读数据
withopen(input_file, "r") as f:
data = json.load(f)
# 处理数据
results = []
for item in data:
score = 0
if item.get("value", 0) > threshold * 100:
score += 0.3
if item.get("weight", 0) > 10:
score += 0.2
if"flagged"in item.get("tags", []):
score += 0.5
result = {
"id": item["id"],
"score": score,
"timestamp": datetime.now().isoformat(),
}
results.append(result)
# 存结果
withopen(output_file, "w") as f:
json.dump(results, f, indent=2)
# 写数据库
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS analysis_results (
id TEXT PRIMARY KEY,
score REAL,
created_at TEXT
)
""")
for r in results:
cursor.execute(
"INSERT OR REPLACE INTO analysis_results VALUES (?, ?, ?)",
(r["id"], r["score"], r["timestamp"]),
)
conn.commit()
conn.close()
return results
```
这个函数的毛病：**它同时做了至少五件事**——读配置、读数据、处理分析、写结果文件、写入数据库。任何一件事的变化都会影响其他部分。
### 5.2 重构后：各司其职
```python
# 第一步：配置加载器（单一职责）
classConfigLoader:
"""负责加载和解析配置"""
DEFAULTS = {
"anomaly_threshold": 0.5,
"database": "analysis.db",
"output_format": "json",
}
def__init__(self, config_path: str = "config.json"):
self.config = dict(self.DEFAULTS)
if os.path.exists(config_path):
withopen(config_path, "r") as f:
self.config.update(json.load(f))
@property
defthreshold(self) -> float:
returnself.config["anomaly_threshold"]
@property
defdb_path(self) -> str:
returnself.config["database"]
# 第二步：数据加载器
classDataLoader:
"""负责从各种源加载数据"""
@staticmethod
defload_json(filepath: str) -> list[dict]:
withopen(filepath, "r", encoding="utf-8") as f:
return json.load(f)
# 第三步：评分引擎（核心算法）
classScoringEngine:
"""计算每项的异常分数"""
def__init__(self, threshold: float):
self.threshold = threshold
defscore_item(self, item: dict) -> dict:
"""给单个物品打分"""
score = 0.0
if item.get("value", 0) > self.threshold * 100:
score += 0.3
if item.get("weight", 0) > 10:
score += 0.2
if"flagged"in item.get("tags", []):
score += 0.5
return {
"id": item["id"],
"score": round(score, 2),
"timestamp": datetime.now().isoformat(),
}
# 第四步：结果存储器
classResultStore:
"""存储分析结果"""
def__init__(self, db_path: str):
self.db_path = db_path
self._init_db()
def_init_db(self):
import sqlite3
conn = sqlite3.connect(self.db_path)
conn.execute("""
CREATE TABLE IF NOT EXISTS analysis_results (
id TEXT PRIMARY KEY,
score REAL,
created_at TEXT
)
""")
conn.commit()
conn.close()
defsave_to_file(self, results: list[dict], filepath: str):
withopen(filepath, "w", encoding="utf-8") as f:
json.dump(results, f, ensure_ascii=False, indent=2)
defsave_to_db(self, results: list[dict]):
import sqlite3
conn = sqlite3.connect(self.db_path)
for r in results:
conn.execute(
"INSERT OR REPLACE INTO analysis_results VALUES (?, ?, ?)",
(r["id"], r["score"], r["timestamp"]),
)
conn.commit()
conn.close()
# 第五步：编排器——把各部分组装起来
classDataAnalyzer:
"""数据分析师——协调各个组件，不做任何具体工作"""
def__init__(self, config_path: str = "config.json"):
self.config = ConfigLoader(config_path)
self.scorer = ScoringEngine(self.config.threshold)
self.store = ResultStore(self.config.db_path)
defanalyze(self, input_file: str, output_file: str) -> list[dict]:
"""分析入口"""
data = DataLoader.load_json(input_file)
results = [self.scorer.score_item(item) for item in data]
self.store.save_to_file(results, output_file)
self.store.save_to_db(results)
return results
```
**重构的收益：**
- 每个类只有**一个存在的理由**（单一职责原则）
- 想换数据存储方式？只需改 `ResultStore`，不动其他代码
- 想改评分规则？只需改 `ScoringEngine`
- 想用不同配置格式？只需换 `ConfigLoader`
- 测试变得超简单：每个类都可以单独单元测试
---
## 六、实操练习
### 练习题
**第 1 题：实现一个简单的缓存装饰器**
基于第 2.1 节学到的装饰器知识，编写一个带 TTL（过期时间）的 LRU 缓存装饰器：
```python
import time
import functools
from typing import Any, Callable
deflru_cache_with_ttl(maxsize: int = 128, ttl_seconds: int = 300):
"""带过期时间的 LRU 缓存装饰器"""
defdecorator(func: Callable) -> Callable:
cache = {}
@functools.wraps(func)
defwrapper(*args, **kwargs):
key = (args, tuple(sorted(kwargs.items())))
if key in cache:
value, expire_at = cache[key]
if time.time() < expire_at:
return value
result = func(*args, **kwargs)
expire_at = time.time() + ttl_seconds
iflen(cache) >= maxsize:
oldest_key = min(cache, key=lambdak: cache[k][1])
del cache[oldest_key]
cache[key] = (result, expire_at)
return result
return wrapper
return decorator
@lru_cache_with_ttl(maxsize=64, ttl_seconds=60)
deffetch_weather(city: str) -> dict:
print(f" 🌤 正在查询城市: {city}（模拟 API 调用）")
return {"city": city, "temp": 22, "humidity": 65}
# 第一次：调用 API
result1 = fetch_weather("北京")
# 第二次：命中缓存
result2 = fetch_weather("北京")
print(result1)
print(result2)
```
**第 2 题：工厂模式扩展——多数据库支持**
使用第 2.2 节的工厂模式思路，设计一个支持 SQLite、PostgreSQL 两种数据库连接器的工厂类：
```python
from abc importABC, abstractmethod
classDatabase(ABC):
@abstractmethod
defconnect(self, uri: str):
pass
@abstractmethod
defquery(self, sql: str) -> list:
pass
@abstractmethod
defclose(self):
pass
classSQLiteDB(Database):
defconnect(self, uri: str):
import sqlite3
self.conn = sqlite3.connect(uri.replace("sqlite:///", ""))
returnself
defquery(self, sql: str) -> list:
returnself.conn.execute(sql).fetchall()
defclose(self):
self.conn.close()
classDatabaseFactory:
_databases = {
"sqlite": SQLiteDB,
"postgresql": None, # 需要你补全
}
@classmethod
defcreate(cls, database_type: str) -> Database:
if database_type notincls._databases:
raiseValueError(f"不支持的数据库类型: {database_type}")
returncls._databases[database_type]()
# 使用
db = DatabaseFactory.create("sqlite")
db.connect("sqlite:///test.db")
results = db.query("SELECT 1 + 1")
db.close()
```
**第 3 题：用策略模式实现多种排序算法**
```python
from abc importABC, abstractmethod
import time
import random
classSortStrategy(ABC):
@abstractmethod
defsort(self, data: list) -> list:
pass
@abstractmethod
defcomplexity(self) -> str:
pass
classBubbleSortStrategy(SortStrategy):
defsort(self, data: list) -> list:
arr = data.copy()
n = len(arr)
for i inrange(n):
for j inrange(0, n - i - 1):
if arr[j] > arr[j + 1]:
arr[j], arr[j + 1] = arr[j + 1], arr[j]
return arr
defcomplexity(self) -> str:
return"O(n^2)"
classQuickSortStrategy(SortStrategy):
defsort(self, data: list) -> list:
iflen(data) <= 1:
return data
pivot = data[len(data) // 2]
left = [x for x in data if x < pivot]
middle = [x for x in data if x == pivot]
right = [x for x in data if x > pivot]
returnself.sort(left) + middle + self.sort(right)
defcomplexity(self) -> str:
return"O(n log n) 平均"
classSorter:
def__init__(self, strategy: SortStrategy):
self.strategy = strategy
defset_strategy(self, strategy: SortStrategy):
self.strategy = strategy
defsort(self, data: list) -> list:
returnself.strategy.sort(data)
# 对比两种策略
data = [random.randint(1, 10000) for _ inrange(10000)]
for name, strategy in [("冒泡", BubbleSortStrategy()), ("快速", QuickSortStrategy())]:
sorter = Sorter(strategy)
start = time.perf_counter()
result = sorter.sort(data)
elapsed = time.perf_counter() - start
print(f"{name}排序({strategy.complexity()}): {elapsed:.4f}s")
```
**第 4 题：组合模式实现文件系统遍历**
组合模式允许你把"单个对象"和"对象集合"统一对待。请实现一个简单的文件系统操作类：
```python
from abc importABC, abstractmethod
classFileSystemComponent(ABC):
@abstractmethod
defshow_details(self, indent: str = "") -> str:
pass
@abstractmethod
defget_size(self) -> int:
pass
classFile(FileSystemComponent):
def__init__(self, name: str, size: int):
self.name = name
self.size = size
defshow_details(self, indent: str = "") -> str:
returnf"{indent}📄 {self.name} ({self.size} bytes)"
defget_size(self) -> int:
returnself.size
classDirectory(FileSystemComponent):
def__init__(self, name: str):
self.name = name
self._children: list[FileSystemComponent] = []
defadd(self, component: FileSystemComponent):
self._children.append(component)
defshow_details(self, indent: str = "") -> str:
lines = [f"{indent}📁 {self.name}/"]
for child inself._children:
lines.append(child.show_details(indent + " "))
return"\n".join(lines)
defget_size(self) -> int:
returnsum(child.get_size() for child inself._children)
# 使用
root = Directory("项目")
src = Directory("src")
src.add(File("main.py", 2048))
src.add(File("utils.py", 1024))
root.add(src)
root.add(File(".gitignore", 128))
print(root.show_details())
print(f"\n总大小: {root.get_size()} bytes")
```
**第 5 题：综合实战——用多个模式搭建一个插件系统**
这是一个综合性题目。请用以下模式设计一个可扩展的插件系统：
-**工厂模式**：创建不同类型的插件
-**策略模式**：每个插件是一个策略
-**观察者模式**：插件可以向事件总线注册回调
-**外观模式**：提供一个简单的 `PluginManager` 类
提示框架：
```python
from abc importABC, abstractmethod
from typing import Callable, Any
classPlugin(ABC):
"""插件基类"""
@abstractmethod
defexecute(self, context: dict) -> dict:
pass
@abstractmethod
defname(self) -> str:
pass
classEventNotifier:
"""事件通知器"""
def__init__(self):
self._handlers: dict[str, list[Callable]] = {}
defon(self, event: str, handler: Callable):
self._handlers.setdefault(event, []).append(handler)
deftrigger(self, event: str, data: dict = None):
for handler inself._handlers.get(event, []):
handler(event, data or {})
classPluginManager:
"""插件管理器——外观模式"""
def__init__(self):
self.notifier = EventNotifier()
self._plugins: dict[str, Plugin] = {}
defregister_plugin(self, plugin: Plugin):
self._plugins[plugin.name()] = plugin
print(f" ✅ 插件 '{plugin.name()}' 已注册")
defexecute(self, plugin_name: str, context: dict) -> dict:
if plugin_name notinself._plugins:
raiseValueError(f"未找到插件: {plugin_name}")
returnself._plugins[plugin_name].execute(context)
deflist_plugins(self) -> list[str]:
returnlist(self._plugins.keys())
# 定义插件
classEmailPlugin(Plugin):
defname(self) -> str:
return"email"
defexecute(self, context: dict) -> dict:
recipient = context.get("recipient", "admin@example.com")
print(f" ✉ 发送邮件至: {recipient}")
return {"status": "sent", "to": recipient}
classSMSPlugin(Plugin):
defname(self) -> str:
return"sms"
defexecute(self, context: dict) -> dict:
phone = context.get("phone", "+8613800138000")
print(f" 📱 发送短信至: {phone}")
return {"status": "delivered", "phone": phone}
# 运行
manager = PluginManager()
manager.register_plugin(EmailPlugin())
manager.register_plugin(SMSPlugin())
print(f"已安装插件: {', '.join(manager.list_plugins())}")
manager.execute("email", {"recipient": "user@example.com"})
```
---
## 七、设计模式的"反模式"
在结束之前，我要说一些很重要但不太受欢迎的话：**不是所有地方都需要设计模式。**
### 7.1 常见陷阱
| 陷阱 | 表现 | 建议 |
|------|------|------|
| **过度设计** | 一个小脚本里用了 8 个模式 | 用小脚本就该有小脚本的写法 |
| **为了模式而模式** | 明明 5 行代码能解决，非要拆成 5 个类 | 先满足功能，再考虑扩展性 |
| **滥用继承** | 三层以上的继承树 | Python 优先用组合（composition），少用继承 |
| **忽视 Pythonic** | 用 Java 的思维写 Python | Python 有 Python 的习惯用法 |
### 7.2 SOLID 原则速记
设计模式背后有一组更底层的原则，叫做 SOLID：
-**S**ingle Responsibility（单一职责）：一个类只做一件事
-**O**pen/Closed（开闭原则）：对扩展开放，对修改关闭
-**L**iskov Substitution（里氏替换）：子类应该能透明替换父类
-**I**nterface Segregation（接口隔离）：接口尽量小
-**D**ependency Inversion（依赖倒置）：高层和低层都应依赖抽象
记住：**模式是手段，SOLID 是目的。** 真正重要的不是你用了几个模式，而是你的代码是否易读、易维护、易测试。
---
## 八、总结
这一期的内容比较多，我们系统地学习了：
1.**单例模式**——保证全局唯一实例。但在 Python 中，优先考虑模块级变量。
2.**工厂模式**——解耦对象的创建和使用，便于扩展。
3.**建造者模式**——链式调用构建复杂对象，替代"参数爆炸"。
4.**适配器模式**——让不同接口的组件可以协同工作。
5.**装饰器模式**——给对象添加新功能，与 Python 的 `@decorator` 相辅相成。
6.**外观模式**——简化复杂子系统的一行调用。
7.**策略模式**——运行时切换算法，符合开闭原则。
8.**观察者模式**——事件驱动架构的基础，解耦发布者和订阅者。
9.**责任链模式**——请求逐层处理，每层职责独立。
10.**重构实战**——把一个"面条函数"拆成职责清晰的多个类。
记住最重要的一句话：**好的代码不是用了多少模式，而是让读代码的人一眼就能看懂你在做什么。**
---
## 九、下期预告
**Episode 23：Python 异步编程深度进阶——asyncio 高级用法与实战**
我们已经在前面的章节（Episode 13）介绍了异步编程的基础概念。这一期我们将深入进阶：
- asyncio 事件循环的内部原理
- Task、Future、Semaphore、Queue 的高级用法
- 异步上下文管理器与异步生成器
- 异步 WebSocket 实战
- 如何将现有同步代码安全地异步化
- 常见陷阱：死锁、竞态条件、阻塞调用
敬请期待！
---
**📚 系列回顾**
到目前为止我们学过的内容：
| Episode | 主题 |
|---------|------|
| 01 | 环境搭建与基础语法 |
| 02 | 数据结构与字符串处理 |
| 03 | 面向对象编程 |
| 04 | 装饰器、生成器与文件 IO |
| 05 | 爬虫入门 |
| 06 | 数据分析入门 |
| 07 | Web 后端开发 |
| 08 | AI 实战入门 |
| 09 | 前端入门 |
| 10 | 项目实战——Dashboard |
| 11 | Docker 容器化部署 |
| 12 | 数据库进阶 |
| 13 | 异步编程与并发 |
| 14 | 日志系统与调试技巧 |
| 15 | 自动化运维与脚本实战 |
| 16 | 单元测试与代码质量 |
| 17 | 微服务入门 |
| 18 | CI/CD 与 DevOps 实战 |
| 19 | 并发与多线程深度实践 |
| 20 | 包管理与发布 |
| 21 | 性能优化全指南 |
| 22 | 设计模式实战 ← **本期** |
恭喜你！从零基础到现在，你已经掌握了一套完整的 Python 工程技能链。设计模式是你走向"资深工程师"的关键一步——它教你如何组织代码的结构，而不仅仅是"怎么写对的逻辑"。接下来就看你想拿这套技能去做什么了！💪
10个月宝宝每天需要喝多少奶粉？
