🛡️ 上下文管理器

上下文管理器（context manager）是 Python 中管理"资源获取—资源释放"配对的机制。无论代码块是正常结束还是因异常退出，它都能保证清理代码被执行——避免传感器连接泄漏、校准文件未关闭、数据缓冲区未刷新等常见问题。with 语句是它的语法入口，而 contextlib 标准库提供了大量工具，便于写出更优雅、更健壮的资源管理代码。

📌 本节要点

• with 语句保证资源在退出时被释放，无论是否发生异常
• 类形式的上下文管理器实现 __enter__ 和 __exit__，__exit__ 返回 True 可抑制异常
• @contextlib.contextmanager 用生成器函数简化实现：yield 前是"进入"，yield 后是"退出"
• ExitStack 用于动态管理数量不定的上下文管理器，按 LIFO 顺序退出
• async with 用于异步代码，对应 __aenter__/__aexit__
• contextlib 实用工具：suppress、redirect_stdout、closing、chdir（3.11+）

上下文管理器快速体验

with 语句

最常见的用法是处理文件：

Python

# 不推荐：手动 open/close
f = open("flight_data.csv")
try:
    content = f.read()
finally:
    f.close()

# 推荐：用 with 自动管理
with open("flight_data.csv", encoding="utf-8") as f:
    content = f.read()
# 离开 with 块后，文件自动关闭，即使 read() 抛异常也会关闭

with 语句保证：

1. 进入时调用上下文管理器的 __enter__
2. 离开时（无论正常退出还是异常）调用 __exit__
3. 即便 with 块中有 return、break、continue 或抛异常，__exit__ 也会执行

多个上下文管理器

Python 3.1+ 支持在一个 with 语句中放多个上下文管理器，按从前往后的顺序进入、从后往前退出：

Python

with open("raw_telemetry.txt", encoding="utf-8") as fin, \
     open("processed_telemetry.txt", "w", encoding="utf-8") as fout:
    for line in fin:
        fout.write(line.upper())
# 两个文件都会被正确关闭

Python 3.10+ 更优雅的多行写法

Python 3.10 起允许用括号包裹多个上下文管理器，无需用反斜杠续行：

Python

with (
    open("raw_telemetry.txt", encoding="utf-8") as fin,
    open("processed_telemetry.txt", "w", encoding="utf-8") as fout,
):
    for line in fin:
        fout.write(line.upper())

这种写法更易读，也更便于添加或删除条目。

enter 与 exit

任何实现了 __enter__ 和 __exit__ 两个方法的对象都是上下文管理器。

• __enter__(self)：进入 with 块时调用，返回值通过 as 赋给变量
• __exit__(self, exc_type, exc_val, exc_tb)：离开 with 块时调用，三个参数描述了"是否发生异常"

最简实现

Python

class MyContext:
    def __enter__(self):
        print("进入上下文")
        return self          # 返回值通过 as 赋给变量

    def __exit__(self, exc_type, exc_val, exc_tb):
        print("退出上下文")
        # 三个参数说明是否发生异常：
        # - 没异常时：exc_type 是 None
        # - 有异常时：exc_type 是异常类，exc_val 是异常实例，exc_tb 是 traceback
        if exc_type is not None:
            print(f"  捕获到异常：{exc_type.__name__}: {exc_val}")
        return False         # False 表示不抑制异常，让异常继续传播


with MyContext() as ctx:
    print("执行 with 块")
# 输出：
# 进入上下文
# 执行 with 块
# 退出上下文

print("---")

with MyContext() as ctx:
    print("抛出异常")
    raise ValueError("出错了")
# 输出：
# 进入上下文
# 抛出异常
# 退出上下文
#   捕获到异常：ValueError: 出错了
# ValueError: 出错了   ← 异常继续传播

exit 的返回值

__exit__ 的返回值控制是否"吞掉"异常：

• 返回 False（或 None）：异常继续传播，与没处理一样
• 返回 True：异常被吞掉，with 块外不会感知到异常

Python

class SuppressErrors:
    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is not None:
            print(f"吞掉了异常：{exc_val}")
            return True      # ← 返回 True 抑制异常
        return False


with SuppressErrors():
    print("正常执行")
    raise RuntimeError("假装出错了")
print("with 块外继续执行")   # 这里会执行，因为异常被吞了
# 输出：
# 正常执行
# 吞掉了异常：假装出错了
# with 块外继续执行

谨慎吞异常

返回 True 吞异常是危险操作——它会掩盖所有异常（包括 KeyboardInterrupt 之外的）。除非有明确理由（如重试逻辑、静默忽略已知错误），否则 __exit__ 应返回 False，让异常照常传播。如果只想忽略特定异常，用 contextlib.suppress（后面会讲）更安全。

完整示例：飞行数据记录器

Python

import time
import numpy as np


class FlightDataRecorder:
    """飞行数据记录上下文管理器，记录飞行测试的时序数据"""

    def __init__(self, test_name: str):
        self.test_name = test_name
        self.elapsed: float = 0.0
        self.data_points: list[dict] = []

    def __enter__(self) -> "FlightDataRecorder":
        self.start = time.perf_counter()
        self.record("test_start", {"timestamp": self.start})
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.elapsed = time.perf_counter() - self.start
        self.record("test_end", {"elapsed": self.elapsed})
        print(f"[{self.test_name}] 记录完成，耗时 {self.elapsed:.6f}s，共 {len(self.data_points)} 个数据点")
        if exc_type is not None:
            print(f"  异常终止：{exc_type.__name__}: {exc_val}")
        return False          # 不吞异常

    def record(self, event_type: str, data: dict):
        self.data_points.append({"event": event_type, **data})

    def get_data_array(self) -> np.ndarray:
        """将记录的数据转换为 numpy 数组"""
        if not self.data_points:
            return np.array([])
        return np.array([dp.get("elapsed", 0) for dp in self.data_points])


with FlightDataRecorder("MAV-001 试飞") as recorder:
    recorder.record("altitude", {"value": 100.0, "unit": "m"})
    recorder.record("velocity", {"value": 25.5, "unit": "m/s"})
    recorder.record("attitude", {"roll": 0.1, "pitch": 2.3, "yaw": 45.0})
    print(f"已记录 {len(recorder.data_points)} 个数据点")
# 输出：
# 已记录 4 个数据点
# [MAV-001 试飞] 记录完成，耗时 0.000123s，共 5 个数据点

# 异常情况也会正确记录
try:
    with FlightDataRecorder("异常测试") as recorder:
        recorder.record("sensor_error", {"code": 500})
        raise ValueError("传感器故障")
except ValueError:
    print("异常已传播到外层")
# 输出：
# [异常测试] 记录完成，耗时 0.000089s，共 2 个数据点
#   异常终止：ValueError: 传感器故障
# 异常已传播到外层

完整示例：传感器校准

Python

class SensorCalibration:
    """传感器校准上下文管理器：进入时加载校准参数，退出时保存校准结果"""

    def __init__(self, sensor_id: str):
        self.sensor_id = sensor_id
        self.calibration_data: dict = {}

    def __enter__(self):
        print(f"加载传感器 {self.sensor_id} 的校准参数...")
        self.calibration_data = {"offset": 0.0, "scale": 1.0, "status": "calibrating"}
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is not None:
            self.calibration_data["status"] = "failed"
            print(f"校准失败：{exc_type.__name__}，已标记状态")
        else:
            self.calibration_data["status"] = "completed"
            print(f"传感器 {self.sensor_id} 校准完成，参数已保存")
        return False

    def update_offset(self, new_offset: float):
        self.calibration_data["offset"] = new_offset
        print(f"  更新偏移量：{new_offset}")

    def update_scale(self, new_scale: float):
        self.calibration_data["scale"] = new_scale
        print(f"  更新缩放因子：{new_scale}")


with SensorCalibration("IMU-01") as cal:
    cal.update_offset(0.05)
    cal.update_scale(1.02)
# 输出：
# 加载传感器 IMU-01 的校准参数...
#   更新偏移量：0.05
#   更新缩放因子：1.02
# 传感器 IMU-01 校准完成，参数已保存

contextlib.contextmanager

写一个完整的类来实现 __enter__/__exit__ 有时显得冗长。contextlib.contextmanager 装饰器允许用生成器函数更简洁地实现上下文管理器：

• yield 之前的代码相当于 __enter__
• yield 出的值相当于 __enter__ 的返回值
• yield 之后的代码相当于 __exit__，放在 finally 中保证异常时也执行

Python

from contextlib import contextmanager
import time


@contextmanager
def flight_timer(test_name: str = "飞行测试"):
    """飞行计时上下文管理器（生成器版本）"""
    start = time.perf_counter()
    try:
        yield                # with 块的代码在此处执行
    finally:
        elapsed = time.perf_counter() - start
        print(f"[{test_name}] 耗时 {elapsed:.6f}s")


with flight_timer("传感器采样"):
    total = sum(i * i for i in range(1_000_000))
    print(f"结果：{total}")
# 输出：
# 结果：333332833333500000
# [传感器采样] 耗时 0.123456s

yield 返回值给 as

yield 的值会通过 as 赋给变量：

Python

from contextlib import contextmanager


@contextmanager
def flight_data_buffer(buffer_id: str, capacity: int = 1000):
    """飞行数据缓冲区管理"""
    print(f"初始化缓冲区 {buffer_id}，容量：{capacity}")
    buffer = {"id": buffer_id, "capacity": capacity, "data": [], "flushed": False}
    try:
        yield buffer                # 把缓冲区对象交给 with 块
    finally:
        buffer["flushed"] = True
        print(f"缓冲区 {buffer_id} 已刷新，共 {len(buffer['data'])} 条数据")


with flight_data_buffer("BUF-01", capacity=500) as buf:
    print(f"使用缓冲区：{buf['id']}")
    buf["data"].append({"t": 0.0, "value": 1.0})
    print(f"缓冲区状态：{len(buf['data'])}/{buf['capacity']}，已刷新={buf['flushed']}")
# 输出：
# 初始化缓冲区 BUF-01，容量：500
# 使用缓冲区：BUF-01
# 缓冲区状态：1/500，已刷新=False
# 缓冲区 BUF-01 已刷新，共 1 条数据

处理异常：try-except-finally

如果想在生成器版本中"处理"异常（类似 __exit__ 返回 True），用 try-except 包住 yield：

Python

from contextlib import contextmanager


@contextmanager
def suppress_sensor_errors(*exceptions):
    """忽略指定的传感器异常"""
    try:
        yield
    except exceptions as e:
        print(f"忽略传感器异常：{e}")
        # 不 re-raise，相当于 __exit__ 返回 True


with suppress_sensor_errors(ValueError, TimeoutError):
    print("正常采样")
    raise ValueError("传感器超时")
print("with 块外继续执行")   # 异常被吞了，继续执行
# 输出：
# 正常采样
# 忽略传感器异常：传感器超时
# with 块外继续执行

contextmanager 的核心机制

@contextmanager 把生成器函数包装成上下文管理器。当 with 进入时，它执行到 yield 并返回 yield 的值；当 with 块抛出异常时，异常会通过 generator.throw() 抛回生成器内部——所以我们可以在 yield 外层包 try-except-finally 来捕获它。如果不捕获，异常会从生成器中再次抛出，传播到 with 块外。

contextmanager vs 类实现

比较项	@contextmanager	类实现
代码量	少，适合简单场景	多，适合复杂状态
跨调用共享状态	不便（每次创建新生成器）	方便（实例属性）
异常处理	try-except-finally	__exit__ 返回值
嵌套使用	容易组合	需手动管理

经验法则：简单的"获取-释放"用 @contextmanager，需要复杂状态或多次调用的用类。

ExitStack

contextlib.ExitStack 是一个动态管理多个上下文管理器的工具。它解决了几个问题：

1. 数量不固定的上下文管理器：运行时才知道要打开多少个传感器连接
2. 条件性进入上下文：某些资源只在特定条件下需要管理
3. 清理回调注册：把任意清理函数注册到栈上，统一在退出时执行

动态管理多个资源

Python

from contextlib import ExitStack


def process_sensor_data(paths: list[str]):
    """同时打开多个传感器数据文件处理，数量在运行时确定"""
    with ExitStack() as stack:
        files = [
            stack.enter_context(open(p, encoding="utf-8"))
            for p in paths
        ]
        # files 是所有已打开的文件对象列表
        # 离开 with 块时，ExitStack 会按"后进先出"顺序关闭所有文件
        for i, f in enumerate(files):
            print(f"传感器 {i} 数据：{f.read(20)!r}")


import tempfile, os

# 创建临时测试文件
tmpdir = tempfile.mkdtemp()
paths = []
for i, content in enumerate(["AAA", "BBB", "CCC"], 1):
    p = os.path.join(tmpdir, f"sensor_{i}.txt")
    with open(p, "w") as f:
        f.write(content * 10)
    paths.append(p)

try:
    process_sensor_data(paths)
finally:
    for p in paths:
        os.unlink(p)
    os.rmdir(tmpdir)

注册清理回调

stack.callback(fn, *args) 把一个函数注册到栈上，退出时按"后进先出"顺序调用：

Python

from contextlib import ExitStack


def cleanup_sensor(name: str):
    print(f"清理传感器 {name}")


with ExitStack() as stack:
    stack.callback(cleanup_sensor, "IMU")
    stack.callback(cleanup_sensor, "GPS")
    stack.callback(cleanup_sensor, "磁力计")
    print("采集中...")
