函数式编程
What — 是什么
函数式编程(Functional Programming, FP)是一种编程范式,将计算视为数学函数的求值,强调纯函数、不可变数据和函数组合。Python 不是纯函数式语言,但提供了丰富的函数式特性——一等公民函数、高阶函数、lambda、闭包、生成器等,允许在面向对象和命令式风格中灵活运用函数式思想。
核心概念:
- 一等公民函数:函数可以赋值给变量、作为参数传递、作为返回值
- 高阶函数:接收函数作为参数或返回函数的函数(map/filter/reduce)
- 纯函数:给定相同输入总返回相同输出,无副作用
- 不可变数据:数据创建后不可修改,避免共享状态
关键特性:
- Python 函数是一等公民,支持闭包和装饰器
functools模块提供partial、reduce、lru_cache、wraps等工具operator模块提供函数化的运算符(itemgetter、attrgetter、methodcaller)- 列表/字典/集合推导式是 map/filter 的 Pythonic 替代
- Python 没有内置不可变数据结构,但
tuple/frozenset可部分替代 - 模式匹配(Python 3.10+
match/case)简化条件分派
函数式工具箱:
| 工具 | 模块 | 用途 |
|---|---|---|
map(func, iter) | builtins | 映射变换 |
filter(func, iter) | builtins | 过滤筛选 |
reduce(func, iter) | functools | 累积归约 |
partial(func, *args) | functools | 偏函数(固定部分参数) |
lru_cache | functools | 记忆化缓存 |
compose | 无内置 | 函数组合(需自实现) |
itemgetter/attrgetter | operator | 函数化取值 |
lambda | builtins | 匿名函数 |
运行机制:
- 惰性求值:
map/filter/生成器表达式返回迭代器,不立即计算 - 闭包捕获:lambda 和内部函数捕获外部变量的引用(非值),注意延迟绑定
- memoization:
@lru_cache缓存函数结果,空间换时间 - 函数组合:
f(g(x))可以链式组合,Python 需手动或用第三方库
Why — 为什么
适用场景:
- 数据转换管道(ETL、数据清洗)
- 配置化行为(策略模式、回调)
- 缓存优化(
@lru_cache) - 事件处理和响应式编程
- 并行计算(
multiprocessing.Pool.map)
函数式 vs 命令式对比:
| 维度 | 函数式 | 命令式 |
|---|---|---|
| 核心思想 | 描述”做什么” | 描述”怎么做” |
| 状态 | 无副作用,无共享状态 | 可变状态,赋值驱动 |
| 可测试性 | 极高(纯函数) | 较低(依赖上下文) |
| 并行安全 | 天然安全 | 需加锁 |
| 可读性 | 声明式,简洁 | 过程式,直观 |
| Python 风格 | 辅助(非主流) | 主流 |
优缺点:
- ✅ 优点:
- 纯函数易于测试和推理
- 不可变数据消除竞态条件
- 函数组合构建复杂逻辑
- 惰性求值节省内存
- ❌ 缺点:
- Python 无尾递归优化,递归深度有限
- 不可变数据的修改需要重建,性能开销
- 过度使用 lambda 降低可读性
- 调试不如命令式直观
How — 怎么用
快速上手
from functools import partial, reduce, lru_cache
from operator import itemgetter, attrgetter
# 高阶函数
nums = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
# map/filter — 推导式通常更 Pythonic
squares = list(map(lambda x: x ** 2, nums))
evens = list(filter(lambda x: x % 2 == 0, nums))
# 但推导式更清晰
squares = [x ** 2 for x in nums] # 更 Pythonic
evens = [x for x in nums if x % 2 == 0] # 更 Pythonic
# reduce:累积
total = reduce(lambda a, b: a + b, nums, 0) # 55
product = reduce(lambda a, b: a * b, nums[0:5]) # 120
# 偏函数:固定部分参数
def power(base, exp):
return base ** exp
square = partial(power, exp=2)
cube = partial(power, exp=3)
print(square(5)) # 25
print(cube(3)) # 27
# operator 模块替代 lambda
from operator import add, mul
total = reduce(add, nums) # 代替 lambda a,b: a+b
product = reduce(mul, nums[:5]) # 代替 lambda a,b: a*b
代码示例1:数据管道与函数组合
from functools import reduce
from operator import itemgetter
from typing import Callable, TypeVar
T = TypeVar('T')
R = TypeVar('R')
# 函数组合工具
def compose(*funcs: Callable) -> Callable:
"""从右到左组合函数: compose(f, g, h)(x) = f(g(h(x)))"""
return reduce(lambda f, g: lambda x: f(g(x)), funcs)
def pipe(*funcs: Callable) -> Callable:
"""从左到右组合函数: pipe(f, g, h)(x) = h(g(f(x)))"""
return reduce(lambda f, g: lambda x: g(f(x)), funcs)
# 数据管道
def read_data():
return [
{"name": "Alice", "dept": "Engineering", "salary": 95000},
{"name": "Bob", "dept": "Marketing", "salary": 72000},
{"name": "Charlie", "dept": "Engineering", "salary": 105000},
{"name": "Diana", "dept": "Marketing", "salary": 68000},
{"name": "Eve", "dept": "Engineering", "salary": 88000},
]
def filter_engineering(employees):
return list(filter(lambda e: e["dept"] == "Engineering", employees))
def get_salaries(employees):
return list(map(itemgetter("salary"), employees))
def average(salaries):
return reduce(lambda a, b: a + b, salaries) / len(salaries) if salaries else 0
# 管道组合
avg_eng_salary = pipe(
read_data,
filter_engineering,
get_salaries,
average,
)
print(f"工程部平均薪资: ${avg_eng_salary():,.0f}")
# 工程部平均薪资: $96,000
代码示例2:@lru_cache 记忆化与偏函数实战
from functools import lru_cache, partial
import time
# lru_cache:记忆化递归
@lru_cache(maxsize=None)
def fibonacci(n: int) -> int:
if n < 2:
return n
return fibonacci(n - 1) + fibonacci(n - 2)
# 无缓存 vs 有缓存
start = time.time()
print(f"fib(100) = {fibonacci(100)}")
print(f"耗时: {time.time() - start:.6f}s") # 几乎瞬间
# 查看缓存信息
print(fibonacci.cache_info())
# CacheInfo(hits=98, misses=101, maxsize=None, currsize=101)
# 实战:HTTP 请求缓存
@lru_cache(maxsize=128)
def fetch_user(user_id: int) -> dict:
"""模拟API请求,相同ID只请求一次"""
time.sleep(0.1) # 模拟网络延迟
return {"id": user_id, "name": f"User{user_id}"}
# 第一次调用
start = time.time()
fetch_user(1)
print(f"首次请求: {time.time() - start:.3f}s") # ~0.1s
# 第二次调用(缓存命中)
start = time.time()
fetch_user(1)
print(f"缓存请求: {time.time() - start:.6f}s") # ~0.000001s
# 偏函数实战:配置化日志
import logging
def log_message(level: int, logger_name: str, message: str):
logger = logging.getLogger(logger_name)
logger.log(level, message)
# 预配置不同级别的日志函数
debug_log = partial(log_message, logging.DEBUG, "myapp")
info_log = partial(log_message, logging.INFO, "myapp")
error_log = partial(log_message, logging.ERROR, "myapp")
info_log("服务启动")
error_log("数据库连接失败")
代码示例3:模式匹配与不可变数据
from dataclasses import dataclass
from typing import Union
# Python 3.10+ 模式匹配(函数式风格的分派)
@dataclass(frozen=True)
class Circle:
radius: float
@dataclass(frozen=True)
class Rectangle:
width: float
height: float
@dataclass(frozen=True)
class Triangle:
base: float
height: float
Shape = Union[Circle, Rectangle, Triangle]
def area(shape: Shape) -> float:
match shape:
case Circle(radius=r):
return 3.14159 * r * r
case Rectangle(width=w, height=h):
return w * h
case Triangle(base=b, height=h):
return 0.5 * b * h
case _:
raise ValueError(f"未知形状: {shape}")
# 不可变数据操作(返回新对象而非修改)
def scale(shape: Shape, factor: float) -> Shape:
match shape:
case Circle(radius=r):
return Circle(radius=r * factor)
case Rectangle(width=w, height=h):
return Rectangle(width=w * factor, height=h * factor)
case Triangle(base=b, height=h):
return Triangle(base=b * factor, height=h)
original = Circle(radius=5)
scaled = scale(original, 2.0)
print(f"原始: {original}, 面积: {area(original):.2f}")
print(f"缩放: {scaled}, 面积: {area(scaled):.2f}")
# original 未被修改(frozen=True + 返回新对象)
常见问题与踩坑
| 问题 | 原因 | 解决方案 |
|---|---|---|
| lambda 延迟绑定 | 闭包引用循环变量 | 用默认参数 lambda x, i=i: ... |
lru_cache 内存泄漏 | 无 maxsize 或缓存大对象 | 设 maxsize,定期 cache_clear() |
| 递归深度溢出 | Python 无尾递归优化 | 改用迭代或 sys.setrecursionlimit() |
| 过度使用 lambda | 嵌套 lambda 难以阅读 | 超过一行用 def 定义命名函数 |
reduce 不如循环直观 | 需要理解累积器语义 | 简单场景用 sum()/any()/all() |
| 不可变数据修改开销 | 每次修改创建新对象 | 小数据用 frozen dataclass,大数据用 pyrsistent |
最佳实践
- 简单映射/过滤优先用推导式,比 map/filter 更 Pythonic
operator模块替代简单 lambda(itemgetter比lambda x: x[0]更快更清晰)@lru_cache缓存纯函数和昂贵计算,设合理的maxsize- 偏函数用于固定配置参数,不用来替代默认参数
- 函数组合用
pipe(左到右),比compose(右到左)更符合直觉 - 真正的不可变数据考虑第三方库
pyrsistent或immutabledict - Python 不是 Haskell,适度使用函数式风格,不追求纯函数式
面试题
Q1: Python 中 map/filter 和推导式哪个更好?
推导式通常更 Pythonic——可读性更好、更灵活(支持条件和嵌套)、性能相当。map/filter 的优势:1) 传入已有函数时更简洁(
map(str.strip, lines));2) 与多进程配合(pool.map);3) 惰性求值(生成器推导式(x for x in ...)也有此特性)。通用建议:简单变换用推导式,需要复用函数时用 map。
Q2: functools.partial 和 lambda 有什么区别?
partial固定函数的部分参数,创建新的可调用对象,保留原函数的__name__、__doc__等元信息。lambda 是匿名函数,每次创建新函数对象。partial 更适合固定参数(如square = partial(pow, exp=2)),lambda 更适合简单表达式(如lambda x: x.name)。partial 性能略好(少一层函数调用),lambda 更灵活(可修改参数顺序)。
Q3: @lru_cache 的原理和注意事项是什么?
@lru_cache用字典缓存函数调用的参数-结果映射,相同参数直接返回缓存值。LRU 策略在缓存满时淘汰最近最少使用的条目。注意事项:1) 参数必须是可哈希的(int/str/tuple/frozenset);2)maxsize=None无限缓存可能导致内存泄漏;3) 缓存的是返回值的引用,可变对象被修改会影响缓存;4) 不适合有副作用的函数(缓存命中时副作用不执行);5) 多线程安全(内部加锁)。
Q4: 为什么 Python 没有尾递归优化?
Guido van Rossum 明确反对尾递归优化(TCO),原因:1) TCO 会改变调用栈行为,使调试困难(traceback 中缺少帧);2) Python 的异常追踪依赖完整的调用栈;3) TCO 与 Python 的动态特性冲突(
sys._getframe()、inspect等);4) 递归不是 Python 的惯用风格,迭代更自然。替代方案:手动改为迭代、使用lru_cache优化递归、或用生成器实现惰性递归。
Q5: Python 3.10 的 match/case 和 if/elif 有什么区别?
match/case 是结构化模式匹配,不仅匹配值,还能解构数据结构、绑定变量、按类型匹配。
match shape: case Circle(radius=r): ...同时检查类型和提取属性。if/elif 只做布尔判断。match/case 的优势:1) 解构赋值内置于语法;2) 编译器可优化匹配顺序;3) 更声明式、更函数式;4) 穷举检查(可选)。劣势:只有 Python 3.10+ 支持,简单条件判断时过于复杂。
Q6: 如何实现函数组合?Python 为什么没有内置 compose?
手动实现:
compose = lambda *f: reduce(lambda f,g: lambda x: f(g(x)), f)或从左到右pipe。Python 没有内置 compose 因为:1) Python 社区偏好命令式风格,函数组合不是主流用法;2) 多参数函数组合复杂(需要柯里化);3) 推导式和生成器管道已覆盖大部分数据变换场景。第三方库toolz/fn提供了更完善的函数式工具。
Q7: 什么是柯里化(Currying)?Python 如何实现?
柯里化将多参数函数转化为一系列单参数函数:
f(a, b, c)→f(a)(b)(c)。Python 没有自动柯里化,可以手动实现:def add(a): return lambda b: lambda c: a + b + c。更实用的方式是用partial逐步固定参数:add3 = partial(add, 3); add3_5 = partial(add3, 5)。工具库toolz.curry装饰器可自动柯里化任意函数。
Q8: Python 中的”纯函数”有什么实际价值?
纯函数(给定输入总返回相同输出,无副作用)的价值:1) 可缓存(
@lru_cache安全使用);2) 可并行(无共享状态,GIL 不影响);3) 易测试(不依赖上下文,只需验证输入输出);4) 易推理(无隐式依赖)。但 Python 中追求纯函数式不现实——IO、数据库、网络都是副作用。实际做法是:核心逻辑用纯函数,副作用推到边界(如函数返回描述,外部执行IO)。
相关链接: