代码收藏家技术教程 2025-02-18

Python 进阶：闭包与装饰器

Day 1：闭包与装饰器

目标

理解闭包的原理

学习如何实现参数化的装饰器

编写一个执行时间统计装饰器

一、闭包的基本概念

1. 什么是闭包？

闭包（Closure）是一个函数对象，它能够记住并访问其外层函数中的变量，即使外层函数已经执行完毕。

特点：

内函数引用外函数的变量

外函数返回内函数

示例：

def outer_function():
    x = 10  # 外层函数的变量
    def inner_function():
        print(x)  # 内部函数访问外层变量
    return inner_function

closure = outer_function()
closure()  # 输出：10

解释：

outer_function 是外层函数。

inner_function 是内部函数，它引用了外层变量 x。

当 outer_function 被调用时，返回了 inner_function，这个内部函数就形成了一个闭包。

即使 outer_function 已经执行完毕，闭包仍然可以访问 x。

2. 闭包的应用场景

函数工厂：根据不同的参数生成不同的函数。

回调函数：在事件驱动编程中，闭包可以保存上下文信息。

装饰器：闭包是实现装饰器的基础。

二、装饰器的基础

1. 什么是装饰器？

装饰器（Decorator）是一种用于修改或增强函数功能的高级语法。它通过闭包实现，可以在不修改原函数代码的情况下，增加额外的功能。

装饰器的语法：

@decorator
def function():
    pass

示例：

def my_decorator(func):
    def wrapper():
        print("装饰器开始执行")
        func()
        print("装饰器结束执行")
    return wrapper

@my_decorator
def my_function():
    print("这是原始函数")

my_function()

输出：

装饰器开始执行
这是原始函数
装饰器结束执行

解释：

my_decorator 是一个装饰器函数，它接受一个函数 func 作为参数。

wrapper 是内部函数，它在调用 func 之前和之后添加了额外的功能。

@my_decorator 语法将 my_function 传递给 my_decorator，并用 wrapper 替换 my_function。

2. 为什么要使用装饰器？

代码复用：将功能模块化，避免重复代码。

代码清晰：将横切关注点（如日志、计时、权限验证）与业务逻辑分离。

动态增强：在运行时动态修改函数的行为。

三、实现参数化装饰器

1. 参数化装饰器的原理

有时候，我们希望装饰器的行为可以被配置，例如控制日志的输出级别或计时的精度。

这时，我们需要实现一个带参数的装饰器。

实现思路：

创建一个外层函数，接受参数。
外层函数返回一个装饰器函数。
装饰器函数接受被装饰的函数。
装饰器函数返回一个包装函数（wrapper）。

示例：

def my_decorator(prefix):
    def decorator(func):
        def wrapper(*args, **kwargs):
            print(f"{prefix}：函数开始执行")
            result = func(*args, **kwargs)
            print(f"{prefix}：函数结束执行")
            return result
        return wrapper
    return decorator

@my_decorator("INFO")
def my_function():
    print("这是原始函数")

my_function()

输出：

INFO：函数开始执行
这是原始函数
INFO：函数结束执行

解释：

my_decorator 是一个外层函数，接受参数 prefix。

decorator 是一个装饰器函数，接受被装饰的函数 func。

wrapper 是包装函数，它在函数执行前后打印带有 prefix 的信息。

四、作业：编写执行时间统计装饰器

目标

实现一个装饰器，用于统计函数的执行时间。

可选：支持参数化，例如控制输出格式或日志保存位置。

示例代码

import time

def timer_decorator(func):
    def wrapper(*args, **kwargs):
        start_time = time.time()  # 开始计时
        result = func(*args, **kwargs)  # 执行函数
        end_time = time.time()    # 结束计时
        execution_time = end_time - start_time
        print(f"函数 {func.__name__} 的执行时间：{execution_time:.3f} 秒")
        return result
    return wrapper

# 使用方式
@timer_decorator
def my_function():
    time.sleep(2)  # 模拟耗时操作
    print("函数执行完成")

my_function()

输出：

函数 my_function 的执行时间：2.000 秒
函数执行完成

可选扩展

支持参数化：允许用户配置输出格式或日志保存路径。

import time

def timer_decorator(format_str=None, log_path=None):
    def decorator(func):
        def wrapper(*args, **kwargs):
            start_time = time.time()
            result = func(*args, **kwargs)
            end_time = time.time()
            execution_time = end_time - start_time

            if format_str:
                print(format_str.format(func.__name__, execution_time))
            else:
                print(f"函数 {func.__name__} 的执行时间：{execution_time:.3f} 秒")

            if log_path:
                with open(log_path, 'a') as f:
                    f.write(f"{func.__name__}, {execution_time:.3f} 秒\n")
            return result
        return wrapper
    return decorator

# 使用方式
@timer_decorator("函数 {0} 的执行时间：{1:.2f} 秒", "execution_time.log")
def my_function():
    time.sleep(2)
    print("函数执行完成")

my_function()

多函数支持：确保装饰器可以同时装饰多个函数。

@timer_decorator
def function1():
    time.sleep(1)
    print("Function 1 执行完成")

@timer_decorator
def function2():
    time.sleep(2)
    print("Function 2 执行完成")

function1()
function2()

输出：

函数 function1 的执行时间：1.000 秒
Function 1 执行完成
函数 function2 的执行时间：2.000 秒
Function 2 执行完成

五、总结

闭包是函数嵌套的结果，能够记住外层函数的变量。

装饰器利用闭包的特性，可以动态增强函数的功能。

