06-函数式工具箱

第 6 章 函数式工具箱：functools 与 itertools

这一章给你一套「不用写循环、不堆临时变量」的武器：map/filter 的取舍、functools.lru_cache 缓存、partial 偏函数、reduce 归约、以及 itertools 的惰性组合。它们不是炫技，是用对了能少写一半样板、还更不容易出 bug。

墨叔：老哥，你在 Go 里想「对切片每个元素算个值」，是不是 for _, v := range s { out = append(out, f(v)) }？Python 里这活儿有更短的写法——列表推导、还有 map。但你一搜会发现，map 在 Python 圈口碑挺分裂：有人奉为圣经，有人见 map 就改推导式。今天咱们把这套「函数式工具箱」摊开，告诉你什么时候用、什么时候不如推导式、收益和代价各是什么。

老哥：我 AI 生成的代码里全是 [f(x) for x in xs]，从没用过 map。

墨叔：那说明你 AI 还挺懂事。但 functools 和 itertools 那两箱子里，有不少 Go 根本没有、能真省事的东西——尤其 lru_cache，你做过任何「重复计算」的事，它都能帮你白嫖性能。

一、map/filter/lambda：怎么用，以及该不该用

先摆清楚三个东西。map(f, xs) 返回「对每个元素套 f」的惰性迭代器；filter(p, xs) 返回「满足谓词 p 的元素」；lambda 是匿名小函数。

nums = [1, 2, 3, 4]
squared = list(map(lambda x: x * x, nums))     # [1, 4, 9, 16]
evens = list(filter(lambda x: x % 2 == 0, nums))  # [2, 4]

看起来比推导式短？其实不短，而且可读性更差。Python 之禅第一句就是 Explicit is better than implicit（显式优于隐式）。[x*x for x in nums] 一眼看懂，map(lambda ...) 要拐个弯。所以默认结论：简单变换用列表推导式，别用 map+lambda。

那 map 就一无是处？不是。两个它更合适的场景：

1. 已经有现成的具名函数，不想为它造个 lambda。比如 list(map(int, ["1","2","3"])) 比 [int(x) for x in ...] 还顺一点，因为 int 是现成的。
2. 惰性拼接多个迭代器，配合 itertools 时 map 返回迭代器不立即计算，省内存——而这个优势列表推导拿不到（它立即生成列表）。

# 惰性：不立刻算，用到才算
squares_iter = map(lambda x: x * x, range(10_000_000))   # 几乎零成本
first = next(squares_iter)                                # 此刻才算第一个

filter 同理：[x for x in nums if x % 2 == 0] 永远比 filter 清楚，除非你有个现成谓词函数要复用。我给你一条铁律：lambda 越长，越该抽成具名函数或换推导式。三行以上的 lambda 是代码异味。

对照 Go：Go 没有内建 map/filter（1.18 泛型后你可以自己写，但官方不提供），你要么手写循环，要么 gofunc 配 channel。Python 这套「把循环抽象成数据变换」的思维，是函数式基因的残留——但 Python 社区的主流选择是推导式优先，因为它同时占了「可读」和「灵活」（能同时 if 过滤+变换）。

二、functools.lru_cache：白嫖缓存最香的一块

这是函数式箱里最值钱的工具，没有之一。给一个纯函数（同样输入永远同样输出、无副作用）套上 @lru_cache，它会把「参数→返回值」存起来，下次同样参数直接返回，不重算。

from functools import lru_cache
import time
 
@lru_cache(maxsize=128)
def fib(n: int) -> int:
    if n < 2:
        return n
    return fib(n - 1) + fib(n - 2)
 
# 朴素递归 fib(35) 要算几亿次；加 lru_cache 后指数级塌成线性
print(fib(35))     # 一瞬间

fib 是教科书例子，但真实项目里更常见的是**「查数据库 / 调外部 API / 重算报表」这类贵操作**：参数不变就别再打库。

@lru_cache(maxsize=256)
def get_user_profile(uid: int) -> dict:
    # 假设这是一次 30ms 的 RPC
    return rpc.fetch_profile(uid)
 
# 同一请求里多处用到同一用户资料，只打一次库

maxsize 是缓存容量（LRU：最近最少用者淘汰）；设 None 表示不限量（小心内存）。cache_info() 能看命中率，调优神器：