# 输出：
# 采集中...
# 清理传感器 磁力计
# 清理传感器 GPS
# 清理传感器 IMU

ExitStack 是资源管理的瑞士军刀

• enter_context(cm)：把一个上下文管理器压入栈
• callback(fn, *args, **kwargs)：注册一个清理回调
• push(cm) 或 push(exit_fn)：手动压入上下文管理器或退出函数
• 退出时按后进先出（LIFO）顺序执行清理，符合"先开后关"的资源管理直觉

条件性进入上下文

Python

from contextlib import ExitStack


def safe_write_sensor_data(path: str, data: str, backup: bool = False):
    """写传感器数据，可选地先备份原文件"""
    with ExitStack() as stack:
        if backup:
            # 仅在需要时进入"备份"上下文
            backup_path = path + ".bak"
            with open(path, encoding="utf-8") as src, \
                 open(backup_path, "w", encoding="utf-8") as dst:
                dst.write(src.read())
            print(f"已备份到 {backup_path}")

        # 写入新内容
        f = stack.enter_context(open(path, "w", encoding="utf-8"))
        f.write(data)
        print(f"已写入 {path}")


import tempfile, os
path = os.path.join(tempfile.gettempdir(), "sensor_data.txt")
with open(path, "w") as f:
    f.write("原始数据")

safe_write_sensor_data(path, "新数据", backup=True)
# 输出：
# 已备份到 .../sensor_data.txt.bak
# 已写入 .../sensor_data.txt

async with

异步代码（async/await）中使用 async with 管理异步资源。异步上下文管理器实现的是 __aenter__ 和 __aexit__ 方法（注意是 a 前缀，且是协程）。

Python

import asyncio


class AsyncFlightLogger:
    """模拟异步飞行数据记录器"""

    async def __aenter__(self):
        print("初始化异步飞行记录器...")
        await asyncio.sleep(0.1)        # 模拟异步初始化
        self.connected = True
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print("关闭异步飞行记录器...")
        await asyncio.sleep(0.05)        # 模拟异步关闭
        self.connected = False
        return False

    async def log_data(self, data: dict) -> str:
        if not self.connected:
            raise RuntimeError("记录器未连接")
        await asyncio.sleep(0.01)
        return f"记录成功：{data}"


async def main():
    async with AsyncFlightLogger() as logger:
        result = await logger.log_data({"altitude": 1000, "speed": 50})
        print(result)
    print(f"记录器状态：connected={logger.connected}")
# 输出：
# 初始化异步飞行记录器...
# 记录成功：{'altitude': 1000, 'speed': 50}
# 关闭异步飞行记录器...
# 记录器状态：connected=False


asyncio.run(main())

contextlib.asynccontextmanager

类似 @contextmanager，@contextlib.asynccontextmanager 用异步生成器函数实现异步上下文管理器：

Python

from contextlib import asynccontextmanager
import asyncio


@asynccontextmanager
async def async_flight_buffer(buffer_id: str):
    """异步飞行数据缓冲区（生成器版本）"""
    print(f"初始化异步缓冲区：{buffer_id}")
    await asyncio.sleep(0.1)
    buffer = {"id": buffer_id, "data": [], "flushed": False}
    try:
        yield buffer
    finally:
        await asyncio.sleep(0.05)
        buffer["flushed"] = True
        print(f"异步缓冲区 {buffer_id} 已刷新")


async def main():
    async with async_flight_buffer("ASYNC-BUF-01") as buf:
        buf["data"].append({"t": 0.0, "value": 42.0})
        print(f"使用缓冲区 id={buf['id']}")


asyncio.run(main())
# 输出：
# 初始化异步缓冲区：ASYNC-BUF-01
# 使用缓冲区 id=ASYNC-BUF-01
# 异步缓冲区 ASYNC-BUF-01 已刷新

同步 with vs 异步 async with

• 同步 with：调用 __enter__/__exit__（普通方法）
• 异步 async with：调用 __aenter__/__aexit__（协程方法）

两者不能混用：在 async 函数里必须用 async with，在普通函数里用 with。异步上下文管理器必须实现 __aenter__/__aexit__，普通 __enter__/__exit__ 不行。

contextlib 实用工具

contextlib 标准库提供了几个开箱即用的上下文管理器，覆盖常见场景。

contextlib.suppress

静默忽略指定的异常，等价于 try-except-pass 但更明确：

Python

from contextlib import suppress

# 旧写法：try-except-pass
try:
    os.remove("可能不存在的校准文件.txt")
except FileNotFoundError:
    pass

# 新写法：用 suppress 更简洁
with suppress(FileNotFoundError):
    os.remove("可能不存在的校准文件.txt")

