函数式编程

What — 是什么

函数式编程(Functional Programming, FP)是一种编程范式,将计算视为数学函数的求值,强调纯函数、不可变数据和函数组合。Python 不是纯函数式语言,但提供了丰富的函数式特性——一等公民函数、高阶函数、lambda、闭包、生成器等,允许在面向对象和命令式风格中灵活运用函数式思想。

核心概念:

  • 一等公民函数:函数可以赋值给变量、作为参数传递、作为返回值
  • 高阶函数:接收函数作为参数或返回函数的函数(map/filter/reduce)
  • 纯函数:给定相同输入总返回相同输出,无副作用
  • 不可变数据:数据创建后不可修改,避免共享状态

关键特性:

  • Python 函数是一等公民,支持闭包和装饰器
  • functools 模块提供 partialreducelru_cachewraps 等工具
  • operator 模块提供函数化的运算符(itemgetterattrgettermethodcaller
  • 列表/字典/集合推导式是 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_cachefunctools记忆化缓存
compose无内置函数组合(需自实现)
itemgetter/attrgetteroperator函数化取值
lambdabuiltins匿名函数

运行机制:

  • 惰性求值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(itemgetterlambda x: x[0] 更快更清晰)
  • @lru_cache 缓存纯函数和昂贵计算,设合理的 maxsize
  • 偏函数用于固定配置参数,不用来替代默认参数
  • 函数组合用 pipe(左到右),比 compose(右到左)更符合直觉
  • 真正的不可变数据考虑第三方库 pyrsistentimmutabledict
  • 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)。


相关链接: