第二十篇：高阶函数——map、reduce、filter 完全指南

前言

**高阶函数（Higher-Order Functions）**是函数式编程的核心概念之一。本文讲解 Python 中三大经典高阶函数 map、reduce、filter 的用法，以及如何将函数作为参数传递来简化代码、提高复用性。

什么是高阶函数？

核心定义

高阶函数 = 接受函数作为参数或返回函数的函数。

def apply_operation(func, value):
    """高阶函数：接收一个函数和一个值"""
    return func(value)

result = apply_operation(lambda x: x * 2, 5)
print(result)  # 10 — lambda(x) → 5*2 = 10

返回函数的函数

def make_multiplier(factor):
    """工厂函数：返回一个乘法函数"""
    def multiplier(x):
        return x * factor
    return multiplier

double = make_multiplier(2)
triple = make_multiplier(3)

print(double(5))  # 10
print(triple(5))  # 15

map() 函数

基本用法

def square(x):
    return x ** 2

nums = [1, 2, 3, 4, 5]
squares = list(map(square, nums))
print(squares)  # [1, 4, 9, 16, 25]

# 等同于列表生成式
squares_lc = [square(x) for x in nums]

map 处理多个可迭代对象

addition = lambda a, b: a + b
a = [1, 2, 3, 4]
b = [10, 20, 30, 40]
result = list(map(addition, a, b))
print(result)  # [11, 22, 33, 44]

# 字符串映射
words = ["hello", "world"]
lengths = list(map(len, words))
print(lengths)  # [5, 5]

# 类型转换
strings = ["1", "2", "3"]
numbers = list(map(int, strings))
print(numbers)  # [1, 2, 3]

map 的性能特点

import timeit

nums = list(range(100_000))

start = timeit.default_timer()
result_map = list(map(lambda x: x * 2, nums))
time_map = timeit.default_timer() - start

start = timeit.default_timer()
result_lc = [x * 2 for x in nums]
time_lc = timeit.default_timer() - start

print(f"map 用时：{time_map:.4f}s")
print(f"生成式用时：{time_lc:.4f}s")
# map 通常比生成式略快 5%-15%（C 层循环优化）

reduce() 函数

functools.reduce

Python 3 中 reduce 被移到 functools 模块中：

from functools import reduce

nums = [1, 2, 3, 4, 5]

# 计算乘积
product = reduce(lambda a, b: a * b, nums)
print(product)  # 120 (1×2×3×4×5)

# 累加
total = reduce(lambda a, b: a + b, nums, 0)  # 0 是初始值
print(total)  # 15

# 找最大值
maximum = reduce(lambda a, b: a if a > b else b, nums)
print(maximum)  # 5

reduce 的工作原理

# reduce(func, [a, b, c, d], initial=0)
# 执行过程：
# result = initial      ← 先取初始值（如果有）
# result = func(result, a)  ← 然后从左到右依次应用
# result = func(result, b)
# result = func(result, c)
# ...直到遍历完所有元素

nums = [1, 2, 3, 4, 5]
product = reduce(lambda acc, x: acc * x, nums, 1)
print(product)  # 1 (初始) × 1 × 2 × 3 × 4 × 5 = 120

实战：字符串连接与扁平化

from functools import reduce
from itertools import chain

# 字符串连接
words = ["Hello", " ", "World", "!"]
joined = reduce(lambda a, b: a + b, words)
print(joined)  # Hello World!

# 扁平化嵌套列表
nested = [[1, 2], [3, 4], [5, 6]]
flattened = reduce(lambda a, b: a + b, nested)
print(flattened)  # [1, 2, 3, 4, 5, 6]

filter() 函数

基本用法

nums = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]

# 过滤偶数
evens = list(filter(lambda x: x % 2 == 0, nums))
print(evens)  # [2, 4, 6, 8, 10]

# 过滤非空字符串
words = ["hello", "", "world", None, "python", "   "]
cleaned = list(filter(None, words))  # None → 过滤所有假值
print(cleaned)  # ['hello', 'world', 'python']

# 自定义过滤
students = [
    {"name": "Alice", "score": 95},
    {"name": "Bob", "score": 45},
    {"name": "Carol", "score": 78},
    {"name": "Dave", "score": 62}
]
passed = list(filter(lambda s: s["score"] >= 60, students))
print(passed)
# [{'name': 'Alice', 'score': 95}, {'name': 'Carol', 'score': 78}, {'name': 'Dave', 'score': 62}]

filter vs 列表生成式

nums = range(1, 21)

# 传统 filter
odds_filter = list(filter(lambda x: x % 2 == 1, nums))

# 等价生成式
odds_lc = [x for x in nums if x % 2 == 1]