# 可以同时忽略多种异常
with suppress(FileNotFoundError, PermissionError):
    os.remove("另一个校准文件.txt")

suppress 比 except pass 更明确

with suppress(FileNotFoundError): 比 try: ... except FileNotFoundError: pass 更能表达"故意忽略此异常"的意图，代码更紧凑，也更不容易写错（不会漏掉异常类型而吞掉其他错误）。

contextlib.redirect_stdout

把 print 等输出到 sys.stdout 的内容重定向到其他流：

Python

from contextlib import redirect_stdout
import io


def noisy_sensor_read():
    print("传感器初始化日志")
    print("校准数据输出")


# 旧写法：手动替换 sys.stdout
import sys
old = sys.stdout
sys.stdout = io.StringIO()
try:
    noisy_sensor_read()
    captured = sys.stdout.getvalue()
finally:
    sys.stdout = old

# 新写法：用 redirect_stdout
buffer = io.StringIO()
with redirect_stdout(buffer):
    noisy_sensor_read()
    print("这条也被捕获")        # 不会显示
print("捕获的内容：", repr(buffer.getvalue()))
# 输出：
# 捕获的内容： '传感器初始化日志\n校准数据输出\n这条也被捕获\n'

# 重定向到文件
with open("sensor_output.log", "w", encoding="utf-8") as f, \
     redirect_stdout(f):
    print("这条写入文件，不在控制台显示")
print("控制台输出正常")

还有 redirect_stderr（重定向 stderr）和 redirect_stdin（重定向 stdin）。

contextlib.redirect_stderr

Python

from contextlib import redirect_stderr
import io
import sys


def sensor_warning():
    print("正常输出", file=sys.stdout)
    print("警告输出", file=sys.stderr)


buffer = io.StringIO()
with redirect_stderr(buffer):
    sensor_warning()
print(f"捕获的 stderr：{buffer.getvalue()!r}")
# 输出：
# 正常输出
# 捕获的 stderr：'警告输出\n'

contextlib.closing

为只有 close() 方法但没有实现 __enter__/__exit__ 的对象提供上下文管理器协议：

Python

from contextlib import closing


class SensorConnection:
    """只有 close() 方法，没有实现上下文管理器协议"""

    def __init__(self, sensor_id: str):
        self.sensor_id = sensor_id
        print(f"连接传感器 {sensor_id}")

    def read(self) -> dict:
        return {"sensor": self.sensor_id, "value": 42.0}

    def close(self):
        print(f"断开传感器 {self.sensor_id}")


# 不用 closing：需要手动 try-finally
conn = SensorConnection("IMU-02")
try:
    data = conn.read()
    print(data)
finally:
    conn.close()

# 用 closing：自动调用 close()
with closing(SensorConnection("IMU-02")) as conn:
    data = conn.read()
    print(data)
# 输出：
# 连接传感器 IMU-02
# {'sensor': 'IMU-02', 'value': 42.0}
# 断开传感器 IMU-02

closing 的现代替代品

Python 标准库的绝大多数资源对象（文件、urllib.request.urlopen 返回的对象、socket 等）都已经实现了完整的上下文管理器协议，可以直接用 with，不需要 closing。closing 主要用于第三方库或老代码中只有 close() 方法的对象。

contextlib.chdir

Python 3.11+ 引入，临时改变当前工作目录，退出时恢复：

Python

from contextlib import chdir
import os


print(f"原工作目录：{os.getcwd()}")

with chdir("/tmp"):
    print(f"块内工作目录：{os.getcwd()}")

print(f"恢复后工作目录：{os.getcwd()}")

Python 3.11+ 新增

contextlib.chdir 是 Python 3.11 才加入的。在更早版本中，需要手动保存并恢复 os.getcwd()：

Python

old_cwd = os.getcwd()
try:
    os.chdir("/tmp")
    # ...
finally:
    os.chdir(old_cwd)

实战：飞行数据缓冲区管理

下面实现一个完整的飞行数据缓冲区系统，综合运用类上下文管理器、@contextmanager、ExitStack 和 suppress：

Python

from __future__ import annotations

from contextlib import contextmanager, suppress
import time
import numpy as np
from collections import deque
from typing import Iterator


# ========== 飞行数据缓冲区（类形式的上下文管理器）==========

