模板方法模式详解:抽象基类定义算法骨架
模板方法模式通过父类定义算法骨架,将具体步骤延迟到子类实现。利用 Python abc 模块构建抽象基类,区分模板方法、抽象步骤和钩子方法。文章涵盖多格式数据处理管道、Web 爬虫框架及报表生成系统三个实战案例,展示了如何通过继承复用核心流程并灵活定制局部逻辑。结合 Mixin 组合行为、dataclass 配置传递及单元测试技巧,有效消除重复代码,提升架构可维护性。

模板方法模式通过父类定义算法骨架,将具体步骤延迟到子类实现。利用 Python abc 模块构建抽象基类,区分模板方法、抽象步骤和钩子方法。文章涵盖多格式数据处理管道、Web 爬虫框架及报表生成系统三个实战案例,展示了如何通过继承复用核心流程并灵活定制局部逻辑。结合 Mixin 组合行为、dataclass 配置传递及单元测试技巧,有效消除重复代码,提升架构可维护性。

在真实项目中,你是否遇到过这种情况:
# 数据处理流程 A
def process_csv_data():
data = read_csv_file() # 读取数据
data = clean_data(data) # 清洗
data = validate_csv(data) # 校验(CSV 特有)
result = analyze(data) # 分析
save_to_database(result) # 保存
send_report_email(result) # 发送报告
# 数据处理流程 B(几乎一样,但某些步骤不同)
def process_json_data():
data = fetch_from_api() # 读取数据(不同)
data = clean_data(data) # 清洗(相同)
data = validate_json(data) # 校验(JSON 特有)
result = analyze(data) # 分析(相同)
save_to_database(result) # 保存(相同)
# 不需要发送报告(不同)
两个流程的骨架完全相同,只有若干步骤有差异。如果用复制粘贴解决,当「分析」逻辑需要修改时,你要同时改两处;如果流程再增加到五种、十种,维护将成为噩梦。
这正是模板方法模式(Template Method Pattern) 要解决的核心问题。
模板方法模式是 GoF 设计模式中最朴素、最实用的模式之一,其核心思想一句话概括:在父类中定义算法的骨架(模板),将某些步骤延迟到子类中实现,子类可以在不改变算法结构的前提下,重新定义特定步骤的具体实现。
Python 的 abc 模块为这一模式提供了天然的语言支持。
模板方法模式由三类方法构成,理解它们是掌握模式的关键:
模板方法(Template Method):定义算法骨架的方法,通常在抽象基类中实现,不允许子类覆盖(Python 中可通过约定或 __init_subclass__ 强制)。它按固定顺序调用其他步骤。
抽象步骤(Abstract Steps):算法中必须由子类实现的步骤,用 @abstractmethod 标注,子类必须覆盖,否则无法实例化。
钩子方法(Hook Methods):提供默认实现(通常为空或返回默认值)的可选步骤,子类可以选择性覆盖,用于控制算法中的条件分支。
from abc import ABC, abstractmethod
class DataProcessor(ABC):
"""抽象基类:数据处理算法骨架"""
def process(self) -> None:
"""模板方法:定义处理流程,子类不应覆盖此方法"""
raw_data = self.read_data() # 抽象步骤
clean = self.clean_data(raw_data) # 抽象步骤
if self.should_validate(): # 钩子:是否需要校验
clean = self.validate(clean)
result = self.analyze(clean) # 抽象步骤
self.save_result(result) # 抽象步骤
if self.should_notify(): # 钩子:是否需要通知
self.send_notification(result)
# ===== 抽象步骤:子类必须实现 =====
@abstractmethod
def read_data(self) -> list:
pass
@abstractmethod
def clean_data(self, data: list) -> list:
pass
@abstractmethod
def analyze(self, data: list) -> dict:
pass
@abstractmethod
def save_result(self, result: dict) -> None:
pass
# ===== 钩子方法:子类可选择覆盖 =====
def validate(self, data: list) -> list:
"""默认校验:过滤 None 值"""
return [item for item in data if item is not None]
def should_validate(self) -> bool:
"""钩子:是否执行校验步骤,默认开启"""
return True
def should_notify(self) -> bool:
"""钩子:是否发送通知,默认关闭"""
return False
def send_notification(self, result: dict) -> None:
"""钩子:通知逻辑,子类可覆盖"""
pass
import csv
import json
import io
from datetime import datetime
class CSVDataProcessor(DataProcessor):
"""CSV 文件数据处理器"""
def __init__(self, filepath: str, email: str = ''):
self.filepath = filepath
self.email = email
def read_data(self) -> list:
print(f"[CSV 处理器] 从文件读取:{self.filepath}")
sample_csv = "name,age,score\n张三,25,88\n李四,30,\n王五,22,95\n"
reader = csv.DictReader(io.StringIO(sample_csv))
return list(reader)
def clean_data(self, data: list) -> list:
print(f"[CSV 处理器] 清洗数据:{len(data)} 条")
for row in data:
row['age'] = int(row['age']) if row.get('age') else 0
row['score'] = float(row['score']) if row.get('score') else None
data
() -> :
valid = [row row data row[] ]
()
valid
() -> :
scores = [row[] row data]
{
: (scores),
: (scores)/(scores) scores ,
: (scores) scores ,
: (scores) scores ,
: datetime.now().isoformat()
}
() -> :
()
() -> :
(.email)
() -> :
()
():
():
.api_url = api_url
() -> :
()
[
{: , : , : },
{: , : , : },
{: , : , : },
{: , : , : },
]
() -> :
()
item data:
item[] = (item[]) item[]
data
() -> :
valid_values = [item[] item data item[]]
tag_groups = {}
item data:
item[]:
tag_groups.setdefault(item[], []).append(item[])
{
: (data),
: (valid_values),
: (valid_values)/(valid_values) valid_values ,
: {tag: (vals)/(vals) tag, vals tag_groups.items()}
}
() -> :
()
() -> :
(*)
()
csv_processor = CSVDataProcessor(, email=)
csv_processor.process()
(+*)
()
api_processor = APIDataProcessor()
api_processor.process()
运行效果:
==================================================
【CSV 数据处理(带邮件通知)】
[CSV 处理器] 从文件读取:students.csv
[CSV 处理器] 清洗数据:3 条
[CSV 处理器] 校验后:2/3 条有效
[CSV 处理器] 保存分析结果到数据库:{'count': 2, 'avg_score': 91.5, ...}
[CSV 处理器] 发送报告邮件至 [email protected]: 平均分 91.5
==================================================
【API 数据处理(无通知)】
[API 处理器] 从接口获取数据:https://api.example.com/data
[API 处理器] 标准化字段格式
[API 处理器] 写入缓存层:{"total": 4, "valid": 3, ...}
核心骨架 process() 方法从未修改,两个子类只是填充了自己负责的那几块「拼图」。
爬虫系统是模板方法的另一个绝佳应用场景——爬取流程(请求→解析→存储→限速)高度一致,但不同网站的解析逻辑完全不同:
import time
import random
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Optional
from datetime import datetime
@dataclass
class CrawlResult:
url: str
title: str
items: list = field(default_factory=list)
error: Optional[str] = None
crawled_at: str = field(default_factory=lambda: datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
class BaseCrawler(ABC):
"""爬虫抽象基类:定义爬取算法骨架"""
def __init__(self, delay_range: tuple = (1.0, 3.0)):
self.delay_range = delay_range
self._results: list[CrawlResult] = []
def crawl(self, urls: list[str]) -> list[CrawlResult]:
"""模板方法:完整爬取流程"""
self.setup() # 钩子:初始化(如登录、设置 headers)
for i, url in enumerate(urls):
print()
:
html = .fetch(url)
.is_valid(html):
()
result = .parse(html, url)
result = .transform(result)
.store(result)
._results.append(result)
Exception e:
()
._results.append(CrawlResult(url=url, title=, error=(e)))
:
._rate_limit()
.teardown()
._results
() -> :
() -> CrawlResult:
() -> :
() -> :
()
() -> :
()
() -> :
(html (html) > )
() -> CrawlResult:
result
() -> :
delay = random.uniform(*.delay_range)
()
time.sleep(delay * )
():
():
().__init__(delay_range=(, ))
._db: = []
() -> :
()
().setup()
() -> :
()
url:
ConnectionError()
() -> CrawlResult:
re
title_match = re.search(, html)
title = title_match.group() title_match
CrawlResult(
url=url, title=title, items=[{:,:}]
)
() -> CrawlResult:
result.title = result.title.replace(, ).strip()
result
() -> :
._db.append(result)
()
():
():
().__init__(delay_range=(, ))
.output_file = output_file
._products: = []
() -> :
()
sku = url.split()[-]
json.dumps({
: sku,
: ,
: (random.uniform(, ), ),
: random.randint(, ),
: (random.uniform(, ), )
})
() -> :
:
data = json.loads(html)
data data.get(, ) >
Exception:
() -> CrawlResult:
data = json.loads(html)
CrawlResult(
url=url, title=data[], items=[{
: data[],
: data[],
: data[]
}]
)
() -> :
result.items:
item = result.items[]
._products.append({**item, : result.title})
()
() -> :
()
().teardown()
报表生成是另一个模板方法的高频场景——数据查询→处理→渲染→输出,骨架固定:
from abc import ABC, abstractmethod
from typing import Any
from datetime import datetime
class ReportGenerator(ABC):
"""报表生成抽象基类"""
def generate(self, params: dict) -> str:
"""模板方法:报表生成全流程"""
print(f"\n[{self.report_name}] 开始生成报表...")
# 1. 参数校验
self._validate_params(params)
# 2. 查询数据
raw_data = self.query_data(params)
print(f" 数据查询完成:{len(raw_data)} 条原始记录")
# 3. 数据聚合
aggregated = self.aggregate(raw_data)
# 4. 应用过滤(钩子)
if self.apply_filter(params):
aggregated = self.filter_data(aggregated, params)
# 5. 渲染输出
output = self.render(aggregated, params)
# 6. 后处理(钩子)
output = self.post_process(output)
print(f" ✓ 报表生成完成")
return output
@property
@abstractmethod
def report_name(self) -> str:
pass
() -> :
() -> :
() -> :
() -> :
params params:
ValueError()
() -> :
params
() -> :
data
() -> :
output
():
() -> :
() -> :
[
{: , : , : , : },
{: , : , : , : },
{: , : , : , : },
{: , : , : , : },
{: , : , : , : },
]
() -> :
total = (row[] row data)
by_product = {}
by_region = {}
row data:
by_product[row[]] = by_product.get(row[], ) + row[]
by_region[row[]] = by_region.get(row[], ) + row[]
{
: total,
: by_product,
: by_region,
: (data)
}
() -> :
target_region = params.get()
()
data
() -> :
lines = [
*,
,
*,
,
,
,
]
product, amount (data[].items(), key= x: -x[]):
pct = amount / data[] *
lines.append()
lines.append()
region, amount (data[].items(), key= x: -x[]):
lines.append()
lines.append(*)
.join(lines)
() -> :
output +
report = SalesReport()
output = report.generate({: , : })
(output)
__init_subclass__ 防止模板方法被意外覆盖class StrictTemplate(ABC):
"""严格模板:防止子类覆盖核心模板方法"""
def __init_subclass__(cls, **kwargs):
super().__init_subclass__(**kwargs)
protected = ['process', 'execute', 'run']
for method in protected:
if method in cls.__dict__:
raise TypeError(f"子类 {cls.__name__} 不允许覆盖模板方法 '{method}'!请覆盖对应的抽象步骤方法。")
class MyProcessor(StrictTemplate):
def process(self): # 这会在类定义时立即报错
pass
class LoggingMixin:
"""日志 Mixin:为任何处理器添加日志能力"""
def process(self):
print(f"[LOG] {self.__class__.__name__} 开始执行")
result = super().process()
print(f"[LOG] {self.__class__.__name__} 执行完成")
return result
class TimingMixin:
"""计时 Mixin:为任何处理器添加计时能力"""
def process(self):
start = time.perf_counter()
result = super().process()
elapsed = time.perf_counter() - start
print(f"[TIMING] 总耗时:{elapsed:.3f}s")
return result
class EnhancedCSVProcessor(LoggingMixin, TimingMixin, CSVDataProcessor):
"""带日志和计时的 CSV 处理器"""
pass
from dataclasses import dataclass
@dataclass
class ProcessorConfig:
source: str
output_dir: str = './output'
validate: bool = True
notify_email: str = ''
batch_size: int = 1000
class ConfigurableProcessor(DataProcessor):
"""通过配置对象控制模板方法行为"""
def __init__(self, config: ProcessorConfig):
self.config = config
def should_validate(self) -> bool:
return self.config.validate
def should_notify(self) -> bool:
return bool(self.config.notify_email)
def read_data(self) -> list:
print(f"从 {self.config.source} 批量读取,批次大小:{self.config.batch_size}")
return []
def clean_data(self, data: list) -> list:
return data
def analyze() -> :
{}
() -> :
()
模板方法模式让测试极为简单——可以单独测试每个抽象步骤:
import pytest
from unittest.mock import patch, MagicMock
class TestCSVDataProcessor:
def setup_method(self):
self.processor = CSVDataProcessor('test.csv')
def test_clean_data_converts_types(self):
raw = [{'name': '张三', 'age': '25', 'score': '88.5'}]
clean = self.processor.clean_data(raw)
assert clean[0]['age'] == 25
assert clean[0]['score'] == 88.5
def test_validate_filters_null_scores(self):
data = [{'name': 'A', 'score': 90.0}, {'name': 'B', 'score': None}, {'name': 'C', 'score': 75.0}]
valid = self.processor.validate(data)
assert len(valid) == 2
assert all(row['score'] is not None row valid)
():
data = [{: }, {: }, {: }]
result = .processor.analyze(data)
result[] ==
result[] ==
():
processor_with_email = CSVDataProcessor(, email=)
processor_no_email = CSVDataProcessor()
processor_with_email.should_notify()
processor_no_email.should_notify()
():
call_log = []
.processor.read_data = : (call_log.append(), [])[]
.processor.clean_data = d: (call_log.append(), d)[]
.processor.analyze = d: (call_log.append(), {})[]
.processor.save_result = r: call_log.append()
.processor.process()
call_log == [, , , ]
陷阱一:模板方法中的步骤太多。当模板方法调用超过 7~8 个步骤时,说明类的职责过重,应考虑将部分步骤提取为子模板或辅助方法。
陷阱二:钩子方法过于复杂。钩子的默认实现应简单(空操作或返回简单布尔值),复杂的默认行为应封装为独立方法,通过钩子调用。
陷阱三:抽象步骤设计过于细碎。不是每一行代码都需要成为抽象步骤,只有真正需要子类差异化实现的步骤才应抽象化。
💡 黄金原则: 好的模板方法应该像一份食谱——步骤清晰、顺序固定、关键环节留给厨师发挥,而不是每一克调料都规定死。
模板方法模式是「好莱坞原则」(Don't call us, we'll call you)的完美体现——父类掌控整体流程,子类只需响应父类的「召唤」实现特定步骤。
回顾本文的核心要点:三类方法(模板方法、抽象步骤、钩子方法)各司其职;抽象步骤保证子类实现完整性,钩子方法提供灵活的条件分支;结合 Mixin 可实现模块化的行为组合;配合单元测试,每个步骤可独立验证;Python 的 abc 模块是实现此模式的天然利器。
模板方法模式最适合的场景正是那些「整体流程固定,局部步骤各异」的业务——数据处理管道、报表生成、Web 爬虫、测试框架、游戏关卡逻辑……在这些场景中,它能帮你消除重复、统一约束,让代码像一首乐曲:主旋律固定,每个乐器演奏自己的声部。
abc 官方文档:https://docs.python.org/3/library/abc.html
微信公众号「极客日志」,在微信中扫描左侧二维码关注。展示文案:极客日志 zeeklog
使用加密算法(如AES、TripleDES、Rabbit或RC4)加密和解密文本明文。 在线工具,加密/解密文本在线工具,online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具,curl 转代码在线工具,online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具,Base64 字符串编码/解码在线工具,online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具,Base64 文件转换器在线工具,online
将 Markdown(GFM)转为 HTML 片段,浏览器内 marked 解析;与 HTML转Markdown 互为补充。 在线工具,Markdown转HTML在线工具,online
将 HTML 片段转为 GitHub Flavored Markdown,支持标题、列表、链接、代码块与表格等;浏览器内处理,可链接预填。 在线工具,HTML转Markdown在线工具,online