Python高级特性：详解装饰器与生成器

来源：投稿  作者：sunny 

编辑：学姐

在Python中，装饰器（Decorator）和生成器（Generator）如同两件神秘法宝，让开发者能以更优雅的方式解决复杂问题，前者赋予代码超越流程控制的能力，后者则开辟了内存优化的新维度。

一、装饰器

1.1 从一个实际需求说起

假设我们正在开发一个Web框架，需要为所有路由函数添加身份验证和日志记录功能。传统做法需要修改每个函数：

def user_profile(request):
    # 身份验证
    if not check_auth(request):
        


    
return redirect('/login')
    
    # 日志记录
    log_access(request.user)
    
    # 业务逻辑
    return render_profile(request.user)

当需要修改认证逻辑时，必须逐个修改所有函数。这显然违反了DRY（Don't Repeat Yourself）原则。

1.2 装饰器的诞生

装饰器本质上是一个高阶函数，它接收一个函数作为参数，返回一个新的增强函数。

基础版装饰器：

def log_decorator(func):
    def wrapper(*args, **kwargs):
        print(f"调用函数 {func.__name__}")
        return func(*args, **kwargs)
    return wrapper

@log_decorator
def add(a, b):
    return a + b

print(add(2,3))
# 输出：
# 调用函数 add
# 5

代码解析：

1. @log_decorator 是语法糖，等价于 add = log_decorator(add)
2. wrapper 函数保留了原函数的参数（*args, **kwargs）
3. 通过闭包特性保留了原函数的引用

1.3 进阶技巧：带参数的装饰器

当需要配置装饰器行为时（如设置日志级别），可以使用三层嵌套：

def repeat(num_times):
    def decorator(func):
        def wrapper(*args, **kwargs):
            for _ in


    
 range(num_times):
                result = func(*args, **kwargs)
            return result
        return wrapper
    return decorator

@repeat(num_times=3)
def greet(name):
    print(f"Hello {name}")

greet("Alice")
# 输出：
# Hello Alice
# Hello Alice
# Hello Alice

关键点：

• 最外层函数接收装饰器参数
• 中间层函数接收被装饰函数
• 最内层函数实现增强逻辑

1.4 类装饰器的奥秘

装饰器不仅限于函数，还可以用类实现：

class CountCalls:
    def __init__(self, func):
        self.func = func
        self.calls = 0

    def __call__(self, *args, **kwargs):
        self.calls += 1
        print(f"调用次数：{self.calls}")
        return self.func(*args, **kwargs)

@CountCalls
def 


    
multiply(a, b):
    return a * b

print(multiply(2,3))  # 调用次数：1 → 6
print(multiply(4,5))  # 调用次数：2 → 20

魔法方法解析：

• __call__ 方法使类实例成为可调用对象
• 通过 self.func 保留原函数引用
• 可以维护状态（如调用次数）

1.5 装饰器的实际应用场景

1. 性能分析：记录函数执行时间

    
import time

def timer(func):
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        duration = time.perf_counter() - start
        print(f"{func.__name__} 执行耗时: {duration:.4f}秒")
        return result
    return wrapper

@timer
def complex_calculation():
    time.sleep(1.5)

complex_calculation()
# 输出：complex_calculation 执行耗时: 1.5001秒

1. 缓存优化：记忆化（Memoization）

def memoize(func):
    cache = {}
    def wrapper(*args):
        if args in cache:
            return cache[args]
        result = func(*args)
        cache[args] = result
        return result
    return wrapper

@memoize
def fibonacci(n):
    return n if n 2else fibonacci(n-1) + fibonacci(n-2)

print(fibonacci(30))  # 首次计算耗时，后续调用瞬间返回

1. 权限控制：基于角色的访问控制

    
def admin_required(func):
    def wrapper(user, *args, **kwargs):
        if user.role != 'admin':
            raise PermissionError("需要管理员权限")
        return func(user, *args, **kwargs)
    return wrapper

class User:
    def __init__(self, role):
        self.role = role

@admin_required
def delete_user(user, user_id):
    print(f"删除用户 {user_id}")

admin = User('admin')
normal_user = User('user')

delete_user(admin, 123)     # 正常执行
delete_user(normal_user, 456)  # 抛出 PermissionError

二、生成器

2.1 从迭代器说起

生成器是迭代器的进化版，先回顾迭代器协议：

    
class Counter:
    def __init__(self, max_num):
        self.max_num = max_num
        self.current = 0

    def __iter__(self):
        return self

    def __next__(self):
        if self.current < self.max_num:
            self.current += 1
            return self.current
        else:
            raise StopIteration

for num in Counter(5):
    print(num)
# 输出：1 2 3 4 5

2.2 生成器的诞生：yield关键字

生成器使用 yield 语句替代复杂的迭代器实现：

def counter(max_num):
    current = 0
    while current < max_num:
        current += 1
        yield current
 
for num in counter(5):
    print(num)
# 输出：1 2 3 4 5

关键特性：

• 惰性求值：每次产生一个值，不保存全部结果
• 状态保存：yield 会保留函数执行上下文
• 内存高效：适合处理大数据流

2.3 生成器表达式

类似列表推导式，但使用圆括号：

