Python 闭包机制与装饰器实现原理
闭包的核心概念
闭包（Closure）是指内部函数对外部函数局部变量的引用关系。这种机制让内部函数能够"记住"其定义时的环境状态，即使外部函数已经执行完毕。
闭包的两大特性
- 状态保持：变量长期驻留内存，供后续调用使用
- 数据隔离：避免全局命名空间污染，提升代码安全性
基础示例
def create_counter():
count = 0
def increment():
nonlocal count
count += 1
return count
return increment
counter = create_counter()
print(counter()) # 输出: 1
print(counter()) # 输出: 2
print(counter()) # 输出: 3此处的 count 变量被闭包捕获，每次调用 counter() 都能访问并修改这个私有状态，而外部代码无法直接干预。
装饰器的设计与实现
装饰器是闭包的经典应用场景，它允许在不修改原函数源码的前提下，为其注入额外功能。
典型应用场景
- 用户身份认证与权限校验
- 操作日志记录与审计追踪
- 性能监控与耗时统计
- 缓存机制与结果复用
实现原理的三要素
- 函数作为参数传递 — 实现代理执行
- 函数作为返回值 — 构建闭包结构
- 函数名重新绑定 — 完成替换操作
逐步推导过程
第一步：直接调用（原始状态）
def run_task():
print("执行业务逻辑")
run_task()第二步：需求变更（添加前置后置操作）
def with_logging(task):
print("[日志] 任务开始")
task()
print("[日志] 任务结束")
def run_task():
print("执行业务逻辑")
with_logging(run_task) # 缺点：调用方式改变第三步：保持调用方式不变（闭包方案）
def with_logging(task):
def wrapper():
print("[日志] 任务开始")
task()
print("[日志] 任务结束")
return wrapper
def run_task():
print("执行业务逻辑")
run_task = with_logging(run_task)
run_task() # 调用方式不变，功能已增强第四步：语法糖简化（@ 装饰器语法）
def with_logging(task):
def wrapper():
print("[日志] 任务开始")
task()
print("[日志] 任务结束")
return wrapper
@with_logging
def run_task():
print("执行业务逻辑")
run_task()通用装饰器模板
def decorator(fn):
def inner(*args, **kwargs):
# 前置处理
result = fn(*args, **kwargs)
# 后置处理
return result
return inner参数与返回值的完整处理
支持任意参数
使用 *args, **kwargs 实现参数透传：
def timer(func):
import time
def wrapper(*args, **kwargs):
start = time.time()
result = func(*args, **kwargs)
elapsed = time.time() - start
print(f"耗时: {elapsed:.4f}秒")
return result
return wrapper
@timer
def fetch_data(url, timeout=30):
print(f"请求: {url}")
return {"status": 200}
data = fetch_data("https://api.example.com", timeout=10)保留原函数元信息
使用 functools.wraps 修复函数标识：
from functools import wraps
def decorator(fn):
@wraps(fn)
def wrapper(*args, **kwargs):
return fn(*args, **kwargs)
return wrapper多层装饰器嵌套
多个装饰器叠加时，执行顺序为就近原则：从下往上包裹，从上往下执行。
def authenticate(fn):
def wrapper(*args, **kwargs):
print("【认证】验证用户身份...")
return fn(*args, **kwargs)
return wrapper
def validate_input(fn):
def wrapper(*args, **kwargs):
print("【校验】检查参数合法性...")
return fn(*args, **kwargs)
return wrapper
@authenticate
@validate_input
def transfer_money(amount, receiver):
print(f"转账 {amount} 元给 {receiver}")
transfer_money(1000, "Alice")执行输出：
【认证】验证用户身份... 【校验】检查参数合法性... 转账 1000 元给 Alice
带参数的装饰器
通过三层嵌套实现装饰器本身可配置：
def retry(max_attempts=3):
def decorator(fn):
def wrapper(*args, **kwargs):
for attempt in range(1, max_attempts + 1):
try:
return fn(*args, **kwargs)
except Exception as e:
print(f"第 {attempt} 次失败: {e}")
if attempt == max_attempts:
raise
return wrapper
return decorator
@retry(max_attempts=5)
def unstable_api():
import random
if random.random() < 0.7:
raise ConnectionError("连接超时")
return "成功获取数据"实战：登录状态管理
class AuthManager:
_session = {}
@classmethod
def is_logged_in(cls, user_id):
return cls._session.get(user_id, False)
@classmethod
def login(cls, user_id):
cls._session[user_id] = True
def require_auth(fn):
def wrapper(user_id, *args, **kwargs):
if not AuthManager.is_logged_in(user_id):
print(f"用户 {user_id} 未登录，请先登录")
# 模拟登录流程
AuthManager.login(user_id)
print(f"用户 {user_id} 登录成功")
return fn(user_id, *args, **kwargs)
return wrapper
@require_auth
def view_dashboard(user_id):
print(f"用户 {user_id} 查看仪表盘数据")
@require_auth
def export_report(user_id, report_type):
print(f"用户 {user_id} 导出 {report_type} 报表")
# 模拟调用
view_dashboard("user_001")
export_report("user_001", "月度销售")
export_report("user_002", "季度财务") # 新用户需重新登录
