03-函数式编程工具
难度：⭐⭐⭐ 专家 预计时间：45分钟 前置知识：函数基础、装饰器概念、lambda表达式、闭包 引入版本：Python 2.5+ (functools 模块)
functools 模块专门用于高阶函数（即操作函数的函数）。它是编写装饰器、缓存优化和构建复杂函数逻辑的神器，底层由 C 实现，性能优异。
为什么需要 functools？
问题场景：重复计算导致性能瓶颈
python
# ❌ 递归无缓存：指数级复杂度
def fib(n):
if n < 2: return n
return fib(n-1) + fib(n-2) # fib(40) 耗时 20 秒
# ❌ 属性每次访问都重新计算
class DataService:
@property
def config(self):
return load_config_from_db() # 每次访问都查数据库
# ❌ 装饰器丢失函数元数据
def timer(func):
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
return wrapper
@timer
def process():
"""处理数据"""
pass
print(process.__name__) # wrapper (丢失了原函数名)
print(process.__doc__) # None (丢失了文档)问题：如何避免重复计算、简化函数接口、保留装饰器元数据？
概念铺垫
┌─────────────────────────────────────────────────────────────┐
│ functools 模块核心组件 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 1. 缓存系列 │
│ ───────────────────────────────────────────── │
│ • lru_cache：LRU缓存装饰器，限制大小 │
│ • cache：无限制缓存（Python 3.9+） │
│ • cached_property：属性缓存（Python 3.8+） │
│ • 生活类比：备忘录（查过的结果直接抄） │
│ │
│ 2. 函数变换 │
│ ───────────────────────────────────────────── │
│ • partial：冻结参数，创建偏函数 │
│ • partialmethod：partial 的方法版本 │
│ • 生活类比：预填表单（部分字段已填好） │
│ │
│ 3. 装饰器工具 │
│ ───────────────────────────────────────────── │
│ • wraps：保留函数元数据 │
│ • 生活类比：签名代理（保留原作者署名） │
│ │
│ 4. 类型分发 │
│ ───────────────────────────────────────────── │
│ • singledispatch：基于类型动态分发 │
│ • singledispatchmethod：方法版本 │
│ • 生活类比：快递分拣（按包裹类型分流） │
│ │
│ 5. 归约工具 │
│ ───────────────────────────────────────────── │
│ • reduce：序列累积归约 │
│ • 生活类比：折叠账单（逐项累加汇总） │
│ │
│ 使用决策树： │
│ 计算慢？→ lru_cache / cache │
│ 属性贵？→ cached_property │
│ 参数多？→ partial │
│ 写装饰器？→ wraps │
│ 类型分发？→ singledispatch │
│ 累积逻辑？→ reduce │
│ │
└─────────────────────────────────────────────────────────────┘L1 理解层：会用
lru_cache / cache 缓存装饰器
语法结构
┌─────────────────────────────────────────────────────────────┐
│ 缓存装饰器语法 │
│ │
│ @lru_cache(maxsize=128) # LRU缓存，限制大小 │
│ @lru_cache(maxsize=None) # 无限制缓存 │
│ @cache # 无限制缓存简写（3.9+） │
│ │
│ 参数： │
│ maxsize：缓存容量，None 表示无限 │
│ typed：是否区分参数类型（False=不区分） │
│ │
│ 方法： │
│ .cache_info() # 获取缓存统计 │
│ .cache_clear() # 清空缓存 │
│ │
└─────────────────────────────────────────────────────────────┘最简示例
python
from functools import lru_cache
@lru_cache(maxsize=128)
def fib(n):
if n < 2: return n
return fib(n-1) + fib(n-2)
print(fib(40)) # 102334155（毫秒级）
print(fib.cache_info()) # CacheInfo(hits=..., misses=...)详细示例
python
from functools import lru_cache, cache
import time
# lru_cache：限制缓存大小
@lru_cache(maxsize=128)
def expensive_compute(n: int) -> int:
time.sleep(0.1) # 模拟耗时操作
return n * n
print(expensive_compute(5)) # 第一次：计算
print(expensive_compute(5)) # 第二次：缓存命中
# cache：无限制缓存（Python 3.9+）
@cache
def factorial(n: int) -> int:
if n < 2: return 1
return n * factorial(n - 1)
print(factorial(10)) # 第一次计算
print(factorial(10)) # 缓存命中
print(factorial.cache_info()) # 查看缓存统计
# cache_clear：清空缓存
expensive_compute.cache_clear()
print(expensive_compute(5)) # 再次计算关键代码说明
| 代码 | 含义 | 为什么这样写 |
|---|---|---|
@lru_cache(maxsize=128) | 限制缓存128个结果 | 防止内存无限增长 |
@cache | 无限制缓存 | 参数组合少时更简洁 |
.cache_info() | 查看命中统计 | 监控缓存效率 |
.cache_clear() | 清空缓存 | 强制重新计算 |
cached_property 属性缓存
语法结构
┌─────────────────────────────────────────────────────────────┐
│ cached_property 语法 │
│ │
│ @cached_property │
│ def method(self): │
│ ... │
│ │
│ 特点： │
│ • 首次访问计算，后续直接返回缓存 │
│ • 线程安全 │
│ • 可用 del 清除缓存 │
│ │
└─────────────────────────────────────────────────────────────┘最简示例
python
from functools import cached_property
class Config:
@cached_property
def data(self):
print("加载配置...")
return {"host": "localhost"}
c = Config()
print(c.data) # 加载配置...
print(c.data) # 直接返回缓存关键代码说明
| 代码 | 含义 | 为什么这样写 |
|---|---|---|
@cached_property | 属性缓存 | 只计算一次，后续访问零成本 |
del obj.prop | 清除缓存 | 强制重新计算 |
partial 偏函数
语法结构
┌─────────────────────────────────────────────────────────────┐
│ partial 语法 │
│ │
│ partial(func, *args, **kwargs) │
│ │
│ 冻结部分参数，生成新函数 │
│ │
│ 示例： │
│ binary_int = partial(int, base=2) │
│ binary_int("1010") # 等价于 int("1010", base=2) │
│ │
└─────────────────────────────────────────────────────────────┘最简示例
python
from functools import partial
binary_int = partial(int, base=2)
print(binary_int("1010")) # 10
print(binary_int("1111")) # 15详细示例
python
from functools import partial
# 冻结多个参数
multiply = partial(lambda x, y, z: x * y * z, y=2, z=3)
print(multiply(4)) # 24（4 * 2 * 3）
# 回调函数传参
def on_click(button_id, event):
print(f"按钮 {button_id} 点击: {event}")
btn1_click = partial(on_click, button_id=1)
btn1_click("click") # 按钮 1 点击: clickwraps 装饰器工具
语法结构
┌─────────────────────────────────────────────────────────────┐
│ wraps 语法 │
│ │
│ @wraps(func) │
│ def wrapper(*args, **kwargs): │
│ ... │
│ │
│ 保留的属性： │
│ __name__, __doc__, __module__, __annotations__ │
│ │
└─────────────────────────────────────────────────────────────┘最简示例
python
from functools import wraps
def timer(func):
@wraps(func)
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
return wrapper
@timer
def process():
"""处理数据"""
pass
print(process.__name__) # process（保留了）
print(process.__doc__) # 处理数据（保留了）singledispatch 类型分发
语法结构
┌─────────────────────────────────────────────────────────────┐
│ singledispatch 语法 │
│ │
│ @singledispatch │
│ def func(arg): │
│ ... │
│ │
│ @func.register │
│ def _(arg: Type): │
│ ... │
│ │
│ 根据第一个参数类型分发 │
│ │
└─────────────────────────────────────────────────────────────┘最简示例
python
from functools import singledispatch
@singledispatch
def process(data):
raise NotImplementedError("Unsupported type")
@process.register
def _(data: int):
return data * 2
@process.register
def _(data: str):
return data.upper()
print(process(10)) # 20
print(process("hi")) # HI详细示例
python
from functools import singledispatch
@singledispatch
def serialize(data):
"""序列化数据"""
return str(data)
@serialize.register
def _(data: dict):
import json
return json.dumps(data)
@serialize.register
def _(data: list):
return "[" + ", ".join(serialize(x) for x in data) + "]"
@serialize.register
def _(data: int):
return str(data)
print(serialize({"a": 1})) # '{"a": 1}'
print(serialize([1, 2, 3])) # '[1, 2, 3]'reduce 归约累积
语法结构
┌─────────────────────────────────────────────────────────────┐
│ reduce 语法 │
│ │
│ reduce(func, iterable, initializer=None) │
│ │
│ func：二元函数 (x, y) → result │
│ iterable：可迭代对象 │
│ initializer：初始值（可选） │
│ │
│ 执行过程： │
│ reduce(f, [a, b, c]) = f(f(a, b), c) │
│ │
└─────────────────────────────────────────────────────────────┘最简示例
python
from functools import reduce
numbers = [1, 2, 3, 4, 5]
result = reduce(lambda x, y: x * y, numbers)
print(result) # 120（1*2*3*4*5）详细示例
python
from functools import reduce
# 连乘（阶乘）
nums = [1, 2, 3, 4, 5]
product = reduce(lambda x, y: x * y, nums)
print(product) # 120
# 合并字典
dicts = [{"a": 1}, {"b": 2}, {"a": 3}]
merged = reduce(lambda d1, d2: {**d1, **d2}, dicts)
print(merged) # {'a': 3, 'b': 2}
# 带初始值
total = reduce(lambda x, y: x + y, [], 0)
print(total) # 0（空列表用初始值）L2 实践层：用好
lru_cache / cache 推荐做法
| 做法 | 原因 | 示例 |
|---|---|---|
| 参数必须可哈希 | 缓存用字典存储 | int, str, tuple ✅；list, dict ❌ |
| 设置合理的 maxsize | 防止内存溢出 | 参数组合多时设128或256 |
| 纯函数优先使用 | 无副作用，缓存安全 | 数学计算、数据转换 |
| 监控 cache_info | 评估缓存效果 | 命中率低则调整策略 |
lru_cache / cache 反模式
python
# ❌ 不可哈希参数（报错）
@lru_cache
def process(items: list): # list 不可哈希
return sum(items)
process([1, 2, 3]) # TypeError: unhashable type: 'list'
# ✅ 使用元组（可哈希）
@lru_cache
def process(items: tuple):
return sum(items)
process((1, 2, 3)) # 正常工作python
# ❌ 有副作用的函数（缓存导致副作用丢失）
@lru_cache
def send_email(user_id: int) -> bool:
# 缓存后，相同 user_id 只发送一次邮件
return api.send_mail(user_id)
# ✅ 缓存结果，副作用单独处理
@lru_cache
def get_email_content(user_id: int) -> str:
return generate_content(user_id)
def send_email(user_id: int) -> bool:
content = get_email_content(user_id)
return api.send_mail(user_id, content)lru_cache / cache 适用场景
| 场景 | 是否推荐 | 原因 |
|---|---|---|
| 递归算法 | ✅ 推荐 | 消除重复计算 |
| 数学计算 | ✅ 推荐 | 纯函数，结果稳定 |
| 数据转换 | ✅ 推荐 | 输入相同，输出相同 |
| API调用 | ⚠️ 谨慎 | 数据可能变化 |
| 写操作 | ❌ 不推荐 | 有副作用 |
cached_property 推荐做法
| 做法 | 原因 | 示例 |
|---|---|---|
| 计算成本高的属性 | 避免重复计算 | 数据库查询、网络请求 |
| 只读配置属性 | 一次加载，多次使用 | config.yaml 解析 |
| 线程安全场景 | 自动处理并发 | 多线程访问同一属性 |
cached_property 反模式
python
# ❌ 可变属性用 cached_property（缓存后无法更新）
class Config:
@cached_property
def settings(self):
return load_settings()
def update_settings(self, new_value):
self.settings['key'] = new_value # 修改的是缓存副本
# ✅ 需要可变时用普通 property
class Config:
@property
def settings(self):
if self._settings is None:
self._settings = load_settings()
return self._settings
def update_settings(self, new_value):
self._settings['key'] = new_valuecached_property 适用场景
| 场景 | 是否推荐 | 原因 |
|---|---|---|
| 数据库连接信息 | ✅ 推荐 | 只读，一次加载 |
| 配置解析 | ✅ 推荐 | 解析成本高 |
| 计算密集型属性 | ✅ 推荐 | 避免重复计算 |
| 需要动态更新 | ❌ 不推荐 | 缓存无法自动刷新 |
partial 推荐做法
| 做法 | 原因 | 示例 |
|---|---|---|
| 简化重复调用 | 避免每次传相同参数 | partial(int, base=2) |
| 回调函数绑定 | GUI/异步场景常用 | partial(handler, user_id=1) |
| 配置默认值 | 统一函数行为 | partial(fetch, timeout=30) |
partial 反模式
python
# ❌ 简单场景用 partial（过度设计）
add_one = partial(lambda x, y: x + y, y=1)
print(add_one(5))
# ✅ 直接写函数更清晰
def add_one(x):
return x + 1
print(add_one(5))wraps 推荐做法
| 做法 | 原因 | 示例 |
|---|---|---|
| 所有装饰器必用 | 保留函数元数据 | @wraps(func) |
| 保持调试友好 | 日志显示原函数名 | traceback 更清晰 |
wraps 反模式
python
# ❌ 不使用 wraps（丢失元数据）
def my_decorator(func):
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
return wrapper
@my_decorator
def say_hello():
"""Say hello."""
pass
print(say_hello.__name__) # wrapper（错误）
print(say_hello.__doc__) # None（丢失）
# ✅ 使用 wraps
from functools import wraps
def my_decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
return wrappersingledispatch 推荐做法
| 做法 | 原因 | 示例 |
|---|---|---|
| 类型判断替代 if-else | 更清晰、可扩展 | @process.register(int) |
| 第三方类型支持 | 注册外部类型 | @process.register(numpy.ndarray) |
| 默认处理抛异常 | 明确不支持类型 | raise NotImplementedError |
singledispatch 反模式
python
# ❌ 用 if-else 判断类型（冗长、难扩展）
def process(data):
if isinstance(data, int):
return data * 2
elif isinstance(data, str):
return data.upper()
elif isinstance(data, list):
return [x * 2 for x in data]
else:
raise TypeError(f"Unsupported type: {type(data)}")
# ✅ 用 singledispatch（清晰、可扩展）
from functools import singledispatch
@singledispatch
def process(data):
raise TypeError(f"Unsupported type: {type(data)}")
@process.register
def _(data: int):
return data * 2
@process.register
def _(data: str):
return data.upper()
@process.register
def _(data: list):
return [x * 2 for x in data]singledispatch 适用场景
| 场景 | 是否推荐 | 原因 |
|---|---|---|
| 多类型处理 | ✅ 推荐 | 代码清晰、易扩展 |
| 序列化/反序列化 | ✅ 推荐 | 类型明确 |
| 验证/转换 | ✅ 推荐 | 按类型分发 |
| 简单两三种类型 | ❓ 可选 | if-else 也足够 |
reduce 推荐做法
| 做法 | 原因 | 示例 |
|---|---|---|
| 复杂累积逻辑 | 自定义归约规则 | 字典合并、树构建 |
| 简单场景用内置 | 更高效易读 | sum() 代替 reduce 求和 |
reduce 反模式
python
# ❌ 简单求和用 reduce（冗长）
total = reduce(lambda x, y: x + y, numbers)
# ✅ 使用内置 sum（更简洁高效）
total = sum(numbers)
# ❌ 简单求最大值用 reduce
maximum = reduce(lambda x, y: x if x > y else y, numbers)
# ✅ 使用内置 max
maximum = max(numbers)reduce 适用场景
| 场景 | 是否推荐 | 原因 |
|---|---|---|
| 连乘 | ✅ 推荐 | 无内置替代 |
| 字典合并 | ✅ 推荐 | 自定义逻辑 |
| 树/图构建 | ✅ 推荐 | 复杂累积 |
| 求和/最大值 | ❌ 不推荐 | 有更简洁的内置函数 |
L3 专家层：深入
lru_cache 底层原理
Python 如何实现
lru_cache 使用双向链表 + 字典实现 LRU 淘汰：
python
from functools import lru_cache
@lru_cache(maxsize=3)
def compute(n):
return n * 2
compute(1) # cache: {1}
compute(2) # cache: {1, 2}
compute(3) # cache: {1, 2, 3}
compute(4) # cache: {2, 3, 4}，淘汰 1（最久未用）
compute(1) # cache: {3, 4, 1}，重新计算，淘汰 2lru_cache 性能考量
| 操作 | 时间复杂度 | 说明 |
|---|---|---|
| 缓存命中 | O(1) | 字典查找 |
| 缓存未命中 | O(f) | f 是原函数复杂度 |
| LRU淘汰 | O(1) | 双向链表头部删除 |
注意： 当 maxsize 远小于参数组合数量时，频繁淘汰会导致缓存命中率下降。
lru_cache 设计动机
| 设计选择 | 原因 |
|---|---|
| 双向链表 | O(1) 淘汰最久未用项 |
| 字典存储 | O(1) 查找缓存结果 |
| typed=False | 合并相同值的不同类型（1 和 1.0） |
lru_cache 知识关联
缓存知识关联：
┌─────────────────┐ ┌─────────────────┐
│ dict │────→│ lru_cache │
│ 字典基础 │ │ 缓存存储 │
└─────────────────┘ └─────────────────┘
│
↓
┌─────────────────┐
│ 双向链表 │
│ LRU 淘汰 │
└─────────────────┘cached_property 底层原理
Python 如何实现
cached_property 是描述符实现：
python
from functools import cached_property
class Config:
@cached_property
def data(self):
return load_config()
# 实际执行过程
c = Config()
c.data # 第一次：调用 __get__，计算并存储到 __dict__
c.data # 第二次：直接从 __dict__ 读取，不触发描述符描述符机制
┌─────────────────────────────────────────────────────────────┐
│ cached_property 描述符流程 │
│ │
│ 第一次访问 obj.prop： │
│ 1. 查 obj.__dict__ → 未找到 │
│ 2. 查类描述符 → 找到 cached_property │
│ 3. 调用 __get__(obj, type) │
│ 4. 计算属性值 │
│ 5. 存入 obj.__dict__['prop'] │
│ 6. 返回结果 │
│ │
│ 第二次访问 obj.prop： │
│ 1. 查 obj.__dict__ → 找到 'prop' │
│ 2. 直接返回值（不触发描述符） │
│ │
│ del obj.prop 后： │
│ 1. 删除 obj.__dict__['prop'] │
│ 2. 下次访问重新触发描述符 │
│ │
└─────────────────────────────────────────────────────────────┘与 property 的区别
| 特性 | property | cached_property |
|---|---|---|
| 存储位置 | 无（每次计算） | obj.dict |
| 触发机制 | 每次调用 get | 首次后直接读取 |
| 线程安全 | 无保证 | 有锁保护 |
| 可清除 | 不需要 | del obj.prop |
cached_property 性能考量
| 操作 | 时间复杂度 | 说明 |
|---|---|---|
| 首次访问 | O(f) | f 是计算函数复杂度 |
| 后续访问 | O(1) | 直接字典读取 |
| 清除缓存 | O(1) | 删除字典键 |
cached_property 设计动机
| 设计选择 | 原因 |
|---|---|
| 存入 dict | 后续访问零成本 |
| 描述符实现 | 统一属性访问接口 |
| 线程安全 | 多线程场景常见 |
singledispatch 底层原理
Python 如何实现
singledispatch 使用字典存储类型到函数的映射：
python
from functools import singledispatch
@singledispatch
def process(data):
return "default"
# 内部结构（简化）
print(process.registry) # {object: <default_func>, int: <int_func>, ...}
print(process.dispatch(int)) # 返回 int 类型的处理函数类型查找机制
┌─────────────────────────────────────────────────────────────┐
│ singledispatch 类型查找流程 │
│ │
│ process(data) │
│ │ │
│ ↓ │
│ type(data) → 查 registry 字典 │
│ │ │
│ ↓ │
│ 找到？→ 直接调用注册函数 │
│ │ │
│ ↓ (未找到) │
│ 查父类（MRO顺序） │
│ │ │
│ ↓ │
│ 找到父类处理函数？→ 调用 │
│ │ │
│ ↓ (未找到) │
│ 调用默认函数 │
│ │
└─────────────────────────────────────────────────────────────┘singledispatch 性能考量
| 操作 | 时间复杂度 | 说明 |
|---|---|---|
| 类型查找 | O(1) | 字典查找 |
| 父类查找 | O(n) | n 是 MRO 长度（通常很小） |
| 函数调用 | O(f) | f 是处理函数复杂度 |
singledispatch 设计动机
| 设计选择 | 原因 |
|---|---|
| 只根据第一个参数分发 | 实现简单，符合 Python 习惯 |
| 用 MRO 查找父类 | 支持继承关系 |
| 注册机制 | 可动态添加新类型 |
本章小结
┌─────────────────────────────────────────────────────────────┐
│ functools 核心功能回顾 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 工具 用途 版本要求 │
│ ─────────────── ────────────────────── ───── │
│ lru_cache LRU缓存，可限制大小 2.5+ │
│ cache 无限制缓存 3.9+ │
│ cached_property 属性缓存 3.8+ │
│ reduce 序列累积归约 2.5+ │
│ partial 冻结参数，偏函数 2.5+ │
│ wraps 装饰器元数据保留 2.5+ │
│ singledispatch 类型动态分发 3.4+ │
│ │
│ 决策指南： │
│ 计算慢 → lru_cache / cache │
│ 属性贵 → cached_property │
│ 参数多 → partial │
│ 写装饰器 → wraps │
│ 类型分发 → singledispatch │
│ 累积逻辑 → reduce │
│ │
└─────────────────────────────────────────────────────────────┘自检清单
回答以下问题，检查你是否掌握了核心概念：
- lru_cache 和 cache 有什么区别？
- 缓存函数的参数为什么必须可哈希？
- cached_property 和 property 有什么区别？
- partial 的作用是什么？
- 为什么装饰器必须使用 wraps？
- singledispatch 如何实现类型分发？
- reduce 和 sum 有什么区别？
答案：
lru_cache可限制缓存大小；cache无限制（Python 3.9+）- 缓存用字典存储，字典键必须是可哈希类型
cached_property首次计算后缓存；property每次访问都计算- 冻结部分参数，生成简化的新函数
wraps保留原函数的__name__、__doc__等元数据，便于调试- 根据第一个参数的类型，查找注册的处理函数
reduce是通用归约；sum是求和专用，更高效
本章术语表
| 术语 | 定义 | 本章位置 |
|---|---|---|
| lru_cache | LRU缓存装饰器 | L1理解层 |
| cache | 无限制缓存装饰器 | L1理解层 |
| cached_property | 属性缓存装饰器 | L1理解层 |
| partial | 偏函数，冻结参数 | L1理解层 |
| wraps | 装饰器元数据保留工具 | L1理解层 |
| singledispatch | 类型动态分发装饰器 | L1理解层 |
| reduce | 序列累积归约函数 | L1理解层 |
| maxsize | lru_cache 的缓存容量参数 | L2实践层 |
| 可哈希 | 能作为字典键的类型 | L2实践层 |
| LRU | Least Recently Used 最近最少使用 | L3专家层 |
扩展阅读
- Python 官方文档：functools 模块
- 《流畅的Python》第7章：函数装饰器和闭包
- Python 3.8 新特性：cached_property
- Python 3.9 新特性：cache 装饰器