print(get_user_profile.cache_info())   # CacheInfo(hits=..., misses=..., maxsize=256, currsize=...)

收益：零侵入的性能优化，加一行装饰器，重复计算的瓶颈消失。代价与雷区：

• 只适合纯函数。lru_cache 把参数当字典 key 缓存结果——如果参数不可哈希（比如传了 list/dict），直接报 TypeError。传可变对象当缓存 key 是错误的根源，要么转成元组，要么别缓存。
• 会掩盖副作用。若函数其实有副作用（写库、打日志），被缓存后副作用只发生一次，后面调用「假」返回，bug 极隐蔽。缓存前先问：它是纯的吗？
• ** TTL 没有**。lru_cache 不自动过期，数据变了缓存还是旧的。需要时间过期请用 cachetools.TTLCache 或自己加失效逻辑。
• 内存。缓存对象一直占着，大对象缓存多了吃内存。

老哥：这和 Go 里我自己写个 map[key]result 加锁有啥区别？

墨叔：本质一样，都是 memoization。区别在 Python 帮你处理了「参数哈希、容量淘汰、线程安全（CPython 下 lru_cache 加了锁）」这些脏活，还白送 cache_info。Go 你得自己 sync.Map + key 拼接 + 淘汰策略。此外 lru_cache 还能缓存有默认参数、关键字参数的函数，key 是整包参数元组——你不用操心拼接 key 字符串。但 Go 那个方案胜在「可控」：TTL、容量、失效时机你全说了算。缓存这东西，两条路都对，看你要省事还是要掌控。

三、partial 偏函数与 reduce 归约

functools.partial 把「一个函数 + 部分参数」固化成一个新函数。场景：你要把一个多参函数塞进只收单参的接口（比如回调、排序 key）。

from functools import partial
 
def send_email(to: str, subject: str, body: str) -> None: ...
 
# 把 to 固定，得到「只剩 subject/body」的单接收者发送器
send_to_boss = partial(send_email, to="[email protected]")
 
send_to_boss("周报", "本周无事故")   # 等价于 send_email("[email protected]", "周报", "本周无事故")

对照 Go：Go 没偏函数，你通常写个闭包 fn := func(subj, body string) { sendEmail("boss", subj, body) }。partial 就是闭包的语法糖，但更直白、不引入新 func 块。另一个常见用法：固定 sorted 的 key——

from functools import partial
rows.sort(key=partial(getattr, "age"))   # 按 age 属性排，等价于 key=lambda r: r.age

functools.reduce（Python 3 移到 functools）把序列「两两归约」成一个值。和 Go 的 for 累加大同小异，但更声明式。

from functools import reduce
 
total = reduce(lambda a, b: a + b, [1, 2, 3, 4], 0)    # 10
# 等价 Go: sum := 0; for _, v := range s { sum += v }

墨叔：但我要泼冷水——reduce 在 Python 里争议比 map 还大。简单的累加用 sum/max/min/any/all 这些内建，别用 reduce；reduce(lambda a,b: a+b, ...) 纯属和 sum() 过不去。只在「没有现成内建、且逻辑确实是两两折叠」时才上 reduce，比如把字典链起来：

from functools import reduce
import operator
 
merged = reduce(operator.or_, [{1}, {2}, {3}], set())   # {1,2,3}

收益：partial 让「固定参数」这件事零样板，reduce 把「折叠」语义显式化。代价：reduce 可读性差，新人看到要脑子转三圈；能用内建就别 reduce，能写循环就别硬 reduce。Python 社区共识是「reduce 偶尔必要，但大概率你不该用」。

四、itertools：惰性组合的军火库

itertools 是「迭代器的迭代器」——全是惰性、省内存、可无限。挑三个你项目里最高频的：

chain：把多个迭代器串成一条。合并多个数据源、不生成大列表。

from itertools import chain
 
for item in chain(shards_a, shards_b, shards_c):
    process(item)     # 不先把三个 shards 拼成大列表，逐段吐

groupby：按 key 分组（前提：输入已按该 key 排序）。做「日志按用户聚合」「订单按日期归档」极顺手。

from itertools import groupby
from operator import itemgetter
 
