在实际编程中，有时会涉及到资源的申请和释放，如果资源申请后没有得到及时和正确的释放，会造成内存泄漏问题。

按照传统的编程方式，一般是先申请资源，使用完之后释放资源。

Python提供了上下文管理器用于自动释放资源，而且可以简化代码。

上下文管理器的一般用法是：

with <context_manager> as <variable>:
    ...

在Python中，open()函数返回的是一个上下文管理器，threading.Lock()返回的也是一个上下文管理器，这两种上下文管理器是系统内置的。

如果要自定义上下文管理器，有两种实现方式。一种是自定义类，重写__enter__()和__exit__()，该类的实例是上下文管理器，另一种是借助@contextmanager修饰一个函数，该函数的返回值是一个上下文管理器。

接下来介绍以下几个和上下文管理器相关的主题：

• 文件操作
• 线程锁
• 使用类自定义上下文管理器
• 使用@contextmanager定义上下文管理器

文件操作

假设需要对一个文件写入数据。

传统的实现方法示例如下：

file = open('tmp.txt', 'w')
file.write('hello python')
file.close()

使用上下文管理器的示例如下：

with open('tmp.txt', 'w') as f:
    f.write('hello python')

可以采用上述写法，是因为open函数返回的就是一个上下文管理器。

线程锁

假设线程1对共享变量num进行加1操作，线程2对num进行减1操作，为了保证数据的一致性，可以通过加锁实现。

传统的实现方法示例如下：

import threading
import time

num = 2
lock = threading.Lock()


def add_one():
    global num

    # 对num进行六次+1操作
    for _ in range(1, 7):
        lock.acquire()
        try:
            num += 1
            print(f"after add_one() num is {num}")
        finally:
            lock.release()
        time.sleep(0.3)  # 休眠0.3秒


def sub_one():
    global num

    # 对num进行六次-1操作
    for _ in range(1, 7):
        lock.acquire()
        try:
            num -= 1
            print(f"after sub_one() num is {num}")
        finally:
            lock.release()
        time.sleep(0.3)  # 休眠0.3秒


# 创建线程
t_add_one = threading.Thread(target=add_one)
t_sub_one = threading.Thread(target=sub_one)

# 启动线程
t_add_one.start()
t_sub_one.start()

# 等待线程完成
t_add_one.join()
t_sub_one.join()

# 输出结果示例
# after add_one() num is 3
# after sub_one() num is 2
# after add_one() num is 3
# after sub_one() num is 2
# after add_one() num is 3
# after sub_one() num is 2
# after sub_one() num is 1
# after add_one() num is 2
# after sub_one() num is 1
# after add_one() num is 2
# after sub_one() num is 1
# after add_one() num is 2

num的初始值为2，经过不同顺序的六次加1和六次减1后，结果仍然是2。

使用上下文管理器的示例如下：

import threading
import time

num = 2
lock = threading.Lock()


def add_one():
    global num

    # 对num进行六次+1操作
    for _ in range(1, 7):
        with lock:
            num += 1
            print(f"after add_one() num is {num}")
        time.sleep(0.3)  # 休眠0.3秒


def sub_one():
    global num

    # 对num进行六次-1操作
    for _ in range(1, 7):
        with lock:
            num -= 1
            print(f"after sub_one() num is {num}")
        time.sleep(0.3)  # 休眠0.3秒


# 创建线程
t_add_one = threading.Thread(target=add_one)
t_sub_one = threading.Thread(target=sub_one)

# 启动线程
t_add_one.start()
t_sub_one.start()

# 等待线程完成
t_add_one.join()
t_sub_one.join()

# 输出结果示例
# after add_one() num is 3
# after sub_one() num is 2
# after sub_one() num is 1
# after add_one() num is 2
# after sub_one() num is 1
# after add_one() num is 2
# after add_one() num is 3
# after sub_one() num is 2
# after sub_one() num is 1
# after add_one() num is 2
# after add_one() num is 3
# after sub_one() num is 2

可以写作with lock，是因为threading.Lock()的返回值也是一个上下文管理器。

使用类自定义上下文管理器

假设有两个或更多的资源，可以通过自定义类作为上下文管理器对资源进行统一管理。

类的__enter__()方法主要实现资源的创建，__exit__()方法主要实现资源的释放。

示例如下：

class MyContextManager:
    def __enter__(self):
        print("Entering MyContextManager")
        print('open resources')
        return self  # 返回的对象可以通过as赋值

    def do_something(self):
        print('use resources')

    def __exit__(self, exc_type, exc_val, exc_tb):
        print("Exiting MyContextManager")
        print('close resources')
        if exc_type:
            print(f"Exception occurred: {exc_val}")
            return True  # 异常被抑制，不向外传播
        return False  # 异常正常传播（默认行为）


with MyContextManager() as cm:
    cm.do_something()


# 输出结果如下:
# Entering MyContextManager
# open resources
# use resources
# Exiting MyContextManager
# close resources

使用@contextmanager定义上下文管理器

也可以使用@contextmanager定义上下文管理器，实现代码更简洁。

示例如下：

from contextlib import contextmanager


class ResourceManager:
    def __init__(self):
        print('open resources')

    def do_something(self):
        print('use resources')

    def close(self):
        print('close resources')


@contextmanager
def my_context_manager() -> ResourceManager:
    print("Entering my_context_manager")
    resource_manager = ResourceManager()
    try:
        # yield 之前的代码相当于 __enter__()，yield 之后的代码相当于 __exit__()
        yield resource_manager  # 这里可以返回值，相当于 `__enter__` 的返回值
        print("Exiting my_context_manager")
        resource_manager.close()
    except Exception as e:
        print(f"Exception occurred: {e}")


with my_context_manager() as cm:
    cm.do_something()

# 输出结果如下:
# Entering my_context_manager
# open resources
# use resources
# Exiting my_context_manager
# close resources