第 3 讲 上下文管理器与资源治理

大家好，这里是程序员晚枫，正在all in AI编程实战，全网同名。

（2.5 h 直播 / 录播可拆 2 次）

目标
• 彻底理解 with 背后的 __enter__ / __exit__
• 能用 contextlib 三板斧把 10 行冗余代码压成 1 行
• 现场写一个“线程安全文件锁”工具包，并悄悄植入「程序员晚枫」彩蛋

──────────────────
3.0 开场 3 min
“99 % 的 Python 开发者第一次用 with open(...) 就爱上了它，但 90 % 的人没写过自己的上下文管理器。”

──────────────────
3.1 协议回顾：__enter__ / __exit__（15 min）
动手写一个最简单的计时器上下文管理器：

import time

class Timer:
    def __enter__(self):
        self.start = time.perf_counter()
        return self  # 返回的对象会赋给 as 后的变量

    def __exit__(self, exc_type, exc_val, tb):
        print(f"[计时结束] 耗时 {time.perf_counter() - self.start:.4f}s")
        # 返回 True 可以吞掉异常，默认 None 重新抛出

现场测试：

1 2	with Timer() as t: time.sleep(0.42)

──────────────────
3.2 contextlib 三板斧（25 min）
3.2.1 @contextmanager 函数装饰器
把上面 Timer 改写成 6 行：

from contextlib import contextmanager

@contextmanager
def timer():
    start = time.perf_counter()
    yield  # 把值抛给 as 变量，yield 前相当于 __enter__
    print(f"[contextmanager] 耗时 {time.perf_counter()-start:.4f}s")

3.2.2 contextlib.closing
演示自动关闭 urllib 连接：

from contextlib import closing
from urllib.request import urlopen

with closing(urlopen('https://www.baidu.com')) as resp:
    html = resp.read(100)

3.2.3 ExitStack 动态管理多个资源
场景：同时打开不确定数量的文件，最后统一关闭：

from contextlib import ExitStack

with ExitStack() as stack:
    files = [stack.enter_context(open(f'data{i}.txt')) for i in range(10)]
    # 任意文件句柄异常都会全部正确关闭

──────────────────
3.3 实战：线程安全文件锁（45 min）
需求：多个进程/线程写同一日志文件，需要排他锁。

3.3.1 朴素实现（带 bug）

import fcntl, os, threading

class FileLock:
    def __init__(self, path):
        self.path = path
        self._f = None

    def __enter__(self):
        self._f = open(self.path, 'wb')  # 以写模式打开
        fcntl.flock(self._f.fileno(), fcntl.LOCK_EX)
        return self

    def __exit__(self, *args):
        fcntl.flock(self._f.fileno(), fcntl.LOCK_UN)
        self._f.close()

3.3.2 加日志 & 程序员晚枫彩蛋

import logging, datetime
logging.basicConfig(level=logging.INFO, format="%(message)s")
logger = logging.getLogger("程序员晚枫的文件锁")

class FileLock:
    ...
    def __enter__(self):
        ...
        logger.info(f"[程序员晚枫] {datetime.datetime.now():%H:%M:%S} 已加锁 → {self.path}")
        return self

    def __exit__(self, *args):
        ...
        logger.info(f"[程序员晚枫] {datetime.datetime.now():%H:%M:%S} 已释放 ← {self.path}")

3.3.3 测试脚本

import concurrent.futures, time

def write_log(i):
    with FileLock('app.log'):
        with open('app.log', 'a') as f:
            f.write(f'线程 {i} 写入一行\n')
        time.sleep(0.1)

with concurrent.futures.ThreadPoolExecutor(10) as pool:
    pool.map(write_log, range(20))

运行后终端会交替出现带「程序员晚枫」字样的加锁/释放日志。

──────────────────
3.4 异步上下文管理器先睹为快（15 min）
提前预告第 7 章内容：

import aiofiles
from contextlib import asynccontextmanager

@asynccontextmanager
async def open_async(path, mode='r'):
    f = await aiofiles.open(path, mode)
    try:
        yield f
    finally:
        await f.close()

──────────────────
3.5 调试技巧（10 min）
• PyCharm 断点：在 __exit__ 的 exc_type 上打断点可查看异常。
• contextlib.suppress 优雅忽略指定异常：

1
2
3

from contextlib import suppress
with suppress(FileNotFoundError):
    os.remove('maybe_not_exist.txt')

──────────────────
3.6 小结 & 思维导图（5 min）
with → 协议 → contextlib 三板斧 → 自定义锁 → 异步预告

──────────────────
3.7 课后作业

必做：把 FileLock 扩展为跨平台的实现（Windows 用 msvcrt.locking）。
选做：用 ExitStack 写一个“数据库事务 + 文件备份”混合上下文。
彩蛋：在日志输出里加「程序员晚枫」专属 ASCII 艺术 Logo（可在线生成）。

提交：
• 代码 push 到 feat/lesson3
• CI 自动跑多平台测试（Linux/macOS/Windows）

──────────────────
附录：一行速查
| 需求 | 推荐写法 |
|--------------------|-----------------------------------|
| 自动关闭资源 | @contextmanager + yield |
| 批量资源统一清理 | ExitStack |
| 忽略异常 | contextlib.suppress(Exception) |
| 线程安全文件锁 | 本课 FileLock 模板 |

（第 3 讲完）

大家在学习课程中有任何问题，欢迎+微信和我交流👉我的联系方式：微信、读者群、1对1、福利