logs = sorted(raw_logs, key=itemgetter("uid"))   # 必须先排序！
for uid, group in groupby(logs, key=itemgetter("uid")):
    lines = list(group)     # 同一 uid 的日志聚一起
    flush(uid, lines)

注意那个坑：groupby 要求输入已排序，它只在相邻相同 key 处切段。没排序就分组，会得到一堆碎段。这跟 SQL 的 GROUP BY 不一样，别想当然。对照 Go：Go 你得手建 map[key][]item 自己塞，逻辑一样但多写几行；groupby 胜在惰性，大文件边读边分，不全load。

islice：迭代器版切片，不 materialize。读大文件「跳过前 N 行、取 M 行」：

from itertools import islice
 
with open("huge.log") as f:
    for line in islice(f, 100, 200):    # 第 100~199 行，不把全文件读进内存
        handle(line)

还有 count（无限递增，配 islice 截取）、cycle（循环）、product（笛卡尔积，做参数组合测试很香）。它们共同的收益：惰性 + 组合，你能把 map(chain(...))、islice(filter(...)) 像管道一样接起来，处理 GB 级数据内存却纹丝不动——这正是第 4 章「生成器惰性」的延伸。

老哥：这些和 Go 的 generator/channel 有点像？

墨叔：神似。Go 用 channel + goroutine 做流式，itertools 用同步迭代器做流式——都是「按需生产、不囤货」。但 itertools 是纯同步、单线程、零调度开销，拿来「组装有界数据流」比起 goroutine 轻得多；要真并发生产，还是得上 asyncio 或线程（第 15、16 章）。选武器看你要「串行惰性管道」还是「并行生产者」。

五、贴近项目的例子：配置驱动的批量校验管道

把 map/filter/partial/islice 揉一起，写一个「读大文件、按规则校验、分批入库」的管道，能直接搬进 ETL 脚本。

from functools import lru_cache, partial
from itertools import islice, chain
 
@lru_cache(maxsize=512)
def normalize_code(raw: str) -> str:
    return raw.strip().upper()          # 假设还有查表逻辑，缓存省重复计算
 
def is_valid(code: str) -> bool:
    return len(code) == 6 and code.isalnum()
 
def load_chunks(paths):
    for p in paths:
        with open(p, encoding="utf-8") as f:
            yield from (line.rstrip("\n") for line in f)
 
def pipeline(paths, batch=1000):
    # 拼接多文件 -> 规整 -> 过滤非法 -> 分批
    codes = map(normalize_code, chain.from_iterable(load_chunks(p) for p in paths))
    valid = filter(is_valid, codes)
    it = iter(valid)
    while batch_items := list(islice(it, batch)):   # 海象算子按批取
        store(batch_items)
 
scan = partial(pipeline, batch=500)    # 偏函数：固化批大小，得到「只收 paths」的扫描器
scan(["a.log", "b.log"])

这里：chain.from_iterable 惰性拼接多文件、map 用现成纯函数规整、filter 用现成谓词、islice 按批切、partial 固化批大小。整条管道全程不把全量数据摊进内存——十 GB 日志也能跑。对照 Go，你得自己 pipe 多个 goroutine + channel 做同样的事，代码量翻几倍。

六、收益与代价总账（对照 Go）

把整章打个表，方便你决策：

工具	用它的收益	代价/雷区	Go 对照
推导式	可读、灵活、内建	立即建列表占内存	for + append
map/filter	复用现成函数时短	lambda 可读性差，不如推导式	自写泛型或手写循环
lru_cache	一行白嫖缓存，零侵入	只纯函数、参数须可哈希、无 TTL	sync.Map+key+淘汰，可控但手写
partial	固定参数零样板	闭包更直接时没必要	闭包 fn := ...
reduce	折叠语义显式	可读差，能用 sum 等内建就别用	for 累加大同小异
itertools	惰性组合、省内存	groupby 需先排序等坑	channel+goroutine 流式

墨叔：最后送你一句——函数式工具是「少写循环」的刀，不是「显得高级」的装饰。Go 工程师的本能是「循环清楚、类型清楚」，到 Python 别走极端去追求 reduce 套 map 的炫技。默认推导式、贵计算上 lru_cache、流式处理上 itertools、固定参数上 partial，其余场景老老实实写循环——循环在 Python 里不丢人，绕晕同事才丢人。

