Python concurrent.futures map 的 buffersize 怎样限制排队

我第一次把一批大文件交给 ThreadPoolExecutor.map() 时，误以为“线程数是 8，所以同时只会准备 8 个输入”。实际观察到的却是：任务函数还没来得及处理完，输入生成器已经被持续消费，内存随着待处理数据一起涨。后来把 buffersize 加上，才把“线程并发数”和“已经提交、但结果还没交付的任务数”分成了两个可以分别控制的旋钮。

官方文档：https://docs.python.org/3/library/concurrent.futures.html

buffersize 限制的是 map 已提交但结果尚未被调用方取走的任务数量；缓冲区满时，输入迭代器会暂停。它不是线程数，也不是结果队列的总容量。该参数从 Python 3.14 开始提供。

先分清 buffersize 到底限制什么

Executor.map(fn, iterable, buffersize=N) 的关键是“提交”和“交付”之间的窗口。调用方还没有从返回迭代器取走结果时，已提交任务会占用这个窗口；窗口满了，map() 不再继续向输入迭代器要新数据，直到至少有一个结果被取走。

这和 max_workers 的职责不同：max_workers 控制同时运行的 worker 数量，buffersize 控制调用方尚未消费结果时，允许任务继续向前排队多少。实际运行中，缓冲区里可能同时包含运行中任务和已经完成但尚未交付的任务。

我用一个慢任务看出排队窗口

下面的生成器每产生一个编号就打印一次。任务故意睡眠一小段时间，主线程也故意慢慢读取结果。把 buffersize 改成 2、4 或 None，可以直观看到输入生成器何时被背压。示例使用 Python 3.14 及更高版本。

import time
from concurrent.futures import ThreadPoolExecutor

def source():
    # 用打印观察 map 何时继续消费输入；真实项目可替换为分页读取器。
    for number in range(8):
        print(f"produce {number}")
        yield number

def work(number):
    # 模拟一个有延迟的 I/O 任务，避免所有结果立刻完成。
    time.sleep(0.2)
    return number * number

with ThreadPoolExecutor(max_workers=2) as executor:
    # buffersize=2 让未被主循环取走的提交任务保持在一个小窗口内。
    results = executor.map(work, source(), buffersize=2)
    for result in results:
        print("consume", result)
        # 模拟下游写入或网络发送变慢的情况。
        time.sleep(0.4)

这里不要把打印顺序当成严格的调度证明；线程什么时候完成取决于机器负载。但有一个稳定的语义：当主循环不取结果且窗口已经满时，source() 会暂时停止产生新输入。buffersize=None 则恢复“先收集可迭代对象，再尽量提交”的旧行为，长输入更容易把内存压力推到调用方看不见的地方。

buffersize、max_workers 和 chunksize 不是一回事

我在第一次调参时把这三个参数一起改，结果很难判断哪个参数真正起作用。更稳妥的做法是先固定 worker 数量，再单独改变提交窗口：

官方文档明确说明，chunksize 只会显著影响 ProcessPoolExecutor 的分块提交；对 ThreadPoolExecutor 和 InterpreterPoolExecutor 没有效果。buffersize 则是 map 层的结果交付窗口，二者可以同时设置，但不要用一个参数代替另一个参数。

按序消费会让一个慢结果挡住后面的结果

map() 返回的迭代器按输入顺序产出结果。即使后面的任务已经完成，只要前面的任务还没完成，调用方仍然会在前面等待。此时降低 buffersize 能控制内存和输入压力，却不会把 map() 变成“谁先完成就先返回”。

如果业务真正需要完成即处理，应改用 submit() 配合 as_completed()，并自行维护任务与输入的对应关系。这个选择的代价是代码需要处理异常、取消和结果排序，不能只替换一个参数。

from concurrent.futures import ThreadPoolExecutor, as_completed

def work(number):
    # 真实任务中应在函数内部做好资源释放和必要的超时控制。
    return number * number

with ThreadPoolExecutor(max_workers=4) as executor:
    futures = {
        executor.submit(work, number): number
        for number in range(8)
    }
    for future in as_completed(futures):
        number = futures[future]
        try:
            print(number, future.result())
        except Exception as exc:
            # 异常在取 Future 结果时显式处理，避免静默丢失失败项。
            print("failed", number, repr(exc))

异常、超时和边界值要提前定规则

如果任务函数抛出异常，map() 迭代器会在调用方取到对应结果时重新抛出。已经提交的其他任务不因为这一项异常就自动全部取消，所以需要在业务层决定是否停止继续消费、关闭 executor 或记录失败输入。

timeout 的计时起点是调用 Executor.map() 的时刻，而不是每次调用 next() 的时刻；当调用方等待超过限制，会得到 TimeoutError。这和 worker 函数内部的网络超时不是同一层控制。

• Python 版本：在 Python 3.13 及更早版本，Executor.map() 没有 buffersize 参数；需要兼容旧版本时应使用手动 submit() 窗口或版本分支。
• 空输入：不会产生任务，buffersize 没有可观察效果。
• 小于等于零：不要把 0 当作“完全不排队”；按当前文档语义应传正整数，非法值应在应用启动或配置读取阶段拒绝。
• 多个 iterable：输入按位置配对，最短 iterable 耗尽后停止；窗口限制仍针对已提交但未交付的结果。

什么时候值得使用 buffersize

对我来说，最值得用它的场景是“输入很长、每项占用资源、下游消费速度不稳定”：例如分页读取后并发处理、批量调用外部服务、把结果持续写入文件或数据库。它能把输入端的消费速度绑定到结果端的推进速度，降低无界预提交带来的内存和连接压力。

如果输入本来就只有几十项，或者任务函数几乎瞬间完成，设置它通常不会带来明显收益。此时优先把任务超时、错误处理、幂等和 worker 数量定清楚。对 CPU 密集型任务，还要先判断线程池是否合适，不能用 buffersize 掩盖执行模型选错的问题。

我的实际判断清单

1. 先确认运行环境是否至少是 Python 3.14。
2. 把 max_workers 按外部服务容量或 CPU 资源单独估算。
3. 如果输入是生成器且不能一次性装进内存，再设置一个小的正整数 buffersize。
4. 确认业务接受按输入顺序交付；需要完成即处理时改用 submit() 与 as_completed()。
5. 为函数异常、全局超时、取消和重复执行定义处理策略，并记录输入标识。

一句话总结：max_workers 管“正在干活的工人”，buffersize 管“还没交到你手里的已提交工作”。把两者分开后，map() 的排队行为才真正可控。

相关问题

buffersize=1 会退化成串行吗？

不会。worker 仍由 max_workers 决定，多个任务可以并行运行；只是调用方不取结果时，允许继续保持的未交付窗口更小。

能用 buffersize 限制外部 API 的 QPS 吗？

不能直接等价。它限制的是提交和交付之间的任务窗口，不是时间维度的速率；QPS 仍需要限速器、并发信号量或服务端配额控制。

为什么用了 buffersize 内存仍然会上升？

窗口只约束 map 的提交侧。worker 自身缓存、任务参数、结果对象、下游写入队列以及外部库的连接池仍可能占用内存，需要分别测量和设置上限。