第7章 7.1 函数式编程：纯函数与高阶

⏱️ 学习节奏：90 分钟｜ 📖 难度：进阶｜ 🔗 上一章：6.5 天气查询 SPA ｜ 🔜 下一章：7.2 闭包与高阶函数

🎯 开场 3 分钟：为什么要学这个？

上一章我们做了一个天气查询的 SPA 页面，点击按钮就能看到北京的天气。你有没有想过：如果天气 API 返回的数据格式乱七八糟，你的页面就会直接崩溃？

这就是我们今天要解决的问题。

痛点场景：你写了一个处理数据的脚本，今天跑得好好的，明天数据变了 1%，整个程序就出 bug。为啥？因为你的代码太依赖外部状态了——数据变了我就跑不通。

学完这章，你手里会多两把利器：

纯函数：让你的代码像数学公式一样可靠，同样输入永远同样输出
高阶函数：把函数当成积木，想怎么组合就怎么组合

想象一下：纯函数就像自动售货机——你按同样的按钮，永远出同样的饮料。而非纯函数就像人工点单——取决于服务员心情，同样的按钮可能出不同饮料。

🧱 基础 25 分钟：核心概\n\n\n\n\n\n念（小白视角）

什么是纯函数？先别想代码，想点餐

你有没有去过这种餐厅：你点「宫保鸡丁 + 米饭」，不管几点、不管哪个服务员、不管你心情如何，端上来的永远是一模一样的一份菜？

这就是纯函数。

纯函数的定义：相同的输入，永远产生相同的输出，且没有任何副作用（side effect）

副作用：修改了外部状态，比如改了全局变量、写了文件、发了网络请求
纯函数只干一件事：根据输入算输出，不动别的

生活类比：纯函数 = 投币游戏机

你往游戏机里投一枚硬币，按「抓娃娃」按钮：

✅ 纯函数版：投一个币，按按钮，娃娃被抓起来（或者没抓到）。你再投一个币、再按，结局一样
❌ 非纯函数版：投一个币，突然发现机器里只剩最后两个娃娃了——这次抓到，下次可能抓不到（因为机器状态变了）

# 纯函数：只依赖输入，不碰任何外部状态
def calculate_price(price, tax_rate):
"""根据单价和税率计算总价"""
return price * (1 + tax_rate)

# 测试一下
result = calculate_price(100, 0.1)
print(result)  # 输出: 110.0
print(result)  # 输出: 110.0 （再跑一次，还是 110.0，稳定！）

这个函数只看了一眼输入，返回了一个计算结果。它没有：
- 修改任何外部变量
- 读写文件
- 打印日志（严格来说 print 算副作用，但这里先忽略）

为什么要用纯函数？三个字：好测试

想象你要测试这个逻辑：「用户买 100 块钱的东西，加 10% 税，应该收 110」

# 纯函数测试：简单粗暴
assert calculate_price(100, 0.1) == 110.0
assert calculate_price(200, 0.1) == 220.0
assert calculate_price(100, 0.05) == 105.0

搞定！没有任何 mocking、没有任何 stub、没有任何依赖注入——因为这函数啥外部东西都不用，自己测自己。

高阶函数：函数也能当参数和返回值

高阶函数听起来吓人，说白了就是两件事：

函数可以当参数传进去
函数可以当返回值吐出来

生活类比：你想寄一个快递

普通做法：自己打包、自己去快递站
高阶函数做法：你把打包指南（函数）给快递员，让他按你的流程打包+寄送

第一个高阶函数：把函数当参数用

# 这是一个「处理器」函数，它自己不做事，只是调用你传进来的函数
def process_data(data, processor):
"""对数据应用某种处理"""
return processor(data)

# 定义具体的处理函数
def double(x):
return x * 2

def square(x):
return x ** 2

# 使用：把处理函数传进去
numbers = [1, 2, 3, 4, 5]

result = process_data(numbers, double)
print(result)  # 输出: [2, 4, 6, 8, 10]

result = process_data(numbers, square)
print(result)  # 输出: [1, 4, 9, 16, 25]