class FlightDataBuffer:
    """模拟飞行数据缓冲区，实现完整的上下文管理器协议"""

    _next_id = 1

    def __init__(self, buffer_name: str, capacity: int = 1000):
        self.buffer_name = buffer_name
        self.capacity = capacity
        self.id = FlightDataBuffer._next_id
        FlightDataBuffer._next_id += 1
        self.active = False
        self.data: deque[dict] = deque(maxlen=capacity)
        self.flushed = False

    def __enter__(self) -> "FlightDataBuffer":
        print(f"  [缓冲区 {self.id}] 激活：{self.buffer_name}（容量 {self.capacity}）")
        self.active = True
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is not None:
            print(f"  [缓冲区 {self.id}] 异常，刷新并丢弃数据")
            self.data.clear()
        else:
            print(f"  [缓冲区 {self.id}] 正常刷新，{len(self.data)} 条数据已保存")
        self.flushed = True
        self.active = False
        print(f"  [缓冲区 {self.id}] 已关闭")
        return False      # 不吞异常

    def write(self, record: dict):
        if not self.active:
            raise RuntimeError(f"缓冲区 {self.id} 未激活")
        self.data.append({"timestamp": time.time(), **record})

    def get_data_array(self) -> np.ndarray:
        """将缓冲区数据转换为 numpy 数组"""
        if not self.data:
            return np.array([])
        values = [d.get("value", 0) for d in self.data]
        return np.array(values)


# ========== 传感器池（用 @contextmanager 生成器版本）==========

class SensorPool:
    """简单的传感器连接池"""

    def __init__(self, sensor_type: str, max_size: int = 5):
        self.sensor_type = sensor_type
        self.max_size = max_size
        self._pool: deque[FlightDataBuffer] = deque()
        self._in_use: set[FlightDataBuffer] = set()
        # 预创建两个缓冲区
        for _ in range(2):
            self._pool.append(FlightDataBuffer(f"{sensor_type}-pool"))

    @contextmanager
    def acquire(self) -> Iterator[FlightDataBuffer]:
        """从池中获取缓冲区，使用完归还。

        这是一个生成器形式的上下文管理器，使用 try-finally
        保证缓冲区无论如何都会归还到池中。
        """
        buf = self._get_or_create()
        self._in_use.add(buf)
        try:
            yield buf
        finally:
            self._in_use.discard(buf)
            self._pool.append(buf)
            print(f"  [缓冲区 {buf.id}] 归还到池中")

    def _get_or_create(self) -> FlightDataBuffer:
        if self._pool:
            return self._pool.popleft()
        if len(self._in_use) < self.max_size:
            return FlightDataBuffer(f"{self.sensor_type}-new")
        raise RuntimeError("传感器池已满，请稍后再试")

    @property
    def stats(self) -> dict:
        return {
            "pool_size": len(self._pool),
            "in_use": len(self._in_use),
            "total": len(self._pool) + len(self._in_use),
        }


# ========== 数据采集上下文管理器（嵌套使用）==========

@contextmanager
def data_collection_session(buf: FlightDataBuffer, test_id: str):
    """数据采集会话上下文：进入时开始采集，退出时根据情况保存或丢弃"""
    print(f"  [采集] 开始会话 {test_id}")
    buf.write({"event": "session_start", "test_id": test_id})
    try:
        yield buf
    except Exception as exc:
        print(f"  [采集] 异常 {exc}，将丢弃数据")
        raise       # 重新抛出，让缓冲区的 __exit__ 也能感知
    else:
        print(f"  [采集] 正常结束，将保存数据")


# ========== 业务逻辑：用传感器池记录数据 ==========

def record_flight_data(pool: SensorPool, test_id: str, altitude: float) -> dict:
    """通过传感器池记录飞行数据"""
    with pool.acquire() as buf:
        # 在缓冲区上再嵌套一个采集会话上下文
        with data_collection_session(buf, test_id):
            buf.write({"altitude": altitude, "event": "recording"})
            return {"test_id": test_id, "altitude": altitude, "points": len(buf.data)}


def record_with_error(pool: SensorPool, test_id: str) -> None:
    """记录数据时演示异常回滚"""
    with pool.acquire() as buf:
        with data_collection_session(buf, test_id):
            buf.write({"event": "bad_data"})
            # 故意在某些情况下让后续操作失败，触发数据丢弃
            if "FAIL" in test_id:
                raise RuntimeError("传感器故障")   # 抛 RuntimeError


# ========== 演示 ==========