# 性能对比（filter 略快，可读性取决于复杂度）
import timeit
n = 100_000
t1 = timeit.timeit("list(filter(lambda x: x % 2 == 0, range({})))".format(n), number=100)
t2 = timeit.timeit("[x for x in range({}) if x % 2 == 0]".format(n), number=100)
print(f"filter: {t1:.4f}s, LC: {t2:.4f}s")

函数作为一等公民

闭包捕获

def outer(x):
    """闭包：外层变量被内层函数捕获"""
    def inner():
        return f"x = {x}"  # 捕获外部作用域的 x
    return inner

closure1 = outer(10)
closure2 = outer(20)
print(closure1())  # x = 10
print(closure2())  # x = 20

偏函数 partial

from functools import partial

def power(base, exponent):
    return base ** exponent

square = partial(power, exponent=2)  # 固定 exponent=2
cube = partial(power, exponent=3)    # 固定 exponent=3

print(square(5))  # 25
print(cube(3))    # 27

实用示例：动态绑定方法

class Service:
    def process(self, data, config, debug=False):
        print(f"处理: {data}, 配置: {config}, 调试: {debug}")

svc = Service()
process_with_debug = partial(svc.process, config="production", debug=True)
process_with_debug("订单数据")
# 输出：处理: 订单数据，配置: production，调试: True

链式调用模式

pipe 管道调用

def pipe(data, *functions):
    """管道调用：依次应用所有函数"""
    for func in functions:
        data = func(data)
    return data

# 使用示例
data = ["hello", "WORLD", "python", "code"]
cleaned = pipe(
    data,
    lambda x: [s.lower() for s in x],  # 转小写
    lambda x: [s for s in x if len(s) > 3],  # 过滤长度 ≤3
    lambda x: sorted(x),  # 排序
    "".join  # 拼接成字符串
)
print(cleaned)  # hellohelloworldpythocode

compose 函数组合

def compose(*funcs):
    """函数组合：f(g(x)) → h(k(f(x)))"""
    def composed(x):
        for func in reversed(funcs):
            x = func(x)
        return x
    return composed

add_ten = lambda x: x + 10
double = lambda x: x * 2
negate = lambda x: -x

fn = compose(add_ten, double, negate)
print(fn(5))  # -(5+10)*2 = -30

实战示例

计时器装饰器增强行为

import time
import functools

def timer(func):
    """计时装饰器"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        end = time.perf_counter()
        elapsed = end - start
        
        # 根据耗时选择不同的提示语
        if elapsed < 0.001:
            unit, display = "μs", int(elapsed * 1_000_000)
        elif elapsed < 1:
            unit, display = "ms", int(elapsed * 1_000)
        else:
            unit, display = "s", round(elapsed, 2)
        
        print(f"{func.__name__:20s} [{unit:2s}] {display:>8}")
        return result
    
    wrapper.elapsed_times = []  # 记录历史耗时
    return wrapper

@timer
def process_data(data):
    return sum(data)

@timer
def calculate(a, b):
    return a + b

process_data(range(1_000_000))
calculate(10, 5)

最佳实践

避免 lambda 滥用

# ❌ 不推荐：lambda 过长时难以阅读
sorted(items, key=lambda x: (x.score, -x.age))

# ✅ 推荐：提取为具名函数
def sort_key(item):
    return (item.score, -item.age)

sorted(items, key=sort_key)

装饰器封装通用记忆化

def memoize(func):
    """通用记忆化装饰器"""
    cache = {}
    
    def wrapper(*args):
        if args not in cache:
            cache[args] = func(*args)
        return cache[args]
    
    wrapper.cache_info = lambda: len(cache)
    wrapper.clear_cache = lambda: cache.clear()
    return wrapper

@memoize
def expensive_function(x):
    total = sum(range(x * 1000000))
    return total

print(expensive_function(5))   # 第一次：慢
print(expensive_function(5))   # 第二次：瞬间！（命中缓存）
expensive_function.clear_cache()  # 清除缓存

总结

本文涵盖了高阶函数的核心知识：

1. map() — 逐元素转换，比生成式略快
2. reduce() — 累积归约，从 functools 引入
3. filter() — 条件过滤，等价于含 if 的生成式
4. 函数传参 — 函数作为一等公民传入高阶函数
5. partial/lambda — 偏函数与匿名函数的场景选择
6. 组合与管道 — compose/pipe 实现函数式风格
7. 最佳实践 — 避免 lambda 过深、善用装饰器
8. 性能对比 — C 层 map/filter 略快于 Python 循环

下一篇文章将深入讲解 sorted 与匿名函数 lambda。

---

本文是 Python 教程系列的第二十篇。