gen = (x**2 for x in range(10))
print(list(gen))  # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

2.4 协程与生成器的进阶应用

生成器可以作为简单的协程使用，通过 send() 方法发送值：

def coroutine():
    while True:
        received = yield
        print(f"收到消息: {received}")
 
c = coroutine()
next(c)  # 启动生成器，执行到第一个yield
c.send("Hello")  # 收到消息: Hello
c.send("World")  # 收到消息: World

2.5 生成器的实际应用场景

1. 大数据处理：逐行读取文件

def read_large_file(file_path):
    with open(file_path, 'r') as f:
        for line in f:
            yield line.strip()
 
for line in read_large_file('big_data.txt'):


    

    process(line)  # 逐行处理，内存友好

1. 数学序列生成：斐波那契数列

def fibonacci():
    a, b = 0, 1
    while True:
        yield a
        a, b = b, a + b
 
fib = fibonacci()
print([next(fib) for _ in range(10)])  # [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]

1. 管道处理：数据流水线

def multiply_by(n):
    whileTrue:
        num = yield
        yield num * n

def add(x):
    whileTrue:
        num = yield
        yield num + x

pipeline = multiply_by(2) | add(5)  # 需要Python 3.5+的生成器管道支持

for num in pipeline:
    print(num)
# 输入序列：1,2,3 → 输出：7,9,11

三、装饰器与生成器的协同作战

3.1 生成器装饰器模式

为生成器添加预处理/后处理逻辑：

def validate_output(generator):
    for value in generator:
        if value 0:
            raise ValueError("生成器产生负数")
        yield value

@validate_output
def temperature_sensor():
    yield25.5
    yield-3.2# 会触发异常
    yield28.1

for temp in temperature_sensor():
    print(f"当前温度: {temp}℃")
# 输出：
# 当前温度: 25.5℃
# 抛出 ValueError: 生成器产生负数

3.2 异步任务调度

结合生成器和装饰器实现简单任务队列：

import time
from collections import deque

class TaskScheduler:
    def __init__(self):
        self.tasks = deque()
    
    def add_task(self, task):
        self.tasks.append(task)
    
    def run(self):
        while self.tasks:
            task = self.tasks.popleft()
            try:
                next(task)
                self.tasks.append(task)
            except StopIteration:
                print(f"任务完成: {task.name}")

def task(func):
    def wrapper(*args, **kwargs):
        task_gen = func(*args, **kwargs)
        task_gen.name = func.__name__
        next(task_gen)  # 预启动到第一个yield
        return task_gen
    return wrapper

scheduler = TaskScheduler()

@task
def download_file():
    print("开始下载...")
    time.sleep(2)
    print("下载完成")
    yield

@task
def process_data():
    print("处理数据中...")
    time.sleep(1)
    print("处理完成")
    yield

scheduler.add_task(download_file())
scheduler.add_task(process_data())
scheduler.run()
# 输出：
# 开始下载...
# 处理数据中...
# 下载完成
# 处理完成
# 任务完成: download_file
# 任务完成: process_data

四、深入理解底层机制

4.1 装饰器的函数签名问题

装饰器可能破坏原函数的元数据，使用 functools.wraps 修复：

from functools import wraps

def my_decorator(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        print("装饰器生效")
        return func(*args, **kwargs)
    return wrapper

@my_decorator
def example():
    """示例函数"""
    pass

print(example.__name__)  # 输出 example（而非 wrapper）
print(example.__doc__)   # 输出 "示例函数"

4.2 生成器的执行流程

通过反汇编观察生成器状态：

import dis

def simple_generator():
    yield1
    yield2

dis.dis(simple_generator)
# 输出包含：
# LOAD_CONST               1 (1)
# YIELD_VALUE
# POP_TOP
# LOAD_CONST               2 (2)
# YIELD_VALUE
# POP_TOP
# LOAD_CONST               0 (None)
# RETURN_VALUE

4.3 生成器的内存优势验证

对比列表和生成器的内存使用：

import sys
 
# 列表方式
list_range = list(range(10**6))
print(f"列表内存占用: {sys.getsizeof(list_range)} bytes")  


    
# 约8MB
 
# 生成器方式
gen_range = (x for x in range(10**6))
print(f"生成器内存占用: {sys.getsizeof(gen_range)} bytes")  # 约128B

五、如何正确使用装饰器、生成器

5.1 装饰器使用原则

1. 保持装饰器简单：避免在装饰器中实现复杂业务逻辑
2. 明确装饰器作用：通过命名清晰表达功能（如 @login_required）
3. 使用 wraps 保持元数据
4. 避免多层嵌套装饰器的执行顺序问题

5.2 生成器使用场景选择

1. 优先选择生成器表达式而非列表推导式处理大数据
2. 需要状态保持时使用生成器函数
3. 复杂数据流处理使用生成器管道
4. 避免在生成器中执行阻塞IO操作（考虑使用异步框架）

5.3 调试技巧

1. 使用 inspect 模块检查装饰器：

import inspect
 
print(inspect.getsource(example))  # 查看被装饰函数源码

1. 生成器调试：

gen = counter(3)
print(next(gen))  # 1
print(gen.gi_frame.f_lasti)  # 查看当前指令位置

- END -

推荐课程