def demo():
    pool = SensorPool("IMU", max_size=3)

    print("=== 场景 1：正常记录 ===")
    result = record_flight_data(pool, "TEST-001", altitude=100.0)
    print(f"记录结果：{result}")
    print(f"传感器池状态：{pool.stats}\n")

    print("=== 场景 2：正常记录 ===")
    result = record_flight_data(pool, "TEST-002", altitude=200.0)
    print(f"记录结果：{result}")
    print(f"传感器池状态：{pool.stats}\n")

    print("=== 场景 3：异常丢弃 ===")
    try:
        record_with_error(pool, "FAIL-TEST")
    except RuntimeError as e:
        print(f"业务层捕获到错误：{e}")
    print(f"传感器池状态：{pool.stats}\n")

    print("=== 场景 4：suppress 忽略特定异常 ===")
    # 假设"删除临时数据文件"是非致命错误，用 suppress 静默忽略
    from contextlib import suppress
    with suppress(RuntimeError):
        with pool.acquire() as buf:
            buf.write({"event": "temp_data"})
            raise RuntimeError("临时文件删除失败")
    print("尽管操作失败，但被静默忽略，继续执行")
    print(f"传感器池状态：{pool.stats}\n")

    print("=== 场景 5：批量采集（模拟并发场景）===")
    # 用 ExitStack 动态管理多个缓冲区
    from contextlib import ExitStack
    test_ids = ["TEST-A", "TEST-B", "TEST-C"]
    with ExitStack() as stack:
        bufs = [stack.enter_context(pool.acquire()) for _ in test_ids]
        print(f"同时持有了 {len(bufs)} 个缓冲区")
        print(f"传感器池状态：{pool.stats}")
        for i, buf in enumerate(bufs, 1):
            buf.write({"test_id": test_ids[i-1], "value": i * 10})
            print(f"  缓冲区 {buf.id}：已写入数据")
    # 退出后所有缓冲区归还
    print(f"传感器池状态：{pool.stats}")


demo()

输出示例：

输出

=== 场景 1：正常记录 ===
  [缓冲区 1] 激活：IMU-pool（容量 1000）
  [采集] 开始会话 TEST-001
  [采集] 正常结束，将保存数据
  [缓冲区 1] 正常刷新，2 条数据已保存
  [缓冲区 1] 已关闭
  [缓冲区 1] 归还到池中
记录结果：{'test_id': 'TEST-001', 'altitude': 100.0, 'points': 2}
传感器池状态：{'pool_size': 2, 'in_use': 0, 'total': 2}

=== 场景 2：正常记录 ===
  [缓冲区 1] 激活：IMU-pool（容量 1000）
  [采集] 开始会话 TEST-002
  [采集] 正常结束，将保存数据
  [缓冲区 1] 正常刷新，2 条数据已保存
  [缓冲区 1] 已关闭
  [缓冲区 1] 归还到池中
记录结果：{'test_id': 'TEST-002', 'altitude': 200.0, 'points': 2}
传感器池状态：{'pool_size': 2, 'in_use': 0, 'total': 2}

=== 场景 3：异常丢弃 ===
  [缓冲区 1] 激活：IMU-pool（容量 1000）
  [采集] 开始会话 FAIL-TEST
  [采集] 异常 传感器故障，将丢弃数据
  [缓冲区 1] 异常，刷新并丢弃数据
  [缓冲区 1] 已关闭
  [缓冲区 1] 归还到池中
业务层捕获到错误：传感器故障
传感器池状态：{'pool_size': 2, 'in_use': 0, 'total': 2}

=== 场景 4：suppress 忽略特定异常 ===
  [缓冲区 1] 激活：IMU-pool（容量 1000）
  [缓冲区 1] 异常，刷新并丢弃数据
  [缓冲区 1] 已关闭
  [缓冲区 1] 归还到池中
尽管操作失败，但被静默忽略，继续执行
传感器池状态：{'pool_size': 2, 'in_use': 0, 'total': 2}

=== 场景 5：批量采集（模拟并发场景）===
  [缓冲区 2] 归还到池中
  [缓冲区 3] 激活：IMU-new（容量 1000）
  [缓冲区 3] 归还到池中
  [缓冲区 4] 激活：IMU-new（容量 1000）
  [缓冲区 4] 归还到池中
同时持有了 3 个缓冲区
传感器池状态：{'pool_size': 0, 'in_use': 3, 'total': 3}
  缓冲区 2：已写入数据
  缓冲区 3：已写入数据
  缓冲区 4：已写入数据
传感器池状态：{'pool_size': 3, 'in_use': 0, 'total': 3}

嵌套上下文管理器的执行顺序

上面 with pool.acquire() as buf: with data_collection_session(buf, test_id): ... 涉及两个嵌套的上下文管理器。执行顺序是：