看到了吗？process_data 是个通用框架，具体怎么处理，由你传入的函数决定。

第二个高阶函数：函数作为返回值

# 这是一个「乘法器工厂」，它返回一个专门做乘法的新函数
def create_multiplier(factor):
"""创建一个乘以特定因子的函数"""
def multiplier(x):
    return x * factor
return multiplier

# 生成了两个不同的「乘法器」
double = create_multiplier(2)   # 创建一个 x*2 的函数
triple = create_multiplier(3)  # 创建一个 x*3 的函数
ten_times = create_multiplier(10)

# 使用
print(double(5))    # 输出: 10
print(triple(5))    # 输出: 15
print(ten_times(5)) # 输出: 50

这就是闭包的基础！create_multiplier 返回的 multiplier 函数「记住」了它出生时的 factor 值。

函数组合：把多个函数串起来用

想象你要做一道菜：洗菜 → 切菜 → 炒菜 → 装盘。你不需要一个人干所有事，而是让每个人专注自己的步骤，最后串起来。

# 定义三个简单的处理函数
def wash(vegetable):
return f"洗净的{vegetable}"

def cut(vegetable):
return f"切好的{vegetable}"

def cook(vegetable):
return f"炒好的{vegetable}"

# 手动组合（这只是演示，下一章会讲更优雅的方式）
def make_dish(vegetable):
return cook(cut(wash(vegetable)))

# 使用
result = make_dish("白菜")
print(result)  # 输出: 炒好的切好的洗净的白菜

Point-Free 风格：不说「x」也能编程

Point-Free 就是不用明确写出参数名的编程风格。

# 普通写法：明确写出参数 x
double = lambda x: x * 2

# Point-Free 写法：借用现成的运算符
from functools import partial
from operator import mul

double = partial(mul, 2)  # 固定 mul 的第一个参数为 2

# 使用
print(double(5))  # 输出: 10

这个风格能让代码更简洁，但别过头—— readability counts（可读性第一）。

🔥 实战 35 分钟：3 个递进的小项目

项目 1（5 分钟）：纯函数数据清洗工具

场景：你有一堆用户数据，格式乱七八糟，需要统一处理。

# 数据清洗工具 - 纯函数版本

def normalize_name(name):
"""规范化姓名：去空格、首字母大写"""
return name.strip().title()

def normalize_age(age):
"""规范化年龄：转整数，超过150当异常处理"""
try:
    age_int = int(age)
    return age_int if age_int <= 150 else None
except (ValueError, TypeError):
    return None

def validate_record(name, age):
"""验证单条记录，返回(是否有效, 清洗后的数据)"""
clean_name = normalize_name(name)
clean_age = normalize_age(age)

if clean_name and clean_age:
    return True, {"name": clean_name, "age": clean_age}
return False, None

# 测试数据
raw_data = [
("  john doe ", "25"),
("JANE SMITH", "30"),
("Bob   ", "invalid"),
("Alice", "28"),
]

# 处理所有数据
valid_records = []
for name, age in raw_data:
is_valid, record = validate_record(name, age)
if is_valid:
    valid_records.append(record)

print("有效记录：")
for record in valid_records:
print(f"  - {record['name']}, {record['age']}岁")

预期输出：

有效记录：
- John Doe, 25岁
- Jane Smith, 30岁
- Alice, 28岁

一句话解释：每个函数都只管自己的输入输出，不碰外部数据——出了问题很好定位。

项目 2（15 分钟）：高阶函数处理 CSV 数据

场景：你需要从一个 CSV 文件里筛选出符合条件的数据，并做统计。

为了让代码直接可运行，我把数据直接写在代码里，实际项目从文件读取就行

import csv
from io import StringIO

# 模拟的 CSV 数据（实际项目用 open('data.csv') 读取）
csv_data = """name,department,salary,years
张三,技术部,15000,3
李四,市场部,12000,5
王五,技术部,20000,8
赵六,人事部,10000,2
孙七,技术部,18000,4
周八,市场部,11000,3"""