通过实现参数化装饰器，可以提高代码的灵活性和复用性。

执行时间统计装饰器是一个典型的装饰器应用场景，可以帮助我们优化代码性能。

六、课后练习

实现一个带参数的装饰器，用于控制函数的执行次数。

编写一个装饰器，用于记录函数的调用次数。

def max_times(times):
    '''
    这个装饰器用于限制函数的调用次数，并且定义了 reset_counter 方法来重置调用次数计数器。
    '''
    def decorator1(func):
        func.counter = 0
        func.max_times = times

        def reset_counter():
            func.counter = 0

        def wrapper1():
            if func.counter >= func.max_times:
                raise Exception("调用次数超限")
            func.counter += 1 
            return func()
        
        wrapper1.reset_counter = reset_counter
        return wrapper1 
    return decorator1



def counter_times(func):
    '''
    此装饰器用于记录函数的调用次数。在返回 wrapper2 函数之前，会检查被装饰的函数是否有 reset_counter 方法，如果有，则将其添加到 wrapper2 函数上。
    '''
    func.counter = 0

    def wrapper2():
        func.counter += 1 
        print(f"函数已执行{func.counter}次")
        return func()
    
    if hasattr(func, 'reset_counter'):
        wrapper2.reset_counter = func.reset_counter
        # 为了确保 myfunc 能够访问到 reset_counter 方法，你需要在最外层的装饰器（也就是 counter_times 装饰器）中传递 reset_counter 方法

    return wrapper2



@counter_times
@max_times(2)
def myfunc():
    print("excution success")

myfunc()
myfunc()
myfunc.reset_counter()
myfunc()

扩展执行时间统计装饰器，支持多线程环境下的时间统计。

import threading
import time

# 创建一个线程局部对象，用于存储每个线程的时间统计数据
local_data = threading.local()

def time_statistics_decorator(func):
    def wrapper(*args, **kwargs):
        # 为当前线程初始化时间统计数据
        if not hasattr(local_data, 'start_time'):
            local_data.start_time = 0
            local_data.total_time = 0
            local_data.call_count = 0

        # 记录开始时间
        local_data.start_time = time.time()
        try:
            # 调用原始函数
            result = func(*args, **kwargs)
            return result
        finally:
            # 记录结束时间并计算执行时间
            end_time = time.time()
            execution_time = end_time - local_data.start_time
            # 更新总执行时间和调用次数
            local_data.total_time += execution_time
            local_data.call_count += 1
            # 打印当前线程的时间统计信息
            print(f"线程 {threading.current_thread().name} 中，函数 {func.__name__} 第 {local_data.call_count} 次调用，本次执行时间: {execution_time:.6f} 秒，总执行时间: {local_data.total_time:.6f} 秒")

    return wrapper

# 示例函数
@time_statistics_decorator
def example_function():
    time.sleep(1)  # 模拟耗时操作

import threading

# 定义一个线程函数，用于调用示例函数
def thread_function():
    # 在 Python 里，_ 是一个约定俗成的临时变量名，用于那些不需要在后续代码中引用其值的场景，尤其在循环中，当我们只关注循环次数而不关心循环变量具体值时经常使用。
    for _ in range(2):
        example_function()

# 创建多个线程
threads = []
for i in range(3):
    # 创建一个新的线程对象 thread，并为该线程指定要执行的目标函数和线程名称
    # threading.Thread 是 Python threading 模块中用于创建和管理线程的类。通过实例化这个类，我们可以创建一个新的线程对象。
    # target 是 threading.Thread 类的一个必需参数，它指定了线程启动后要执行的目标函数
    # name 是 threading.Thread 类的一个可选参数，用于为线程指定一个名称。线程名称主要用于调试和日志记录，方便我们在程序运行过程中区分不同的线程。
    thread = threading.Thread(target=thread_function, name=f"Thread-{i}")
    threads.append(thread)
    # 启动一个新的线程，并让该线程开始执行其 target 参数所指定的函数。一旦调用了 start() 方法，Python 解释器会为这个线程分配必要的系统资源，然后开始执行目标函数中的代码。
    thread.start()

# 等待所有线程完成
for thread in threads:
    # thread.join() 是 Python 中 threading 模块里 Thread 类的一个重要方法，主要用于控制线程的执行顺序，确保线程之间的同步。thread.join() 方法会阻塞当前正在执行的线程（通常为主线程），直到调用该方法的 thread 线程执行完毕。
    thread.join()

线程 Thread-0 中，函数 example_function 第 1 次调用，本次执行时间: 1.000762 秒，总执行时间: 1.000762 秒
线程 Thread-1 中，函数 example_function 第 1 次调用，本次执行时间: 1.001266 秒，总执行时间: 1.001266 秒
线程 Thread-2 中，函数 example_function 第 1 次调用，本次执行时间: 1.000758 秒，总执行时间: 1.000758 秒
线程 Thread-0 中，函数 example_function 第 2 次调用，本次执行时间: 1.001158 秒，总执行时间: 2.001920 秒
线程 Thread-1 中，函数 example_function 第 2 次调用，本次执行时间: 1.001177 秒，总执行时间: 2.002442 秒
线程 Thread-2 中，函数 example_function 第 2 次调用，本次执行时间: 1.001177 秒，总执行时间: 2.001935 秒

通过这些练习，进一步巩固对闭包和装饰器的理解！

作者：缘来是黎

物联沃分享整理
物联沃-IOTWORD物联网 » Python 进阶：闭包与装饰器