七、functools.singledispatch：按类型分派的函数

墨叔：函数式箱里还有一颗被低估的明珠 singledispatch——它让你写「根据第一个参数的类型，自动选不同实现」的泛型函数。听着耳熟？Go 工程师会说「这不就是方法的类型 switch 吗」。

场景：你有个 serialize(obj) 想把各种东西转成 JSON 友好结构。整型、列表、自定义类行为各不同。朴素写法是一长串 isinstance 链；singledispatch 把分支收成「同名函数的多个重载」：

from functools import singledispatch
import datetime, json
 
@singledispatch
def serialize(obj):
    raise TypeError(f"不支持的类型: {type(obj)}")
 
@serialize.register
def _(obj: int):
    return obj
 
@serialize.register
def _(obj: datetime.date):
    return obj.isoformat()          # 日期→字符串
 
@serialize.register
def _(obj: list):
    return [serialize(x) for x in obj]   # 递归处理元素
 
print(serialize(42))                     # 42
print(serialize(datetime.date(2026, 1, 1)))   # "2026-01-01"
print(serialize([1, datetime.date(2026, 1, 1)]))  # [1, "2026-01-01"]

@singledispatch 装饰基础实现（处理「未知类型」），每个 @serialize.register 用注解标注参数类型来挂载该类型的实现。调用 serialize(x) 时，运行时按 x 的实际类型挑实现——类型是「运行时才定」的，所以这是动态分派（对照 Go 的 type switch 也是运行时）。

对照 Go：Go 没泛型方法重载，你要么写 func serializeInt、func serializeDate 一堆函数，要么 switch v := obj.(type) { case int: ... case time.Time: ... }。singledispatch 的优雅在于「同名、可扩展、还能被第三方库追加 register」——比如你库的调用方想支持自己的类型，不必改你的源码，直接 @serialize.register 挂上去即可。这是「开放封闭原则」的函数式写法。

代价：① 只按第一个参数的类型分派，多参数重载做不到（那是 singledispatchmethod 也只管一个）；② 分派发生在运行时，类型复杂时有微小开销；③ 调试时要知道「到底走了哪个 register」，不如显式 if/elif 一眼看清。所以分支少、且类型固定时，isinstance 链更直接；分支多、且希望可扩展时，singledispatch 更香。

再补一个顺手的：functools.cache（Python 3.9+）是 lru_cache(maxsize=None) 的简写——无界缓存。如果你确定缓存不会无限增长（参数空间有限，比如枚举值），用 @cache 比 @lru_cache(maxsize=None) 更直白。但凡参数空间可能膨胀，老老实实用带 maxsize 的 lru_cache，别让内存悄悄涨。

八、itertools 补充：tee / permutations / combinations 实战

墨叔：第六章收尾前再补三个 itertools 里项目高频、但新手常不知道的成员，尤其做测试时极好用。

permutations 和 combinations：生成排列/组合，用来「穷举参数组合跑测试」比手写嵌套循环干净十倍。比如你要测一个排序函数对「各种输入顺序」都正确：

from itertools import permutations, combinations
 
for order in permutations([3, 1, 2]):      # 所有排列: (3,1,2)(3,2,1)...
    assert is_sorted(my_sort(list(order)))
 
# 组合：从 N 个候选里挑 2 个做对照实验
for a, b in combinations(["mysql", "pg", "sqlite"], 2):
    compare(a, b)

对照 Go：Go 没有内建排列组合，你得自己写递归或引第三方包，且生成器式惰性在 Go 里要 goroutine+channel 模拟。itertools 一行到位还惰性，N 大时也只按需要生成。

tee：把一个迭代器「复制」成 N 个独立迭代器，各自从头消费，原迭代器不被消耗掉。场景：你有一条昂贵的流式数据，想同时喂给「校验」和「落库」两条管道，又不希望原数据被第一条管道读完第二条读不到：

from itertools import tee
 
src = expensive_stream()          # 惰性、只能消费一次
a, b = tee(src, 2)
validate(a)                       # 第一条管道
persist(b)                        # 第二条管道，从头独立消费同一份流