# 解析 CSV
def parse_csv(csv_string):
"""解析 CSV 字符串，返回列表字典"""
reader = csv.DictReader(StringIO(csv_string))
return list(reader)

# 高阶函数：通用的筛选器
def filter_employees(employees, predicate):
"""根据条件筛选员工
- employees: 员工列表
- predicate: 一个函数，接收员工字典，返回 True/False
"""
return [emp for emp in employees if predicate(emp)]

# 高阶函数：通用的统计器
def aggregate_employees(employees, aggregator, field):
"""对员工列表的某个字段做统计
- aggregator: 一个函数，比如 sum, max, min
"""
values = [int(emp[field]) for emp in employees]
return aggregator(values)

# 使用：解析数据
all_employees = parse_csv(csv_data)

# 筛选条件 1：技术部的员工
tech_employees = filter_employees(all_employees, lambda e: e["department"] == "技术部")
print(f"技术部员工：{[e['name'] for e in tech_employees]}")

# 筛选条件 2：工龄超过 3 年的员工
senior_employees = filter_employees(all_employees, lambda e: int(e["years"]) > 3)
print(f"工龄超过3年：{[e['name'] for e in senior_employees]}")

# 统计：技术部平均薪资
tech_salaries = [int(e["salary"]) for e in tech_employees]
avg_salary = sum(tech_salaries) / len(tech_salaries)
print(f"技术部平均薪资：{avg_salary:.2f}元")

# 统计：最高薪资
max_salary = max(int(e["salary"]) for e in all_employees)
max_earner = [e["name"] for e in all_employees if int(e["salary"]) == max_salary]
print(f"最高薪资：{max_salary}元，获奖者：{max_earner}")

预期输出：

技术部员工：['张三', '王五', '孙七']
工龄超过3年：['李四', '王五', '孙七']
技术部平均薪资：17666.67元
最高薪资：20000元，获奖者：['王五']

一句话解释：筛选和统计逻辑完全分离，你想换什么条件就换什么条件，核心代码不用动。

项目 3（15 分钟）：组合函数做待办清单管理

场景：你需要一个小工具，能给待办事项打标签、筛选、完成，还能导出统计。

from datetime import datetime

# 模拟的任务数据
tasks = [
{"id": 1, "content": "写周报", "priority": "中", "done": False, "tags": ["工作"]},
{"id": 2, "content": "买菜", "priority": "高", "done": True, "tags": ["生活"]},
{"id": 3, "content": "健身", "priority": "高", "done": False, "tags": ["健康", "生活"]},
{"id": 4, "content": "读书", "priority": "低", "done": False, "tags": ["学习"]},
{"id": 5, "content": "开会", "priority": "高", "done": True, "tags": ["工作"]},
]

# ============ 纯函数：不变的数据操作 ============

def add_tag(task, tag):
"""给任务加标签（返回新任务，原任务不变）"""
return {**task, "tags": task["tags"] + [tag]}

def mark_done(task):
"""标记任务完成"""
return {**task, "done": True}

def mark_undone(task):
"""标记任务未完成"""
return {**task, "done": False}

def set_priority(task, priority):
"""设置优先级"""
return {**task, "priority": priority}

# ============ 高阶函数：通用的列表处理器 ============

def process_tasks(tasks, *operations):
"""对任务列表应用一系列操作"""
result = tasks
for op in operations:
    result = op(result)
return result

def filter_by(predicate):
"""返回一个函数：过滤任务列表"""
def apply(tasks):
    return [t for t in tasks if predicate(t)]
return apply

def sort_by(key):
"""返回一个函数：按某个字段排序"""
def apply(tasks):
    return sorted(tasks, key=lambda t: t[key])
return apply

def group_by(key):
"""返回一个函数：按某个字段分组"""
def apply(tasks):
    groups = {}
    for task in tasks:
        k = task[key]
        if k not in groups:
            groups[k] = []
        groups[k].append(task)
    return groups
return apply

# ============ 组合使用 ============

# 场景1：找出所有未完成的高优先级任务
high_priority_pending = filter_by(lambda t: t["priority"] == "高" and not t["done"])(tasks)
print("🔥 高优先级待办：")
for task in high_priority_pending:
print(f"  [{task['id']}] {task['content']}")

# 场景2：按优先级分组显示
grouped = group_by("priority")(tasks)
print("\n📊 按优先级分组：")
for priority, group in grouped.items():
print(f"  {priority}级：{[t['content'] for t in group]}")

# 场景3：统计完成率
total = len(tasks)
done = len([t for t in tasks if t["done"]])
print(f"\n📈 完成率：{done}/{total} = {done/total*100:.0f}%")

# 场景4：给「健身」任务加上「运动」标签（返回新列表）
updated_tasks = [
add_tag(t, "运动") if t["content"] == "健身" else t
for t in tasks
]
print(f"\n✨ 更新后的健身任务标签：{[t for t in updated_tasks if t['content']=='健身'][0]['tags']}")

预期输出：

🔥 高优先级待办：
[1] 写周报
[3] 健身

📊 按优先级分组：
中级：['写周报']
低级：['读书']
高优先级：['买菜', '健身', '开会']

📈 完成率：2/5 = 40%

✨ 更新后的健身任务标签：['健康', '生活', '运动']

一句话解释：所有操作都是纯函数，原数据不变，新数据从旧数据生成——想回滚？直接用原数据就行。

💪 进阶 20 分钟：常见坑 + 性能小贴士

坑 1：以为纯函数就不会报错

# ❌ 错误示例：虽然「看起来」是纯函数，但会抛出异常
def divide(a, b):
return a / b  # b=0 时程序崩溃

# ✅ 正确示例：考虑边界情况
def safe_divide(a, b):
if b == 0:
    return None  # 或者抛出明确的异常
return a / b

提示：纯函数不等于不会出错，而是出错了也是输入决定的结果。

坑 2：高阶函数里忘记返回值

# ❌ 错误示例：filter_tasks 返回 None
def filter_tasks(tasks, condition):
filtered = [t for t in tasks if condition(t)]
# 忘记 return！filtered 变量丢了

# ✅ 正确示例：明确返回结果
def filter_tasks(tasks, condition):
return [t for t in tasks if condition(t)]

坑 3：在高阶函数里修改外部变量

# ❌ 错误示例：修改了外部状态
count = 0
def count_tasks(tasks):
global count
count = len(tasks)  # 副作用！
return count

# ✅ 正确示例：只返回值
def count_tasks(tasks):
return len(tasks)  # 没有副作用

坑 4：函数组合顺序搞反

def add_one(x):
return x + 1

def double(x):
return x * 2

# ❌ 错误示例：先加一再翻倍 = (x+1)*2
result = double(add_one(5))  # 12

# ✅ 正确示例：如果是「先翻倍再加一」= x*2+1
result = add_one(double(5))  # 11

提示：函数组合从内到外执行，想清楚先后顺序。

坑 5：Point-Free 过头反而难读

# ❌ 过度追求 Point-Free
from functools import reduce
import operator

# 这行代码能跑，但谁能看懂？
result = reduce(operator.add, map(lambda x: x * 2, filter(lambda x: x > 0, [-1, 2, -3, 4])))

# ✅ 适度 Point-Free，保持可读性
positive_numbers = [x for x in [-1, 2, -3, 4] if x > 0]
doubled = list(map(lambda x: x * 2, positive_numbers))
result = sum(doubled)

提示：代码是给人看的，偶尔为了优雅牺牲可读性不值得。

性能小贴士：生成器配合高阶函数

import time

# ❌ 如果数据量很大，用 list 会一次性加载到内存
def slow_process(items):
return [x * 2 for x in items]  # 所有数据都在内存里

# ✅ 用生成器惰性求值，按需计算
def fast_process(items):
return (x * 2 for x in items)  # 一次只处理一个

# 测试大数据量的区别
large_data = range(1000000)

start = time.time()
result1 = list(slow_process(large_data))
print(f"list 方式：{time.time() - start:.3f}秒")

start = time.time()
result2 = fast_process(large_data)  # 生成器对象，几乎不耗时
print(f"生成器方式：{time.time() - start:.3f}秒（还没真正计算）")

# 真正需要结果时再计算
print(f"第一个元素：{next(result2)}")  # 触发计算，但只计算一个

调试技巧：用装饰器打印中间状态

# 一个简单的调试装饰器
def debug(func):
"""打印函数调用时的输入和输出"""
def wrapper(*args, **kwargs):
    result = func(*args, **kwargs)
    print(f"🔍 {func.__name__}({args}, {kwargs}) → {result}")
    return result
return wrapper

# 使用：给任何函数加上调试功能
@debug
def calculate_price(price, tax_rate):
return price * (1 + tax_rate)

@debug
def filter_employees(employees, predicate):
return [emp for emp in employees if predicate(emp)]

# 测试
calculate_price(100, 0.1)
filter_employees([{"name": "张三"}], lambda e: e["name"] == "张三")

输出：

🔍 calculate_price((100, 0.1), {}) → 110.0
🔍 filter_employees(([{'name': '张三'}],), {}) → [{'name': '张三'}]

✏️ 练习题 + 作业题

练习题（5 道，10 分钟内完成）

练习 1（2 分钟）：纯函数改写
- 输入：把下面的函数改成纯函数

# 原始版本（有副作用）
total = 0
def add_to_total(x):
total += x
return total

预期输出：add_to_total(5)(10) 返回 15
提示：用参数代替全局变量

练习 2（2 分钟）：加个判断
- 输入：在项目 1 的 normalize_age 函数里，加一个判断：如果年龄小于 0，返回 None
- 预期输出：normalize_age(-5) 返回 None
- 提示：加一行 if age_int < 0: return None

练习 3（3 分钟）：筛选新数据
- 输入：用项目 2 的 filter_employees 筛选出 salary > 15000 的员工
- 提示：改一下 lambda 条件就行

练习 4（3 分钟）：串起两个项目
- 输入：把项目 2 的筛选功能 + 项目 3 的分组功能组合起来用
- 提示：先筛选再分组，filter_employees(...)(data) 然后把结果传给 group_by(...)

练习 5（挑战题，5 分钟）：修复报错

# 运行这段代码会报错
def create_counter():
count = 0
def counter():
    count = count + 1  # 哪里错了？
    return count
return counter

c = create_counter()
print(c())

预期输出：1
提示：闭包里的变量作用域问题，想一想怎么让内层函数访问外层变量

作业题（30 分钟 - 2 小时）

作业：做一个「个人支出记录分析工具」

需求描述：
做一个命令行小工具，可以添加支出、查看统计、按月份/类别筛选。

功能点：
1. 添加支出记录（金额、类别、备注）
2. 查看所有记录 + 总支出
3. 按月份筛选（本月/上月/指定月份）
4. 按类别统计总支出

加分项：
1. 用纯函数实现所有数据操作
2. 用高阶函数实现筛选和统计
3. 支持从 JSON 文件读写数据（程序重启不丢数据）

验收标准：
- 能跑起来
- 能添加记录
- 能显示统计
- 代码有注释

提交方式：评论区贴代码或 GitHub 链接

📚 总结 + 资源

本章 3 个核心知识点

纯函数：相同输入产生相同输出，不碰外部状态——好测试、好推理、好Debug
高阶函数：函数可以当参数传、可以当返回值返——代码复用更灵活
函数组合：把简单函数组合成复杂逻辑——像搭积木一样写代码

延伸学习资源

Python 官方文档 - functools（高阶函数和装饰器的官方支持）
《Python Cookbook》第 5 章「函数」（很多实战技巧）
Fluent Python 第 5-7 章（深入理解 Python 的函数式特性）

互动钩子

💬 你在处理数据的时候，遇到过「同样的代码这次跑没问题、下次跑就崩」的情况吗？当时是怎么解决的？评论区聊聊，老粉优先回复！

