二、函数与函数式编程
阶段定位 :函数是 Python 中”一等公民”(First-Class Citizen),理解函数的本质是掌握 Python 高级特性的基石。本阶段从函数定义延伸到函数式编程范式,帮你建立”以函数组合解决问题”的编程思维。
0. 本章知识地图 在深入学习之前,先用一张图看清本章的知识脉络:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 函数基础 ├── 函数是对象(赋值、传递、返回、存储) ├── 定义与 return └── 多返回值与解包 参数系统 ├── 传对象引用 ├── 位置参数 / 关键字参数 ├── 默认参数陷阱 ├── *args / **kwargs ├── 参数顺序规则 └── 类型注解 作用域 ├── LEGB 规则 ├── global / nonlocal └── 为闭包做准备 函数式工具 ├── lambda ├── 高阶函数 map / filter / reduce ├── 闭包 ├── 装饰器 ├── 递归 └── 生成器 工程实践 ├── 纯函数与副作用 ├── functools 模块 └── itertools 模块
配图占位:建议在这里放一张”Python 函数与函数式编程知识地图”思维导图,文件路径 /images/python/phase2_mindmap.png。
1. 函数定义与调用 1.1 函数的本质:一切都是对象 在 Python 中,函数是对象 。这意味着函数可以:
赋值给变量
作为参数传递给其他函数
作为其他函数的返回值
存储在数据结构(列表、字典)中
1 2 3 4 5 6 7 8 9 10 11 12 13 14 def greet (name ): return f"Hello, {name} !" say_hi = greet print (say_hi("Alice" )) print (say_hi is greet) operations = { "add" : lambda a, b: a + b, "mul" : lambda a, b: a * b, } print (operations["add" ](3 , 4 ))
函数对象的重要属性 :
属性
含义
示例
__name__
函数名
greet.__name__ → 'greet'
__doc__
文档字符串
greet.__doc__
__defaults__
默认参数值元组
用于调试
__code__
代码对象
包含字节码、局部变量名等
__call__
使对象可被调用
greet() 等价于 greet.__call__(...)
1 2 3 4 5 6 7 8 9 10 11 import inspectdef sample (a, b=10 , *args, c=20 , **kwargs ): """示例函数。""" pass print (sample.__name__) print (sample.__doc__) print (sample.__defaults__) print (sample.__kwdefaults__) print (inspect.signature(sample))
面试真题 :Python 中函数是”一等公民”体现在哪些方面?
1.2 函数定义与 return 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 def calculate (x, y, op="add" ): """执行基础运算。 Args: x: 左操作数 y: 右操作数 op: 操作类型,可选 "add" | "sub" | "mul" | "div" Returns: 运算结果 Raises: ValueError: 当 op 不支持时 ZeroDivisionError: 除零时 """ if op == "add" : return x + y if op == "sub" : return x - y if op == "mul" : return x * y if op == "div" : return x / y raise ValueError(f"不支持的操作: {op} " ) def log_message (msg ): print (f"[LOG] {msg} " ) result = log_message("test" ) print (result)
文档字符串规范(Google Style) :
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 def fetch_user (user_id, include_deleted=False ): """根据用户 ID 获取用户信息。 Args: user_id: 用户唯一标识,正整数。 include_deleted: 是否包含已删除用户,默认为 False。 Returns: dict: 用户信息字典;未找到时返回 None。 Raises: ValueError: user_id 不是正整数时。 """ if not isinstance (user_id, int ) or user_id <= 0 : raise ValueError("user_id 必须是正整数" )
工程建议 :
函数只做一件事(Single Responsibility),过长时考虑拆分。
函数名应当是动词或动词短语:get_user()、validate_email()、send_notification()。
返回 None 要明确,不要依赖隐式返回。
避免使用隐式返回作为有意义的业务结果。
1.3 多返回值与解包 1 2 3 4 5 6 7 8 9 10 11 def get_min_max (numbers ): """同时返回最小值和最大值。""" if not numbers: return None , None return min (numbers), max (numbers) minimum, maximum = get_min_max([3 , 1 , 4 , 1 , 5 , 9 ]) print (minimum, maximum) min_val, _ = get_min_max([3 , 1 , 4 ])
解包的高级用法 :
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 first, *middle, last = [1 , 2 , 3 , 4 , 5 ] print (first, middle, last) (a, b), (c, d) = (1 , 2 ), (3 , 4 ) print (a, b, c, d) from collections import namedtuplePoint = namedtuple("Point" , ["x" , "y" ]) def get_origin (): return Point(0 , 0 ) origin = get_origin() print (origin.x, origin.y)
2. 参数系统(面试高频) 2.1 Python 的传参机制:传对象引用 Python 的参数传递机制是**”传对象引用”**(Pass by Object Reference)—— 传入的是对象的引用,而非值本身。
1 2 3 调用函数时: 实参变量 ──→ 对象 ←── 形参变量 (同一个对象)
1 2 3 4 5 6 7 8 9 10 11 12 13 def try_change (x ): x = 100 a = 1 try_change(a) print (a) def append_one (lst ): lst.append(1 ) my_list = [0 ] append_one(my_list) print (my_list)
判断口诀 :
函数内对参数重新赋值 ,不影响外部(指向了新对象)。
函数内对参数原地修改 ,会影响外部(同一对象被修改)。
面试真题 :Python 是值传递还是引用传递?
答:都不是,Python 是”传对象引用”。对于不可变对象(int、str、tuple),效果类似值传递;对于可变对象(list、dict),效果类似引用传递。
2.2 参数类型总览 Python 3.8+ 支持完整的参数分类:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 def demo (pos, /, pos_or_kw, *, kw_only, **kwargs ): """参数分类演示(3.8+ 语法)。 pos: 仅限位置参数(/ 左侧) pos_or_kw: 位置或关键字参数(/ 和 * 之间) kw_only: 仅限关键字参数(* 右侧) **kwargs: 接收多余关键字参数 """ print (f"pos={pos} , pos_or_kw={pos_or_kw} , kw_only={kw_only} , kwargs={kwargs} " ) demo(1 , 2 , kw_only=3 , extra="x" ) demo(1 , pos_or_kw=2 , kw_only=3 ) demo(pos=1 , pos_or_kw=2 , kw_only=3 )
参数分类速查表 :
参数类型
语法位置
调用方式
典型用途
仅限位置
/ 左侧
只能位置
内部参数、避免破坏兼容性
位置或关键字
/ 和 * 之间
任意
最常见的参数
仅限关键字
* 右侧
只能关键字
提高可读性、明确语义
*args
接收多余位置
自动打包
变长参数、转发
**kwargs
接收多余关键字
自动打包
变长参数、配置转发
配图占位:建议放一张”Python 参数类型示意图”,文件路径 /images/python/phase2_parameters.png。
2.3 位置参数与关键字参数 1 2 3 4 5 6 7 8 9 10 11 12 13 def create_user (name, age, active=True , role="user" ): return {"name" : name, "age" : age, "active" : active, "role" : role} create_user("Alice" , 30 ) create_user(name="Bob" , age=25 , role="admin" ) create_user("Carol" , 28 , role="moderator" ) create_user(name="Dave" , 35 )
工程建议 :调用函数时,超过 2-3 个参数尽量使用关键字参数,提高可读性:
1 2 3 4 5 set_timeout(30 , True , False ) set_timeout(seconds=30 , retry=True , fail_silent=False )
2.4 默认参数值 1 2 3 4 5 6 def connect (host, port=3306 , timeout=30 ): print (f"Connecting to {host} :{port} , timeout={timeout} s" ) connect("localhost" ) connect("localhost" , 5432 ) connect("localhost" , timeout=60 )
经典陷阱:可变对象作为默认参数
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 def append_item (item, items=[] ): items.append(item) return items print (append_item(1 )) print (append_item(2 )) def append_item_safe (item, items=None ): if items is None : items = [] items.append(item) return items print (append_item_safe(1 )) print (append_item_safe(2 ))
原理剖析 :
1 2 3 4 5 6 7 8 函数定义时: def append_item(item, items=[]) -> 创建一个空列表对象 L 第一次 append_item(1): items 指向 L,L 变为 [1] 第二次 append_item(2): items 仍然指向 L,L 变为 [1, 2]
默认参数值在函数定义时(模块加载时)求值一次,后续调用复用同一个对象。不可变类型(int、str、tuple)作为默认值是安全的,但可变类型(list、dict、set)必须使用 None 作为占位符。
面试真题 :下面代码输出什么?为什么?
1 2 3 4 5 6 def add_end (L=[] ): L.append("END" ) return L print (add_end())print (add_end())
答:两次都输出 ['END', 'END'],因为默认参数在定义时求值,列表被所有调用共享。
2.5 可变参数:*args 与 **kwargs 1 2 3 4 5 6 7 8 9 def flexible (*args, **kwargs ): """*args 接收多余的位置参数,打包成元组。 **kwargs 接收多余的关键字参数,打包成字典。""" print (f"位置参数: {args} " ) print (f"关键字参数: {kwargs} " ) flexible(1 , 2 , 3 , name="Alice" , age=30 )
拆包(Unpacking)调用 :
1 2 3 4 5 6 7 8 9 10 11 12 def sum_three (a, b, c ): return a + b + c nums = [1 , 2 , 3 ] print (sum_three(*nums)) config = {"a" : 1 , "b" : 2 , "c" : 3 } print (sum_three(**config)) data = [1 , 2 ] print (sum_three(*data, c=3 ))
实战:函数参数转发
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 import loggingdef logged (func ): """装饰器:记录函数调用日志(后续会详细讲解装饰器)。""" def wrapper (*args, **kwargs ): logging.info(f"Calling {func.__name__} with args={args} , kwargs={kwargs} " ) return func(*args, **kwargs) return wrapper @logged def compute (x, y, operation="add" ): if operation == "add" : return x + y return x - y compute(10 , 5 , operation="sub" )
2.6 参数顺序规则(必须遵守) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 def correct (a, b=2 , /, c=3 , d=4 , *, e=5 , f=6 ): """正确顺序: 1. 仅限位置参数(可有默认值) 2. / 分隔符 3. 位置或关键字参数(可有默认值) 4. * 分隔符 5. 仅限关键字参数(可有默认值) 6. **kwargs """ pass
2.7 类型注解(Type Hints) Python 3.5+ 支持类型注解,虽然运行时不强制,但能显著提升代码可维护性。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 from typing import List , Dict , Optional , Union , Callable def greet (name: str , times: int = 1 ) -> str : return (f"Hello, {name} !\n" ) * times def process_users (users: List [Dict [str , Union [str , int ]]] ) -> List [str ]: return [user["name" ] for user in users if user.get("active" )] def find_user (user_id: int ) -> Optional [Dict ]: pass def apply_operation (a: int , b: int , op: Callable [[int , int ], int ] ) -> int : return op(a, b)
常用类型注解速查 :
类型
含义
示例
int, str, float, bool
基本类型
x: int
List[T]
列表
items: List[int]
Dict[K, V]
字典
config: Dict[str, Any]
Optional[T]
T 或 None
name: Optional[str]
Union[T1, T2]
多种类型之一
value: Union[int, str]
Callable[[Arg], Ret]
可调用对象
fn: Callable[[int], str]
Any
任意类型
data: Any
面试真题 :Python 类型注解在运行时会检查吗?
答:不会。类型注解主要用于静态类型检查工具(如 mypy)、IDE 提示和文档。运行时不会进行类型强制校验。
3. 作用域与 LEGB 规则 3.1 命名空间与作用域层级 Python 查找变量的顺序遵循 LEGB 规则 :
层级
含义
范围
L ocal
局部作用域
当前函数内部
E nclosing
嵌套作用域
外层(闭包)函数
G lobal
全局作用域
当前模块顶层
B uilt-in
内置作用域
builtins 模块
1 2 3 4 5 6 7 8 9 10 11 12 x = "global" def outer (): x = "enclosing" def inner (): x = "local" print (x) inner() outer()
配图占位:建议放一张”LEGB 作用域查找顺序”示意图,文件路径 /images/python/phase2_legb.png。
3.2 global 声明 1 2 3 4 5 6 7 8 counter = 0 def increment (): global counter counter += 1 increment() print (counter)
工程建议 :
尽量避免使用 global,它破坏函数的纯粹性和可测试性。
需要共享状态时,考虑使用类封装或使用闭包。
3.3 nonlocal 声明 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 def make_counter (): count = 0 def increment (): nonlocal count count += 1 return count return increment counter_a = make_counter() print (counter_a()) print (counter_a()) counter_b = make_counter() print (counter_b())
global vs nonlocal 对比 :
特性
global
nonlocal
修改目标
模块级全局变量
外层(闭包)函数的局部变量
使用场景
全局状态管理(不推荐)
闭包状态保持
报错条件
变量不存在时创建
变量必须已存在在外层作用域
3.4 locals() 与 globals() 1 2 3 4 5 6 def show_scope (): x = 10 print (locals ()) show_scope() print (globals ().keys())
注意事项 :
locals() 返回的字典不要修改,行为不可靠。
globals() 返回的字典可以修改,但不推荐。
4. 匿名函数 lambda 4.1 基本语法与使用场景 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 square = lambda x: x ** 2 print (square(5 )) add = lambda x, y: x + y print (add(3 , 4 )) users = [ {"name" : "Alice" , "age" : 30 }, {"name" : "Bob" , "age" : 25 }, {"name" : "Carol" , "age" : 35 }, ] users.sort(key=lambda u: u["age" ]) print ([u["name" ] for u in users])
4.2 lambda 的限制
只能包含一个表达式,不能写语句(如赋值、循环)。
没有文档字符串,调试时显示为 <lambda>。
不能包含复杂的错误处理。
4.3 lambda vs def
特性
lambda
def
名称
匿名
有名称
复杂度
单行表达式
任意复杂
可读性
简单场景好
复杂场景好
调试
显示 <lambda>
显示真实函数名
文档字符串
无
有
工程建议 :
简单的一行表达式用 lambda,复杂逻辑用 def 定义命名函数。
若 lambda 超过一行,说明它不应该用 lambda。
1 2 3 4 5 6 7 process = lambda data: [x.strip().lower() for x in data if x.startswith("http" )] def extract_urls (data ): """从数据中筛选并处理 URL。""" return [x.strip().lower() for x in data if x.startswith("http" )]
面试真题 :lambda 能包含多个语句吗?
答:不能。lambda 只能包含一个表达式,不能包含语句(如赋值、循环、try/except)。
5. 高阶函数 高阶函数 :接收函数作为参数,或返回函数的函数。
5.1 map、filter、reduce 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 from functools import reducenums = [1 , 2 , 3 , 4 , 5 ] squares = list (map (lambda x: x ** 2 , nums)) print (squares) squares = [x ** 2 for x in nums] evens = list (filter (lambda x: x % 2 == 0 , nums)) print (evens) evens = [x for x in nums if x % 2 == 0 ] product = reduce(lambda acc, x: acc * x, nums, 1 ) print (product) from operator import mulproduct = 1 for x in nums: product *= x import mathproduct = math.prod(nums)
工程建议 :
优先使用列表推导式 和生成器表达式 替代 map/filter,可读性更好。
reduce 的使用场景较少,能不用就不用。
5.2 使用 operator 模块避免 lambda 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 from operator import add, mul, itemgetter, attrgetterprint (reduce(add, [1 , 2 , 3 , 4 ])) users = [{"name" : "Alice" , "age" : 30 }, {"name" : "Bob" , "age" : 25 }] users.sort(key=itemgetter("age" )) class Person : def __init__ (self, name, age ): self .name = name self .age = age people = [Person("Alice" , 30 ), Person("Bob" , 25 )] people.sort(key=attrgetter("age" ))
5.3 自定义高阶函数 1 2 3 4 5 6 7 8 9 10 11 def create_multiplier (factor ): """返回一个乘以 factor 的函数。""" def multiply (x ): return x * factor return multiply triple = create_multiplier(3 ) print (triple(5 )) double = create_multiplier(2 ) print (double(7 ))
5.4 函数组合 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 from functools import reducedef compose (*functions ): """函数组合:compose(f, g, h)(x) == f(g(h(x)))""" def composed (value ): return reduce(lambda acc, fn: fn(acc), reversed (functions), value) return composed def add_one (x ): return x + 1 def double (x ): return x * 2 def to_str (x ): return str (x)transform = compose(to_str, double, add_one) print (transform(3 ))
配图占位:建议放一张”函数组合流程图”,文件路径 /images/python/phase2_compose.png。
6. 闭包(Closure) 6.1 闭包的定义 闭包 :一个函数记住了它被创建时的环境(外层作用域的变量),即使外层函数已经执行完毕。
1 2 3 4 5 6 7 8 9 10 11 def make_power (exponent ): """创建计算 x^n 的函数。""" def power (base ): return base ** exponent return power square = make_power(2 ) cube = make_power(3 ) print (square(4 )) print (cube(3 ))
闭包的本质 :
1 2 3 4 5 6 7 8 9 10 11 12 make_power(2) 调用时: 创建局部变量 exponent = 2 创建函数 power power.__closure__ 中保存了对 exponent 的引用 make_power 返回后: 局部变量 exponent 本应被销毁 但因为闭包引用了它,所以被保留 square(4) 调用时: power 通过闭包找到 exponent = 2 计算 4 ** 2 = 16
1 2 print (square.__closure__) print (square.__closure__[0 ].cell_contents)
6.2 闭包的实际应用 1 2 3 4 5 6 7 8 9 10 11 def make_logger (level ): """创建带有固定日志级别的日志函数。""" def log (message ): print (f"[{level.upper()} ] {message} " ) return log info_log = make_logger("info" ) error_log = make_logger("error" ) info_log("服务启动成功" ) error_log("数据库连接失败" )
6.3 闭包的陷阱:延迟绑定 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 functions = [] for i in range (3 ): def func (): return i functions.append(func) print (functions[0 ]()) print (functions[1 ]()) print (functions[2 ]()) functions = [] for i in range (3 ): def func (x=i ): return x functions.append(func) print (functions[0 ]()) print (functions[1 ]()) print (functions[2 ]())
面试真题 :为什么下面的 lambda 列表输出都是 4?
1 2 funcs = [(lambda : i) for i in range (5 )] print ([f() for f in funcs])
答:与闭包延迟绑定同理,所有 lambda 都引用了同一个 i,循环结束时 i=4。
7. 装饰器(Decorator) 装饰器是闭包的高级应用,用于在不修改原函数代码的前提下,增强函数功能 。
配图占位:建议放一张”装饰器工作原理”示意图,文件路径 /images/python/phase2_decorator.png。
7.1 装饰器基础 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 import functoolsimport timedef timing (func ): """计算函数执行时间的装饰器。""" @functools.wraps(func ) def wrapper (*args, **kwargs ): start = time.perf_counter() result = func(*args, **kwargs) elapsed = time.perf_counter() - start print (f"{func.__name__} 耗时: {elapsed:.4 f} s" ) return result return wrapper @timing def slow_function (n ): """模拟耗时操作。""" time.sleep(0.1 ) return n * 2 result = slow_function(5 )
装饰器执行过程 :
1 2 3 4 5 6 7 8 9 10 11 12 @timing def slow_function(n): ... 等价于: slow_function = timing(slow_function) 调用 slow_function(5) 时: 实际调用 wrapper(5) wrapper 记录开始时间 wrapper 调用原 slow_function(5) wrapper 记录结束时间并打印 wrapper 返回结果
1 2 3 4 5 6 7 8 9 10 11 12 @timing def my_func (): """我的函数。""" pass print (my_func.__name__) print (my_func.__doc__) print (my_func.__name__) print (my_func.__doc__)
wraps 到底做了什么?
1 2 3 4 5 6 7 wrapper.__name__ = func.__name__ wrapper.__doc__ = func.__doc__ wrapper.__module__ = func.__module__ wrapper.__qualname__ = func.__qualname__ wrapper.__dict__.update(func.__dict__) wrapper.__wrapped__ = func
7.3 带参数的装饰器 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 def retry (max_attempts=3 , delay=1 ): """带参数的装饰器:失败时自动重试。""" def decorator (func ): @functools.wraps(func ) def wrapper (*args, **kwargs ): for attempt in range (1 , max_attempts + 1 ): try : return func(*args, **kwargs) except Exception as e: print (f"第 {attempt} 次尝试失败: {e} " ) if attempt < max_attempts: time.sleep(delay) raise Exception(f"{max_attempts} 次尝试全部失败" ) return wrapper return decorator @retry(max_attempts=3 , delay=0.5 ) def fetch_data (): """模拟可能失败的请求。""" import random if random.random() < 0.7 : raise ConnectionError("连接失败" ) return "数据"
带参数装饰器的执行过程 :
1 2 3 4 5 6 7 8 9 @retry(max_attempts=3) def fetch_data(): ... 等价于: fetch_data = retry(max_attempts=3)(fetch_data) retry(max_attempts=3) 返回 decorator decorator(fetch_data) 返回 wrapper fetch_data 指向 wrapper
7.4 类装饰器 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 class CountCalls : """统计函数被调用次数的类装饰器。""" def __init__ (self, func ): functools.update_wrapper(self , func) self .func = func self .count = 0 def __call__ (self, *args, **kwargs ): self .count += 1 print (f"{self.func.__name__} 被调用了 {self.count} 次" ) return self .func(*args, **kwargs) @CountCalls def say_hello (): print ("Hello!" ) say_hello() say_hello()
7.5 多个装饰器堆叠 1 2 3 4 5 6 7 8 @decorator_a @decorator_b @decorator_c def my_func (): pass
配图占位:建议放一张”多个装饰器堆叠执行顺序”示意图,文件路径 /images/python/phase2_decorators_stack.png。
7.6 常用装饰器模式 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 def require_login (func ): @functools.wraps(func ) def wrapper (user, *args, **kwargs ): if not user.is_authenticated: raise PermissionError("请先登录" ) return func(user, *args, **kwargs) return wrapper def simple_cache (func ): cache = {} @functools.wraps(func ) def wrapper (*args ): if args not in cache: cache[args] = func(*args) return cache[args] return wrapper def validate_types (**types ): def decorator (func ): @functools.wraps(func ) def wrapper (*args, **kwargs ): bound = inspect.signature(func).bind(*args, **kwargs) bound.apply_defaults() for name, expected_type in types.items(): if name in bound.arguments: value = bound.arguments[name] if not isinstance (value, expected_type): raise TypeError(f"{name} 期望 {expected_type} ,实际 {type (value)} " ) return func(*args, **kwargs) return wrapper return decorator import inspect@validate_types(name=str , age=int ) def register (name, age ): print (f"注册用户: {name} , {age} " )
7.7 内置装饰器 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 class Math : @staticmethod def add (a, b ): return a + b class User : count = 0 def __init__ (self, name ): self .name = name User.count += 1 @classmethod def get_count (cls ): return cls.count class Circle : def __init__ (self, radius ): self .radius = radius @property def area (self ): return 3.14159 * self .radius ** 2 c = Circle(5 ) print (c.area)
8. 递归 8.1 递归基础 1 2 3 4 5 6 7 def factorial (n ): """计算阶乘(递归版)。""" if n <= 1 : return 1 return n * factorial(n - 1 ) print (factorial(5 ))
递归三要素 :
终止条件 :n <= 1 时返回 1。
递归调用 :factorial(n - 1)。
状态传递 :通过参数 n 传递状态。
1 2 3 4 5 6 factorial(5) └── 5 * factorial(4) └── 4 * factorial(3) └── 3 * factorial(2) └── 2 * factorial(1) └── 1
8.2 递归的局限 1 2 3 4 5 6 7 8 9 import sysprint (sys.getrecursionlimit()) def infinite_recursion (n ): return infinite_recursion(n + 1 )
工程建议 :
Python 不擅长深度递归,能用循环解决的不要用递归。
树/图的遍历等天然递归结构,考虑使用栈模拟递归或改用迭代。
8.3 尾递归(了解即可) 1 2 3 4 5 6 7 8 9 def factorial_tail (n, acc=1 ): """尾递归版阶乘(acc 为累积器)。 注意:CPython 不支持尾递归优化,这仍然会使用调用栈! 但在支持尾递归优化的语言(如 Scheme、Erlang)中,这不会增加栈深度。 """ if n <= 1 : return acc return factorial_tail(n - 1 , n * acc)
8.4 记忆化递归 1 2 3 4 5 6 7 8 9 from functools import lru_cache@lru_cache(maxsize=None ) def fibonacci (n ): if n < 2 : return n return fibonacci(n - 1 ) + fibonacci(n - 2 ) print (fibonacci(100 ))
记忆化的作用 :
避免重复计算,将时间复杂度从 O(2^n) 降到 O(n)。
使用 lru_cache 自动缓存结果。
9. 生成器(Generator) 9.1 生成器函数与 yield 生成器是一种特殊的迭代器,使用 yield 关键字可以暂停 函数执行并返回一个值,下次调用时从暂停处继续。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 def count_up_to (n ): """生成 1 到 n 的整数。""" count = 1 while count <= n: yield count count += 1 counter = count_up_to(3 ) print (type (counter)) print (next (counter)) print (next (counter)) print (next (counter)) for num in count_up_to(5 ): print (num, end=" " )
配图占位:建议放一张”生成器 yield 暂停与恢复”示意图,文件路径 /images/python/phase2_generator.png。
9.2 生成器 vs 列表
特性
列表
生成器
内存占用
存储所有元素
惰性计算,每次只存一个
可迭代次数
无限次
只能迭代一次
随机访问
支持 lst[i]
不支持,只能顺序访问
长度
len() 可知
不知道总长度
1 2 3 4 5 6 7 8 9 10 squares_list = [x ** 2 for x in range (10_000_000 )] squares_gen = (x ** 2 for x in range (10_000_000 )) for sq in squares_gen: if sq > 100 : break
9.3 yield from(3.3+) 1 2 3 4 5 6 7 8 9 10 def flatten (nested ): """展平嵌套的可迭代对象。""" for item in nested: if isinstance (item, (list , tuple )): yield from flatten(item) else : yield item nested = [1 , [2 , [3 , 4 ]], 5 ] print (list (flatten(nested)))
yield from 的作用 :
将子生成器的值直接转发给调用方。
简化了嵌套生成器的写法。
在协程中还可以转发 send() 和 throw()。
9.4 生成器的双向通信:send() 方法 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 def running_average (): """动态计算平均值。""" total = 0 count = 0 average = None while True : value = yield average total += value count += 1 average = total / count avg = running_average() next (avg) print (avg.send(10 )) print (avg.send(20 )) print (avg.send(30 ))
生成器协议总结 :
方法
作用
next(g)
获取下一个值
g.send(value)
向生成器发送值,并获取下一个值
g.throw(exc)
向生成器抛异常
g.close()
关闭生成器
9.5 生成器表达式 1 2 3 4 5 6 7 8 9 squares = [x ** 2 for x in range (100 )] squares_gen = (x ** 2 for x in range (100 )) total = sum (x ** 2 for x in range (100 )) print (total)
10. 函数式编程工程实践 10.1 纯函数与副作用 纯函数 :相同输入永远产生相同输出,且没有副作用。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 def add (a, b ): return a + b total = 0 def add_to_total (value ): global total total += value import randomdef random_add (a ): return a + random.randint(1 , 10 )
纯函数的优势 :
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 from functools import partial, reduce, lru_cache, cmp_to_keybasetwo = partial(int , base=2 ) print (basetwo("1010" )) product = reduce(lambda x, y: x * y, [1 , 2 , 3 , 4 ]) @lru_cache(maxsize=128 ) def expensive (n ): return n * n def compare_length (a, b ): return len (a) - len (b) words = ["apple" , "pie" , "banana" ] words.sort(key=cmp_to_key(compare_length))
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 import itertoolscount = itertools.count(1 , 2 ) cycle = itertools.cycle("AB" ) perms = list (itertools.permutations([1 , 2 , 3 ])) combs = list (itertools.combinations([1 , 2 , 3 ], 2 )) for key, group in itertools.groupby("AAABBBCC" ): print (key, list (group))
配图占位:建议放一张”Python functools 与 itertools 工具速查”示意图,文件路径 /images/python/phase2_functools_itertools.png。
附录 A:第二阶段自检清单 在继续学习第三阶段之前,请确保你能:
附录 B:面试题精选
**Python 的传参机制是什么?**传对象引用。
**可变默认参数有什么陷阱?**默认参数在函数定义时求值,可变对象被所有调用共享。
*args 和 **kwargs 的星号分别做了什么?* 拆包序列,** 拆包字典。
global 和 nonlocal 的区别?global 修改全局变量,nonlocal 修改外层闭包变量。
**什么是闭包?**函数记住其创建时的环境。
**闭包延迟绑定问题如何修复?**使用默认参数 x=i 绑定当前值。
**装饰器的本质是什么?**返回函数的高阶函数。
**@functools.wraps 的作用?**保留原函数的元数据。
**生成器和列表的区别?**生成器惰性求值、省内存、只能迭代一次。
**yield from 有什么用?**委托子生成器,简化嵌套生成器。
工作实战场景 场景 1:处理超大日志文件 需求 :分析 10GB 的日志文件,提取所有 ERROR 级别的日志行。
问题 :直接 read() 会撑爆内存。
解决方案 :使用生成器逐行处理
1 2 3 4 5 6 7 8 9 10 11 12 13 def parse_large_log (filepath ): """生成器:逐行读取大文件,避免内存溢出""" with open (filepath, 'r' , encoding='utf-8' ) as f: for line in f: if 'ERROR' in line: yield line.strip() for error_line in parse_large_log('/var/log/app.log' ): process_error(error_line) error_count = sum (1 for _ in parse_large_log('/var/log/app.log' ))
场景 2:配置验证器 需求 :验证用户配置,返回所有错误而非遇到第一个错误就停止。
解决方案 :使用生成器收集所有错误
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 def validate_config (config ): """验证配置,返回所有错误""" def check_required_fields (): required = ['host' , 'port' , 'database' ] for field in required: if field not in config: yield f"缺少必填字段: {field} " def check_port_range (): port = config.get('port' ) if port and not (1 <= port <= 65535 ): yield f"端口 {port} 超出有效范围 (1-65535)" def check_timeout (): timeout = config.get('timeout' , 30 ) if timeout <= 0 : yield f"超时时间必须为正数: {timeout} " yield from check_required_fields() yield from check_port_range() yield from check_timeout() config = {'host' : 'localhost' , 'port' : 99999 , 'timeout' : -5 } errors = list (validate_config(config)) if errors: print ("配置错误:" ) for err in errors: print (f" - {err} " ) else : print ("配置有效" )
场景 3:缓存装饰器 需求 :为频繁调用的函数添加缓存,减少数据库查询。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 import timefrom functools import wrapsdef cache_result (ttl_seconds=60 ): """带 TTL 的缓存装饰器""" cache = {} def decorator (func ): @wraps(func ) def wrapper (*args, **kwargs ): key = (args, tuple (sorted (kwargs.items()))) if key in cache: result, timestamp = cache[key] if time.time() - timestamp < ttl_seconds: return result result = func(*args, **kwargs) cache[key] = (result, time.time()) return result return wrapper return decorator @cache_result(ttl_seconds=30 ) def get_user_info (user_id ): """模拟数据库查询""" print (f"查询数据库: user_id={user_id} " ) return {'id' : user_id, 'name' : f'User{user_id} ' } user1 = get_user_info(1 ) user1_again = get_user_info(1 )
场景 4:重试装饰器 需求 :为不稳定的外部 API 调用添加自动重试机制。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 import timeimport randomfrom functools import wrapsdef retry (max_attempts=3 , delay=1 , backoff=2 , exceptions=(Exception, ) ): """重试装饰器 Args: max_attempts: 最大重试次数 delay: 初始延迟(秒) backoff: 延迟增长因子 exceptions: 捕获的异常类型 """ def decorator (func ): @wraps(func ) def wrapper (*args, **kwargs ): current_delay = delay last_exception = None for attempt in range (max_attempts): try : return func(*args, **kwargs) except exceptions as e: last_exception = e if attempt < max_attempts - 1 : time.sleep(current_delay) current_delay *= backoff raise last_exception return wrapper return decorator @retry(max_attempts=3 , delay=0.5 , exceptions=(ConnectionError, TimeoutError ) ) def call_external_api (url ): """调用外部 API""" if random.random() < 0.5 : raise ConnectionError("网络错误" ) return {"status" : "ok" } try : result = call_external_api("https://api.example.com/data" ) print (f"成功: {result} " ) except Exception as e: print (f"最终失败: {e} " )
实战项目:任务调度器 通过一个完整的任务调度器项目,综合运用函数式编程知识。
项目目标 创建一个支持以下功能的任务调度器:
定时执行任务
支持任务依赖
支持任务重试
记录任务状态
完整代码 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 """ 任务调度器 - 第二阶段实战项目 知识点:闭包、装饰器、生成器、高阶函数 """ import timefrom datetime import datetimefrom functools import wrapsfrom typing import Callable , List , Dict , Any , Optional from dataclasses import dataclassfrom enum import Enumclass TaskStatus (Enum ): """任务状态""" PENDING = "pending" RUNNING = "running" SUCCESS = "success" FAILED = "failed" RETRYING = "retrying" @dataclass class TaskResult : """任务执行结果""" task_name: str status: TaskStatus start_time: datetime end_time: Optional [datetime] = None result: Any = None error: Optional [str ] = None attempts: int = 1 def timer (func ): """计时装饰器""" @wraps(func ) def wrapper (*args, **kwargs ): start = time.time() result = func(*args, **kwargs) elapsed = time.time() - start print (f"[Timer] {func.__name__} 执行耗时: {elapsed:.3 f} s" ) return result return wrapper def retry_on_failure (max_retries=3 , delay=1 ): """重试装饰器""" def decorator (func ): @wraps(func ) def wrapper (*args, **kwargs ): last_error = None for attempt in range (max_retries): try : return func(*args, **kwargs) except Exception as e: last_error = e if attempt < max_retries - 1 : print (f"[Retry] {func.__name__} 第{attempt + 1 } 次失败,{delay} 秒后重试..." ) time.sleep(delay) raise last_error return wrapper return decorator class TaskScheduler : """任务调度器""" def __init__ (self ): self .tasks: Dict [str , Callable ] = {} self .dependencies: Dict [str , List [str ]] = {} self .history: List [TaskResult] = [] def register (self, name: str , depends_on: List [str ] = None ): """注册任务装饰器""" def decorator (func ): self .tasks[name] = func self .dependencies[name] = depends_on or [] return func return decorator def _check_dependencies (self, task_name: str ) -> bool : """检查任务依赖是否满足""" for dep in self .dependencies.get(task_name, []): dep_results = [r for r in self .history if r.task_name == dep] if not dep_results or dep_results[-1 ].status != TaskStatus.SUCCESS: return False return True def _get_execution_order (self ) -> List [str ]: """获取拓扑排序后的执行顺序""" visited = set () order = [] def visit (task_name ): if task_name in visited: return visited.add(task_name) for dep in self .dependencies.get(task_name, []): visit(dep) order.append(task_name) for task_name in self .tasks: visit(task_name) return order def run_task (self, name: str ) -> TaskResult: """执行单个任务""" start_time = datetime.now() if not self ._check_dependencies(name): return TaskResult( task_name=name, status=TaskStatus.FAILED, start_time=start_time, end_time=datetime.now(), error="依赖任务未完成" ) try : result = self .tasks[name]() return TaskResult( task_name=name, status=TaskStatus.SUCCESS, start_time=start_time, end_time=datetime.now(), result=result ) except Exception as e: return TaskResult( task_name=name, status=TaskStatus.FAILED, start_time=start_time, end_time=datetime.now(), error=str (e) ) @timer def run_all (self ): """执行所有任务(按依赖顺序)""" order = self ._get_execution_order() print (f"\n[Scheduler] 执行顺序: {' → ' .join(order)} \n" ) for task_name in order: print (f"[Scheduler] 开始执行: {task_name} " ) result = self .run_task(task_name) self .history.append(result) if result.status == TaskStatus.SUCCESS: print (f"[Scheduler] ✓ {task_name} 完成" ) else : print (f"[Scheduler] ✗ {task_name} 失败: {result.error} " ) def get_summary (self ) -> str : """获取执行摘要""" success = sum (1 for r in self .history if r.status == TaskStatus.SUCCESS) failed = sum (1 for r in self .history if r.status == TaskStatus.FAILED) total_time = sum ( (r.end_time - r.start_time).total_seconds() for r in self .history if r.end_time ) return f""" 执行摘要: 总任务数: {len (self.history)} 成功: {success} 失败: {failed} 总耗时: {total_time:.2 f} s """ scheduler = TaskScheduler() @scheduler.register("fetch_data" ) @retry_on_failure(max_retries=2 ) def fetch_data (): """获取数据""" print (" → 从数据库获取数据..." ) time.sleep(0.5 ) return {"users" : [1 , 2 , 3 ], "orders" : [100 , 200 ]} @scheduler.register("process_data" , depends_on=["fetch_data" ] ) def process_data (): """处理数据""" print (" → 处理数据..." ) time.sleep(0.3 ) return {"processed" : 6 } @scheduler.register("generate_report" , depends_on=["process_data" ] ) @timer def generate_report (): """生成报告""" print (" → 生成报告..." ) time.sleep(0.2 ) return "report_20240101.pdf" @scheduler.register("send_notification" , depends_on=["generate_report" ] ) def send_notification (): """发送通知""" print (" → 发送邮件通知..." ) time.sleep(0.1 ) return "sent" if __name__ == "__main__" : print ("=" * 50 ) print (" 任务调度器 v1.0" ) print ("=" * 50 ) scheduler.run_all() print (scheduler.get_summary())
运行示例 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 ================================================== 任务调度器 v1.0 ================================================== [Scheduler] 执行顺序: fetch_data → process_data → generate_report → send_notification [Scheduler] 开始执行: fetch_data → 从数据库获取数据... [Scheduler] ✓ fetch_data 完成 [Scheduler] 开始执行: process_data → 处理数据... [Scheduler] ✓ process_data 完成 [Scheduler] 开始执行: generate_report → 生成报告... [Timer] generate_report 执行耗时: 0.201s [Scheduler] ✓ generate_report 完成 [Scheduler] 开始执行: send_notification → 发送邮件通知... [Scheduler] ✓ send_notification 完成 [Timer] run_all 执行耗时: 1.104s 执行摘要: 总任务数: 4 成功: 4 失败: 0 总耗时: 1.10s
知识点覆盖
功能
涉及知识点
装饰器
参数化装饰器、@wraps、嵌套装饰器
闭包
装饰器内部状态保持
生成器
任务执行顺序生成
高阶函数
register 返回装饰器
数据类
@dataclass 定义结果对象
枚举
Enum 定义状态
进阶练习 尝试为任务调度器添加以下功能:
支持定时任务(cron 表达式)
支持并行执行无依赖的任务
添加任务进度追踪
支持任务超时设置
添加任务取消功能
工程师寄语 :函数式编程的精髓不在于用了多少 lambda 或 map,而在于能否用纯函数 和不可变数据 来构建系统。在 Python 中,适度地借鉴函数式思想(如避免副作用、使用生成器处理大数据流),能让你的代码更健壮、更可测试。