Python 有界异步流水线：背压、保序、取消与清理
前置：迭代器与值对象、线程与协程起点。项目处理 UTF-8 的 名字,整数 逐行数据，上调名字大小写并将数值乘2；它是简单行格式，不支持CSV引号/嵌入逗号，完整CSV请用CSV项目。
本轮复核仅静态阅读正文、源码与测试规格，未执行示例或测试；文中原有实测版本记录属于此前批次，本轮不复用为修改后的运行证明。命令与断言结果均为读者后续验证预期。
协作调度与取消
调用 async def 得到协程对象，await 或创建任务才推进它。事件循环一般在同一线程执行任务，只有遇到实际挂起的 await 才交出执行权；await 一个已就绪结果未必让出。阻塞文件操作、time.sleep 或长CPU循环都会拖住其他任务。本项目 transform 的 sleep(0) 仅提供协作挂起点，不伪装成真实联网或加速CPU。
TaskGroup 为一组相关任务建立退出边界；内部一个任务抛普通异常，组会取消其他任务并等它们结束，再用 ExceptionGroup 报告错误。外层取消传播为 CancelledError；它属于 BaseException，通常不会被 except Exception 捕获。清理放 finally，然后继续传播取消。吞掉 CancelledError 可能破坏超时和组退出，不能随意拿 uncancel 作为补丁。
timeout 是取消驱动的期限，不会终止拒绝协作的CPU函数。to_thread 可避免阻塞事件循环，但取消等待者不代表后台线程停止；需要线程自己的退出协议。默认带GIL的CPython，纯Python CPU任务通常考虑进程；进程要求可序列化参数、模块入口保护和显式关闭，Windows spawn 下不能依赖 fork 继承。无GIL构建和扩展模块有不同前提，仍需保护业务共享不变量。
项目与内存边界
目录 examples/python/pipeline-lab，只需 Python≥3.11；使用3.12.14和3.14.4实测。queue 容量等于 workers，只有固定数量 worker 协程和一个 producer，所以不会为百万行先创建百万任务。producer 在队列满时等待，形成背压。结果按输入序号保存，因此O(n)输出内存；不能宣称整个管道恒定内存。同步 next 仍在事件循环内，慢文件、阻塞迭代器应换接口或隔离。
每个 worker 遇到 None 退出；None 是内部队列哨兵，实际用户 item 被包装在(index,item)中，因此传入None也不会误判结束。发生源读取错误、转换错误或外层取消时，TaskGroup 不再依靠发送哨兵收尾，而是取消并等待所有任务。返回前没有遗留工作协程；外部I/O资源仍由回调自身在finally/async with中清理。不做部分输出，失败时CLI stdout为空。
完整实现 pipeline.py：
"""A bounded, ordered async map; Python 3.11+, standard library only."""
import argparse
import asyncio
from dataclasses import dataclass
from collections.abc import Awaitable, Callable, Iterable, Iterator
from typing import TypeVar
T = TypeVar('T')
R = TypeVar('R')
@dataclass(frozen=True, slots=True)
class Record:
name: str
amount: int
def records(lines: Iterable[str]) -> Iterator[Record]:
"""Yield validated rows lazily; caller owns and closes the input."""
for number, line in enumerate(lines, 1):
fields = line.strip().split(',')
if len(fields) != 2 or not fields[0]:
raise ValueError(f'line {number}: expected name,integer')
try:
amount = int(fields[1])
except ValueError as exc:
raise ValueError(f'line {number}: invalid integer') from exc
yield Record(fields[0], amount)
async def bounded_map(items: Iterable[T], work: Callable[[T], Awaitable[R]], limit: int) -> list[R]:
"""O(limit) tasks/queued input; O(n) ordered output. Fail atomically."""
if limit < 1:
raise ValueError('limit must be positive')
queue: asyncio.Queue[tuple[int, T] | None] = asyncio.Queue(maxsize=limit)
results: dict[int, R] = {}
async def produce():
for index, item in enumerate(items):
await queue.put((index, item))
for _ in range(limit):
await queue.put(None)
async def consume():
while (entry := await queue.get()) is not None:
index, item = entry
results[index] = await work(item)
async with asyncio.TaskGroup() as group:
group.create_task(produce())
for _ in range(limit):
group.create_task(consume())
return [results[index] for index in range(len(results))]
async def transform(row: Record) -> Record:
# A cooperative suspension point stands in for an async I/O API, not a CPU accelerator.
await asyncio.sleep(0)
return Record(row.name.upper(), row.amount * 2)
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument('file')
parser.add_argument('--workers', type=int, default=2)
args = parser.parse_args()
try:
with open(args.file, encoding='utf-8') as stream:
output = asyncio.run(bounded_map(records(stream), transform, args.workers))
for row in output:
print(f'{row.name},{row.amount}')
return 0
except (OSError, ValueError, ExceptionGroup) as exc:
print(f'pipeline: {exc}', file=__import__('sys').stderr)
return 1
if __name__ == '__main__':
raise SystemExit(main())运行与失败结果
在目录下建立 UTF-8 input.txt 两行 ada,2、lin,3。PowerShell 用 [System.IO.File]::WriteAllText((Join-Path (Get-Location) "input.txt"), "ada,2nlin,3n", [System.Text.UTF8Encoding]::new($false)) 明确写入无 BOM 的 UTF-8（兼容 Windows PowerShell 5.1 与 PowerShell 7），Bash 用 printf 'ada,2\nlin,3\n' > input.txt。PowerShell 将下文 python3 换成所选虚拟环境的 python：
python3 -m unittest -v
python3 pipeline.py input.txt --workers 2
python3 pipeline.py input.txt --workers 0
python3 pipeline.py missing.txt测试全部通过；正常输出两行 ADA,4、LIN,6，退出0。workers=0、缺文件、无效UTF-8、坏整数/坏行返回1并在stderr说明失败；缺参数由argparse返回2。整数使用 Python 的 int(text) 解析，支持负号、首尾空白、数字下划线及 Unicode 十进制数字；它不是 ASCII 整数字面量的严格校验器。整数运算本身为任意精度，但解释器的十进制文本转换位数限制仍可能使超长输入或输出失败，这里没有擅自施加金额非负规则。不要给输入加UTF-8 BOM；若对接真实带BOM导出文件，应明确选utf-8-sig并增加测试，而不是隐式吞首字符。
测试与排错
下列全部测试是实际项目的一部分。Event 保证另一个worker已进入等待再触发错误，finally信号确认被取消者确实清理完；外层取消和输入迭代器自身出错分开测试。wait_for 的2秒只做防挂起看门狗，正确性不依赖机器快慢或sleep时序。
import asyncio
import io
import unittest
from pipeline import Record, records, bounded_map
class RecordsTest(unittest.TestCase):
def test_lazy_and_invalid(self):
source = io.StringIO('a,2\nb,3\nwrong\n')
it = records(source)
self.assertEqual(next(it), Record('a', 2))
self.assertEqual(next(it), Record('b', 3))
with self.assertRaisesRegex(ValueError, 'line 3'):
next(it)
def test_frozen_value_and_hash(self):
row = Record('a', 2)
self.assertEqual(len({row, Record('a', 2)}), 1)
with self.assertRaises(AttributeError):
row.amount = 4
class PipelineTest(unittest.IsolatedAsyncioTestCase):
async def test_order_limit_and_empty(self):
active = peak = 0
gate = asyncio.Event()
async def work(n):
nonlocal active, peak
active += 1
peak = max(peak, active)
try:
if n == 0:
await gate.wait()
if n == 1:
gate.set()
await asyncio.sleep(0)
return n * 2
finally:
active -= 1
self.assertEqual(await asyncio.wait_for(bounded_map(range(20), work, 2), 2), list(range(0, 40, 2)))
self.assertEqual(peak, 2)
self.assertEqual(active, 0)
self.assertEqual(await bounded_map([], work, 2), [])
with self.assertRaises(ValueError):
await bounded_map([1], work, 0)
async def test_failure_cancels_and_waits(self):
started = asyncio.Event()
stopped = asyncio.Event()
async def work(n):
if n == 0:
await started.wait()
raise ValueError('bad record')
started.set()
try:
await asyncio.Event().wait()
finally:
stopped.set()
with self.assertRaises(ExceptionGroup) as caught:
await asyncio.wait_for(bounded_map([0, 1], work, 2), 2)
self.assertIsInstance(caught.exception.exceptions[0], ValueError)
self.assertTrue(stopped.is_set())
async def test_outer_cancel_and_source_failure(self):
started = asyncio.Event()
stopped = asyncio.Event()
async def work(n):
started.set()
try:
await asyncio.Event().wait()
finally:
stopped.set()
task = asyncio.create_task(bounded_map([0], work, 1))
await asyncio.wait_for(started.wait(), 2)
task.cancel()
with self.assertRaises(asyncio.CancelledError):
await task
self.assertTrue(stopped.is_set())
def broken():
yield 1
raise ValueError('source failed')
with self.assertRaises(ExceptionGroup):
await asyncio.wait_for(bounded_map(broken(), work, 1), 2)
if __name__ == '__main__':
unittest.main()如果出现 Task was destroyed but it is pending，先查任务是否仍有拥有者、退出前是否await；不要关闭警告。RuntimeWarning: coroutine was never awaited 表示只创建了协程而未运行。程序退出时仍有线程则进一步检查 executor/to_thread 的后台工作，TaskGroup只拥有其创建的异步任务。
练习
① 只有Semaphore而先创建一百万Task，输入内存是否有界？② 一个转换失败，另一个正在等锁，它何时退出？③ 结果保存dict为什么仍有O(n)内存？④ 如何把结果边产生边输出同时保持顺序？
参考解法
① 无界任务仍占用协程帧、参数与调度开销；应使用固定worker和有限队列。② 在可取消等待点收到取消，finally完成后组才退出；若是阻塞同步锁则会卡循环。③ 每行结果都保留到最后，即使输入队列很小也不会释放输出。④ 需单独有序消费者与有限重排窗口，限制序号领先量；若前项很慢无限接受后项，重排缓冲又无界。该扩展不是把dict换成yield就完成。
来源
2026-10-17核对英文协程任务与组、官方中文协程任务、队列、multiprocessing。本项目没有承诺未执行的网络调用、第三方框架或公网部署。
继续阅读
- 知识专题
- 排错
- 排错