1. 进入外层：pool.acquire() 的 __enter__（缓冲区出池）
2. 进入内层：data_collection_session() 的 __enter__（开始采集）
3. 执行 with 块
4. 退出内层：data_collection_session() 的 __exit__（保存或丢弃）
5. 退出外层：pool.acquire() 的 __exit__（缓冲区归还）

外层管理"缓冲区资源"，内层管理"采集状态"——这种分层设计让每层只关注一件事。

高级实战：异步飞行数据记录器

🔧 异步飞行数据记录器实战

下面实现一个生产级的异步飞行数据记录器，将异常处理、上下文管理与异步并发完美融合：

Python

import asyncio
from contextlib import asynccontextmanager


class AsyncFlightRecorderPool:
    """模拟异步飞行数据记录器池"""

    async def acquire(self):
        await asyncio.sleep(0.1)
        print("  [Pool] 🟢 获取记录器")
        return self

    async def release(self, recorder):
        await asyncio.sleep(0.05)
        print("  [Pool] 🔵 释放记录器")

    async def flush(self):
        await asyncio.sleep(0.1)
        print("  [Recorder] ✅ 数据已刷新")

    async def discard(self):
        await asyncio.sleep(0.1)
        print("  [Recorder] 🔴 数据已丢弃")


@asynccontextmanager
async def async_flight_session(pool: AsyncFlightRecorderPool):
    """异步飞行数据采集会话"""
    recorder = await pool.acquire()
    try:
        yield recorder
    except Exception as e:
        print(f"  [Session] ❌ 异常丢弃: {e}")
        await recorder.discard()
        raise
    else:
        await recorder.flush()
    finally:
        await pool.release(recorder)


async def log_flight_data(recorder, data_id: int):
    if data_id < 0:
        raise ValueError(f"非法数据ID: {data_id}")
    await asyncio.sleep(0.2)
    print(f"  [Biz] 📊 记录数据 {data_id} 成功")


async def main():
    pool = AsyncFlightRecorderPool()
    try:
        async with async_flight_session(pool) as recorder:
            await log_flight_data(recorder, 101)
            await log_flight_data(recorder, -1)
    except ValueError as e:
        print(f"[Main] 🛑 捕获业务阻断: {e}")


asyncio.run(main())

执行流程深度解析：

输出

[Pool] 🟢 获取记录器          ← await pool.acquire()
[Biz] 📊 记录数据 101 成功     ← await log_flight_data(recorder, 101)
[Session] ❌ 异常丢弃: 非法...  ← except 捕获 ValueError + await discard()
[Pool] 🔵 释放记录器           ← finally 中 await release() (100%执行)
[Main] 🛑 捕获业务阻断...      ← 外层 try/except 接住重新 raise 的异常

关键设计要点：

1. @asynccontextmanager 替代手写 __aenter__ / __aexit__ 的繁琐模板代码
2. yield 必须被 try/finally 包裹，确保异常时记录器必定被释放
3. except 中丢弃后必须 raise，会话管理器不应吞掉异常
4. 所有 I/O 操作都用 await，事件循环不会被卡住

进阶技巧：contextlib 工具链速查

工具	核心作用	典型场景
closing(obj)	自动调用对象的 close() 方法	处理未实现上下文协议的旧式 API
suppress(*exc)	静默忽略指定的异常	替代 try...except...pass 的冗余写法
nullcontext()	什么都不做的空上下文	条件分支中保持 with 语句结构统一
redirect_stdout()	临时重定向标准输出	捕获第三方库的 print 输出

📋 小结

• with 语句保证资源在退出时被释放，无论是否发生异常
• 类形式的上下文管理器实现 __enter__ 和 __exit__，前者返回值通过 as 绑定，后者返回 True 可抑制异常
• @contextlib.contextmanager 用生成器函数简化实现：yield 前是"进入"，yield 后是"退出"，通常用 try-finally 保证清理
• ExitStack 用于动态管理数量不定的上下文管理器、条件性进入、注册回调，按 LIFO 顺序退出
• async with 用于异步代码，对应 __aenter__/__aexit__，@contextlib.asynccontextmanager 是异步生成器版本
• contextlib 实用工具：
  • suppress(*exceptions)：静默忽略指定异常
  • redirect_stdout/stderr：重定向输出流
  • closing(obj)：为只有 close() 的对象提供上下文管理
  • chdir(path)（3.11+）：临时切换工作目录
• 实战：飞行数据缓冲区管理展示分层资源管理——SensorPool.acquire() 管理缓冲区、data_collection_session() 管理采集状态、嵌套使用各司其职

至此，异常处理章节告一段落。从识别异常、try-except 处理、自定义异常到上下文管理，已涵盖编写健壮 Python 代码所需的核心技能。