模板方法模式详解：抽象基类定义算法骨架

模板方法模式详解：抽象基类定义算法骨架 | 极客日志

模板方法模式详解：抽象基类定义算法骨架

一、引言：你是否写过这样的重复代码？

在真实项目中，你是否遇到过这种情况：

# 数据处理流程 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 模块为这一模式提供了天然的语言支持。

二、核心概念：骨架与填充

2.1 三个关键元素

模板方法模式由三类方法构成，理解它们是掌握模式的关键：

模板方法（Template Method）：定义算法骨架的方法，通常在抽象基类中实现，不允许子类覆盖（Python 中可通过约定或 __init_subclass__ 强制）。它按固定顺序调用其他步骤。

抽象步骤（Abstract Steps）：算法中必须由子类实现的步骤，用 @abstractmethod 标注，子类必须覆盖，否则无法实例化。

钩子方法（Hook Methods）：提供默认实现（通常为空或返回默认值）的可选步骤，子类可以选择性覆盖，用于控制算法中的条件分支。

2.2 最小骨架示例

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

三、实战案例一：多格式数据处理管道

3.1 实现 CSV 和 API 两种数据处理器

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() 方法从未修改，两个子类只是填充了自己负责的那几块「拼图」。

四、实战案例二：Web 爬虫框架

爬虫系统是模板方法的另一个绝佳应用场景——爬取流程（请求→解析→存储→限速）高度一致，但不同网站的解析逻辑完全不同：

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)

六、Python 进阶技巧：让模板方法更 Pythonic

6.1 用 `__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

6.2 用 Mixin 组合多个「局部模板」

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

6.3 结合 dataclass 简化配置传递

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() -> :
         {}

     () -> :
        ()

6.4 单元测试：聚焦每个步骤

模板方法模式让测试极为简单——可以单独测试每个抽象步骤：

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 爬虫、测试框架、游戏关卡逻辑……在这些场景中，它能帮你消除重复、统一约束，让代码像一首乐曲：主旋律固定，每个乐器演奏自己的声部。

参考资料

《Design Patterns: Elements of Reusable Object-Oriented Software》- GoF，第 325 页模板方法
《Head First 设计模式》- Freeman，第八章「封装算法」
《流畅的 Python》第二版 - Luciano Ramalho，第 11 章「接口：从协议到 ABC」
Python abc 官方文档：https://docs.python.org/3/library/abc.html
Refactoring Guru 模板方法：https://refactoring.guru/design-patterns/template-method/python/example
PEP 3119 – Introducing Abstract Base Classes：https://peps.python.org/pep-3119/

模板方法模式详解：抽象基类定义算法骨架