在 Python 编程中，资源管理是一个常见且重要的问题。无论是文件操作、网络连接还是数据库事务，都需要确保资源在使用后能够正确地释放或恢复到初始状态。Python 通过上下文管理器提供了一种优雅的方式来处理资源的获取与释放，使得代码更加简洁、安全。
什么是上下文管理协议
Python 的上下文管理协议是一组特殊方法的集合，它们允许对象与 with 语句配合使用，以确保在代码块执行前后正确地管理资源。这个协议分为同步上下文管理协议和异步同步上下文管理协议。协议主要是实现两种方法。
同步上下文管理协议
enter() 方法：当进入 with 语句块时，该方法被调用。它应该返回一个对象，通常是管理器对象本身，该对象在 with 块中使用。这个方法允许你执行一些设置工作，比如打开文件、获取锁或初始化资源。
exit(exc_type, exc_value, traceback) 方法：当退出 with 语句块时，无论是否发生异常，该方法都会被调用。它接收三个参数：
exc_type：如果 with 块中发生异常，此参数为异常类型；否则为 None。
exc_value：如果发生异常，此参数为异常实例；否则为 None。
traceback：如果发生异常，此参数为 traceback 对象；否则为 None。
__exit__方法允许你执行清理工作，比如关闭文件、释放锁或释放资源。如果__exit__方法返回 False 或没有返回值（这意味着返回了 None），异常（如果发生了的话）将被重新抛出；如果返回 True，则表明异常已经被处理，并且不会重新抛出。
异步上下文管理协议：
对于异步代码，Python 3.7+ 引入了异步上下文管理器，它使用以下两个方法：
aenter() 方法：异步上下文管理器的进入方法，类似于__enter__()，但它是一个异步方法，可以使用 await。
aexit(exc_type, exc_value, traceback) 方法：异步上下文管理器的退出方法，也是一个异步方法。它接收与同步版本相同的参数，并在退出 async with 语句块时被调用。
什么是上下文管理器？
上下文管理器是实现了上下文管理协议的对象。通过上下文管理器，能够实现精确控制资源创建和释放时机。它允许你执行一些设置和清理工作，而不需要显式地编写这些代码。Python 中的上下文管理器主要通过两个魔法方法实现：enter()和__exit__()。
使用上下文管理器
Python 中最常见的上下文管理器是文件操作。例如：
with open('example.txt', 'r') as file: content = file.read() # 对文件内容进行操作# 文件在这里自动关闭
复制代码
在这个例子中，open 函数返回一个文件对象，它实现了上下文管理器协议。使用 with 语句可以确保文件在使用后自动关闭。
创建自定义上下文管理器
除了使用内置的上下文管理器，你还可以创建自定义的上下文管理器。这可以通过定义一个类并实现__enter__()和__exit__()方法来完成。
使用类定义上下文管理器
class MyContextManager: def __enter__(self): print("Entering the context.") return self
def __exit__(self, exc_type, exc_value, traceback): print("Exiting the context.") # 处理异常或进行清理工作 if exc_type: print(f"An exception occurred: {exc_value}") return False # 重新抛出异常
with MyContextManager() as manager: print("Inside the context.") # 可以执行一些操作，如果发生异常，__exit__()会处理
复制代码
使用 contextlib 模块简化上下文管理器
对于简单的上下文管理器，Python 的 contextlib 模块提供了一个更简洁的写法。使用 @contextlib.contextmanager 装饰器，你可以将资源的获取和释放逻辑放在一个生成器函数中。
from contextlib import contextmanager
@contextmanagerdef my_context(): print("Resource acquisition") yield print("Resource release")
with my_context(): # 使用资源 pass
复制代码
创建异步上下文管理器
在 Python 中，异步上下文管理器（asynchronous context manager）是一种特殊类型的上下文管理器，它允许在异步环境中使用 async with 语句来管理资源的获取和释放。这种上下文管理器通过定义__aenter__()和__aexit__()两个异步方法（coroutine）来实现，这两个方法可以在进入和退出上下文时执行异步操作。
使用类定义上下文管理器
要创建一个异步上下文管理器，你需要定义一个类，并在该类中实现__aenter__()和__aexit__()方法。这两个方法必须使用 async def 进行定义，以便它们可以作为协程执行。例如：
class AsyncContextManager: async def __aenter__(self): # 进入上下文时执行的操作 await asyncio.sleep(1) # 模拟异步操作 print('Entering the context.') return self
async def __aexit__(self, exc_type, exc, tb): # 退出上下文时执行的操作 print('Exiting the context.') await asyncio.sleep(1) # 模拟异步操作
复制代码
在上面的代码中，aenter()方法在进入上下文时被调用，而__aexit__()方法则在退出上下文时被调用，无论是否发生异常。
import asynciofrom contextlib import asynccontextmanager
@asynccontextmanagerasync def async_lock(): print('Attempting to acquire lock.') # 模拟异步获取锁的过程 await asyncio.sleep(1) print('Lock acquired.') try: yield # 进入上下文，执行yield之后的代码块 finally: # 退出上下文，执行yield之前的代码块 print('Lock released.') # 模拟异步释放锁的过程 await asyncio.sleep(1)
# 使用异步上下文管理器async def main(): async with async_lock() as lock: print('Inside the context with lock:') # 这里是需要同步执行的代码块
# 运行异步主函数asyncio.run(main())
# 输出结果# Lock acquired.# Inside the context with lock:# Lock released.
复制代码
如何使用异步上下文管理器
使用异步上下文管理器非常简单，你只需要使用 async with 语句，如下所示：
async def main(): async with AsyncContextManager() as manager: print('Inside the context with manager:', manager)
# 运行异步主函数asyncio.run(main())
复制代码
在这个例子中，AsyncContextManager()实例被创建，并在 async with 语句中使用。当进入 async with 块时，会自动调用__aenter__()方法，并等待其完成。当退出这个块时，会自动调用__aexit__()方法，并等待其完成。
异步上下文管理器与同步上下文管理器的区别
异步上下文管理器与同步上下文管理器的主要区别在于它们使用的魔法方法不同。同步上下文管理器使用__enter__()和__exit__()方法，而异步上下文管理器使用__aenter__()和__aexit__()方法。此外，异步上下文管理器只能在异步函数中使用，并且必须与 async with 语句一起使用。
上下文管理器使用场景
文件操作：使用上下文管理器可以自动管理文件的打开和关闭，即使在读取或写入文件时发生异常也能确保文件被正确关闭。
数据库连接：数据库连接通常需要明确地关闭以释放资源，上下文管理器可以保证即使在查询过程中发生错误也能关闭连接。
网络连接：网络请求可能需要打开和关闭连接，使用上下文管理器可以自动处理这些操作。
线程和锁：在多线程编程中，使用上下文管理器可以自动获取和释放锁，避免死锁的发生。
模拟资源环境：在测试或某些特定操作中，可能需要模拟某些资源环境，上下文管理器可以在进入和退出时设置和清理环境。
资源池管理：对于从资源池中获取和释放资源的操作，上下文管理器可以确保资源被正确归还。
异常处理：在需要进行复杂异常处理的场景中，上下文管理器可以在退出时统一处理异常。
配置上下文：在需要临时改变配置并在操作完成后恢复原有配置的场景中，上下文管理器可以很方便地管理配置的变更。
常见上下文管理器的问题
注意点（坑点）
确保实现所有必要的方法：自定义上下文管理器时，需要实现__enter__()和__exit__()方法。对于异步上下文管理器，则需要实现__aenter__()和__aexit__()。
异常处理：在__exit__()或__aexit__()方法中，确保正确处理所有可能的异常。考虑是否需要捕获异常、记录日志或者重新抛出异常。
资源清理：上下文管理器的主要目的是管理资源的生命周期。确保在退出上下文时，所有资源（如文件句柄、网络连接、锁等）都被正确释放或重置。
避免副作用：enter()方法应该只负责初始化操作，避免产生副作用，比如修改外部状态或执行 I/O 操作。
使用 as 子句：当使用 with 语句时，使用 as 子句来赋予上下文管理器返回的对象一个名称，这样可以在块内引用该对象。
注意上下文管理器的嵌套：当上下文管理器嵌套使用时，确保内层上下文管理器的退出不会影响外层上下文管理器的状态。
线程安全：大多数同步上下文管理器不是线程安全的。如果你的上下文管理器涉及共享资源，确保在多线程环境中正确地同步访问。
避免阻塞操作：在异步上下文管理器中，避免在__aenter__()或__aexit__()中执行阻塞操作，这会破坏异步性能。
使用上下文管理器协议：如果你的类需要与上下文管理器一起使用，确保它遵循上下文管理器协议，即实现必要的特殊方法。
避免循环依赖：在使用上下文管理器时，避免创建循环依赖，这可能导致资源无法释放。
实际操作
（以下代码示例以同步的为例）
如果在使用异步上下文管理器时发生异常，aexit()方法仍然会被调用。你可以在__aexit__()方法中处理异常，或者根据需要返回 False 来重新抛出异常。
class MockDatabaseConnect: def __init__(self): self.is_connected = False
def connect(self): self.is_connected = True print("模拟：数据库已连接")
def close(self): self.is_connected = False print("模拟：数据库连接已关闭")
class DatabaseConnectionManager: def __init__(self, db:MockDatabaseConnect): self.db = db
def __enter__(self): self.db.connect() return self.db
def __exit__(self, exc_type, exc_val, exc_tb): self.db.close()
mock_db = MockDatabaseConnect()
# 使用上下文管理器with DatabaseConnectionManager(mock_db) as db: # 在 with 块中使用连接 print("使用连接")
# 输出结果# 模拟：数据库已连接# 使用连接# 模拟：数据库连接已关闭
复制代码
class MockDatabaseConnect: def __init__(self): self.is_connected = False
def connect(self): self.is_connected = True print("模拟：数据库已连接")
def close(self): self.is_connected = False print("模拟：数据库连接已关闭")
class DatabaseConnectionPool: def __init__(self): self.connections = []
def add_connection(self, connection): self.connections.append(connection)
def get_connection(self) -> MockDatabaseConnect: if not self.connections: raise Exception("资源池中没有可用的连接") return self.connections.pop(0)
def release_connection(self, connection): self.connections.append(connection)
class DatabaseConnectionContextManager: def __init__(self, pool:DatabaseConnectionPool): self.pool = pool
def __enter__(self): self.connection = self.pool.get_connection() self.connection.connect() print("模拟：数据库连接已获取") return self.connection
def __exit__(self, exc_type, exc_val, exc_tb): if self.connection.is_connected: self.pool.release_connection(self.connection) print("模拟：数据库连接已释放")
pool = DatabaseConnectionPool()
# 添加模拟连接到池中mock_conn1 = MockDatabaseConnect()pool.add_connection(mock_conn1)
mock_conn2 = MockDatabaseConnect()pool.add_connection(mock_conn2)
# 使用with语句从池中获取连接with DatabaseConnectionContextManager(pool) as conn: # 在with块中使用连接 print("使用连接:", conn)
# 退出with块后，连接会自动返回到池中print("池中的连接:", pool.connections)
# 模拟：数据库已连接# 模拟：数据库连接已获取# 使用连接: <__main__.MockDatabaseConnect object at 0x106811b80># 模拟：数据库连接已释放# 池中的连接: [<__main__.MockDatabaseConnect object at 0x106837950>, <__main__.MockDatabaseConnect object at 0x106811b80>]
复制代码
import threadingimport time
class ThreadSafeContextManager: def __init__(self): self.lock = threading.Lock()
def __enter__(self): self.lock.acquire() print(f"线程 {threading.current_thread().name} 获取锁") return self
def __exit__(self, exc_type, exc_value, traceback): self.lock.release() print(f"线程 {threading.current_thread().name} 释放锁")
def worker(context_manager): with context_manager as cm: print(f"线程 {threading.current_thread().name} 正在执行任务") time.sleep(1) # 模拟任务执行 print(f"线程 {threading.current_thread().name} 任务完成")
# 创建上下文管理器对象context_manager = ThreadSafeContextManager()
# 创建线程并传入上下文管理器thread1 = threading.Thread(target=worker, args=(context_manager,))thread2 = threading.Thread(target=worker, args=(context_manager,))
# 启动线程thread1.start()thread2.start()
# 等待线程结束thread1.join()thread2.join()
# 线程 Thread-7 (worker) 获取锁# 线程 Thread-7 (worker) 正在执行任务# 线程 Thread-7 (worker) 任务完成# 线程 Thread-7 (worker) 释放锁# 线程 Thread-6 (worker) 获取锁# 线程 Thread-6 (worker) 正在执行任务# 线程 Thread-6 (worker) 任务完成# 线程 Thread-6 (worker) 释放锁
复制代码
from contextlib import contextmanager
class File: def __init__(self, name): self.name = name
def open(self): print(f"Opening file {self.name}") return self
def close(self): print(f"Closing file {self.name}")
@contextmanagerdef open_file(name): f = File(name) f.open() try: yield f finally: f.close()
@contextmanagerdef open_database(): # 模拟数据库连接 db = "database" try: yield db finally: # 模拟数据库关闭 print("Database closed")
# 使用嵌套的上下文管理器with open_database() as db: print("Database opened") with open_file("example.txt") as f: print("File opened") print(f.name) print("File closed")print("Database closed")
# 输出结果# Database opened# Opening file example.txt# File opened# example.txt# Closing file example.txt# File closed# Database closed# Database closed
复制代码
上下文管理器与装饰器：描述上下文管理器与装饰器模式之间的相似之处和不同之处，并讨论它们各自的使用场景。
相似之处：
包装代码：两者都用于包装现有代码，以添加新的行为或功能。
可重用性：两者都支持代码的可重用性，可以在多个地方使用相同的包装代码。
不同之处：
目的：上下文管理器主要用于管理资源的生命周期，例如打开和关闭文件、数据库连接等。装饰器模式主要用于扩展函数或类的行为，而不改变它们的接口。
语法：上下文管理器使用 with 语句块，而装饰器模式使用 @ 符号。
作用域：上下文管理器的作用域是 with 语句块内的代码，而装饰器模式的作用域是整个函数或类。
控制流：上下文管理器可以控制进入和退出 with 语句块的流程，而装饰器模式不能直接控制函数或类的执行流程。
编写一个支持超时的上下文管理器：实现一个上下文管理器，它接受一个超时参数，并在指定的时间内自动退出上下文。
import timefrom contextlib import contextmanager
# 定义一个超时异常class TimeoutException(Exception): pass
# 上下文管理器@contextmanagerdef timeout(seconds): start_time = time.time() try: # 执行上下文代码 yield except TimeoutException: # 如果超时，抛出异常 raise finally: elapsed_time = time.time() - start_time if elapsed_time > seconds: print("raise timeout exception") raise TimeoutException(f"Timeout after {seconds} seconds")
# 使用上下文管理器try: with timeout(3): # 模拟一个长时间的操作 time.sleep(5)except TimeoutException as e: print("catch timeout exception "+str(e))
# 输出结果# raise timeout exception# catch timeout exception Timeout after 3 seconds
复制代码
上下文管理器与性能：讨论在哪些情况下使用上下文管理器可能会影响程序的性能，并解释原因。
每次进入和退出上下文管理器时，都需要进行资源的分配和释放操作。如果上下文管理器用于高频调用的小函数或方法中，这些开销可能会累积起来，对性能产生负面影响。
在__exit__方法中处理异常可能会增加额外的计算负担。如果异常处理逻辑复杂或者涉及到大量的错误检查，这可能会降低程序的执行效率。
如果在一个 with 块中嵌套了多个上下文管理器，每个管理器都需要单独的资源管理和异常处理逻辑，这可能会导致性能问题，尤其是在资源竞争激烈的情况下。
在多线程环境中，如果上下文管理器涉及到锁或其他同步机制，争用这些资源可能会导致线程阻塞或上下文切换，从而影响性能。
如果上下文管理器用于执行 I/O 操作（如文件读写、网络通信等），而这些操作被设计为同步执行，它们可能会阻塞当前线程，直到操作完成，这会影响程序的响应性和吞吐量。
上下文管理器可能会在 with 块的整个生命周期内持有资源，这可能导致内存占用增加，尤其是在长时间运行的 with 块或大量并发的上下文管理器中。
增加代码负责性：使用上下文管理器可能会使代码逻辑更加复杂，尤其是在涉及多个资源和多层嵌套时。这种复杂性可能会导致代码难以维护和优化。
上下文管理器与元类：讨论是否可以使用元类来自动地将一个类转换为上下文管理器，并讨论这样做的优缺点。
优点：
简洁：使用元类可以简洁地将类转换为上下文管理器，无需手动添加 enter 和 exit 方法。
一致性：元类可以确保所有类都遵循相同的上下文管理器接口。
缺点：
灵活性：使用元类可能会限制类的灵活性，因为所有类都必须遵循相同的上下文管理器接口。
可读性：元类可能会使代码的可读性降低，因为类的行为是通过元类隐式定义的，而不是通过显式定义的方法。
class ContextMeta(type): def __new__(cls, name, bases, dct): def __enter__(self): print("enter") return self
def __exit__(self, exc_type, exc_val, exc_tb): print("exit") pass
dct['__enter__'] = __enter__ dct['__exit__'] = __exit__ return super().__new__(cls, name, bases, dct)
class MyClass(metaclass=ContextMeta): pass
with MyClass() as obj: print(obj)
# 输出结果# enter# <__main__.MyClass object at 0x1061d6900># exit
复制代码
import contextlib
@contextlib.contextmanagerdef multi_resource_manager(*resources): try: # Open all resources opened_resources = [resource.open() for resource in resources] yield opened_resources finally: # Close all resources for resource in opened_resources: resource.close()
# 例子：同时打开两个文件class File: def __init__(self, name): self.name = name
def open(self): print(f"Opening file {self.name}") return self
def close(self): print(f"Closing file {self.name}")
file1 = File("file1.txt")file2 = File("file2.txt")
with multi_resource_manager(file1, file2) as (f1, f2): print("Doing something with files...")
复制代码
作者：goasleep
链接：https://juejin.cn/post/7403193330271617076
评论