注意代价：tee 的 N 个分支里，若某个分支消费得比别的快，落后的分支需要把「超前分支已经读过的元素」缓存下来——所以两分支速度差越大、内存越涨。它不是「零成本复制」，而是「按需缓存的复制」。如果两个分支速度接近（典型 pipeline），开销很小；若一个分支读完另一个才刚开始，等于把全量数据囤内存。所以 tee 适合「多消费者速度相近」的流式分叉，不适合「一条跑完另一条才起步」。

把 itertools 整箱收个尾：它是「惰性数据管道」的乐高——chain 拼接、islice 截取、groupby 分组、tee 分叉、permutations/combinations 穷举。配 map/filter/reduce 和 functools 那箱，你能把「数据变换」写成一条声明式、省内存、可读的链。再强调一遍铁律：推导式解决 80% 的简单变换；剩下 20% 的「惰性、组合、分组、分叉」才上 itertools；纯函数重复计算上 lru_cache；固定参数上 partial。函数式工具是刀，不是勋章，别为显高级而用。

【思考题】

1. 你给 get_config(env: str, cache: dict) 这种「读文件 + 合并默认值」的函数加了 @lru_cache，结果线上配置改了，服务还是返回旧值。可能的原因有哪些？怎么排查？如果要「配置热更新」，你该怎么改？
2. groupby 和 SQL GROUP BY 行为上的关键差异是什么？下面这段代码为什么输出不符合预期，怎么修？

   logs = [{"uid": 2}, {"uid": 1}, {"uid": 2}, {"uid": 1}]
   for uid, g in groupby(logs, key=lambda x: x["uid"]):
       print(uid, list(g))

3. 你在 Go 里习惯 for 循环把切片每个元素转成另一种结构。现在让你用 Python 实现「读 5GB JSONL、每行解析后只保留 status==200 的 URL、每 1000 条批量上报」。用本章哪几个工具最合适？给个骨架，并说明为什么不用 list 一次性加载。

【参考答案】

1. 原因：① lru_cache 按「参数」缓存，env 相同就返回旧结果，函数体内读文件的结果被缓存了，配置文件改了但参数没变，所以一直命中旧值——这是把「有外部副作用/读外部状态」的函数当纯函数缓存的典型错误。② 没清缓存，进程不重启就永远旧。排查：get_config.cache_info() 看 hits 是否远高于 misses，确认命中；打印文件 mtime 比对。改法：A. 别缓存读文件函数，把「昂贵的计算」和「读文件」分开，只缓存纯计算部分；B. 需要热更新就别用 lru_cache，改用带 TTL 的 cachetools.TTLCache 或外部缓存（Redis），或监听文件 mtime 主动 get_config.cache_clear()；C. 更稳的是把配置加载做成「显式 reload」入口，更新时调用 cache_clear()。核心教训：lru_cache 只缓存「参数决定结果」的纯函数，外部状态变了它不知道。
2. 关键差异：groupby 是「相邻相同 key 才成组」，不全局重排；SQL GROUP BY 是全局聚合。logs 里 uid 顺序是 2,1,2,1，未排序，于是 groupby 会切成 [{2}],[{1}],[{2}],[{1}] 四段，而不是按 uid 聚成两组。修复：先按 key 排序再 groupby——logs.sort(key=lambda x: x["uid"])，或 for uid, g in groupby(sorted(logs, key=itemgetter("uid")), key=itemgetter("uid"))。这就是 groupby 最常踩的坑：忘记「输入须已排序」。
3. 适用工具：load_chunks 用生成器逐行 yield（不建列表）；map(json.loads, lines) 解析；filter 留 status==200；islice 按 1000 分批；partial 固化批大小给上报函数。骨架：

   def read_jsonl(path):
       with open(path) as f:
           yield from (json.loads(l) for l in f)
   ok = filter(lambda r: r.get("status") == 200, map(json.loads, read_jsonl("big.jsonl")))
   it = iter(ok)
   report = partial(report_batch, size=1000)
   while batch := list(islice(it, 1000)):
       report([r["url"] for r in batch])

   为什么不用 list 一次性加载：5GB JSONL 全读进内存会撑爆（尤其容器内存受限），且首行到首批上报之间有巨大延迟；用生成器+迭代器全程惰性，内存恒定在「一批」的量级，边读边处理，这正是 itertools/生成器「惰性救内存」的本意（呼应第 4 章）。