🔜 下章预告：
这一章我们学会了把函数当积木用，下一章我们要深入一个更有意思的话题：闭包——函数是怎么「记住」出生时的环境的？以及装饰器——怎么在不修改原函数的情况下，给它加上新功能？敬请期待！

声明：本站所有文章，如无特殊说明或标注，均为本站原创发布。任何个人或组织，在未征得本站同意时，禁止复制、盗用、采集、发布本站内容到任何网站、书籍等各类媒体平台。如若本站内容侵犯了原著者的合法权益，可联系我们进行处理。

第7章 7.1 函数式编程：纯函数与高阶

第7章 7.1 函数式编程：纯函数与高阶

🎯 开场 3 分钟：为什么要学这个？

🧱 基础 25 分钟：核心概\n\n\n\n\n\n念（小白视角）

什么是纯函数？先别想代码，想点餐

生活类比：纯函数 = 投币游戏机

为什么要用纯函数？三个字：好测试

高阶函数：函数也能当参数和返回值

第一个高阶函数：把函数当参数用

第二个高阶函数：函数作为返回值

函数组合：把多个函数串起来用

Point-Free 风格：不说「x」也能编程

🔥 实战 35 分钟：3 个递进的小项目

项目 1（5 分钟）：纯函数数据清洗工具

项目 2（15 分钟）：高阶函数处理 CSV 数据

项目 3（15 分钟）：组合函数做待办清单管理

💪 进阶 20 分钟：常见坑 + 性能小贴士

坑 1：以为纯函数就不会报错

坑 2：高阶函数里忘记返回值

坑 3：在高阶函数里修改外部变量

坑 4：函数组合顺序搞反

坑 5：Point-Free 过头反而难读

性能小贴士：生成器配合高阶函数

调试技巧：用装饰器打印中间状态

✏️ 练习题 + 作业题

练习题（5 道，10 分钟内完成）

作业题（30 分钟 - 2 小时）

📚 总结 + 资源

本章 3 个核心知识点

延伸学习资源

互动钩子

评论(0)

提示：请文明发言取消回复

排行榜展示

第1章 1.3 字符串与 f-string 格式化

第10章 10.3 终极项目：仿 V2EX 论坛 API

第3章 3.3 作用域与闭包

第10章 10.5 终极实战：CLI 工具开发

第10章 10.1 实时聊天系统（WebSocket）

第5章 5.4 浏览器存储：localStorage/sessionStorage/IndexedDB

第7章 7.1 函数式编程：纯函数与高阶

第7章 7.1 函数式编程：纯函数与高阶

🎯 开场 3 分钟：为什么要学这个？

🧱 基础 25 分钟：核心概\n\n\n\n\n\n念（小白视角）

什么是纯函数？先别想代码，想点餐

生活类比：纯函数 = 投币游戏机

为什么要用纯函数？三个字：好测试

高阶函数：函数也能当参数和返回值

第一个高阶函数：把函数当参数用

第二个高阶函数：函数作为返回值

函数组合：把多个函数串起来用

Point-Free 风格：不说「x」也能编程

🔥 实战 35 分钟：3 个递进的小项目

项目 1（5 分钟）：纯函数数据清洗工具

项目 2（15 分钟）：高阶函数处理 CSV 数据

项目 3（15 分钟）：组合函数做待办清单管理

💪 进阶 20 分钟：常见坑 + 性能小贴士

坑 1：以为纯函数就不会报错

坑 2：高阶函数里忘记返回值

坑 3：在高阶函数里修改外部变量

坑 4：函数组合顺序搞反

坑 5：Point-Free 过头反而难读

性能小贴士：生成器配合高阶函数

调试技巧：用装饰器打印中间状态

✏️ 练习题 + 作业题

练习题（5 道，10 分钟内完成）

作业题（30 分钟 - 2 小时）

📚 总结 + 资源

本章 3 个核心知识点

延伸学习资源

互动钩子

评论(0)

提示：请文明发言 取消回复

相关文章

排行榜展示

提示：请文明发言取消回复