Python Academy · 课时

ThreadPoolExecutor

Q: 「ThreadPoolExecutor」这节课中我会学到什么？

并发运行任务 你通过在浏览器中直接运行的动手代码来练习 Python Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

并发运行任务

ThreadPoolExecutor 是 CoddyKit 上的免费 Python Academy 课时。这是第 2 节课，共 4 节。你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。这是 Python Academy 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 Python Academy 课程共包含 4 节课。

为什么需要线程池？

手动创建和加入线程非常繁琐。concurrent.futures.ThreadPoolExecutor 会为您管理一个 worker 线程池，并以简洁的方式返回结果。

from concurrent.futures import ThreadPoolExecutor

def square(n):
    return n * n

with ThreadPoolExecutor() as ex:
    future = ex.submit(square, 5)
    print(future.result())

submit 与结果对象

submit() 会安排一次调用，并立即返回一个结果对象。您可以对它调用 .result() 来获取值；如果值尚未准备好，该调用会一直阻塞。

from concurrent.futures import ThreadPoolExecutor

def greet(name):
    return 'hi ' + name

with ThreadPoolExecutor(max_workers=2) as ex:
    f1 = ex.submit(greet, 'Ana')
    f2 = ex.submit(greet, 'Bob')
    print(f1.result())
    print(f2.result())

使用 map 处理多个输入

executor.map(func, iterable) 会并发地对每个项目运行函数，并按照输入顺序生成结果。

from concurrent.futures import ThreadPoolExecutor

def cube(n):
    return n ** 3

with ThreadPoolExecutor() as ex:
    results = ex.map(cube, range(5))
    print(list(results))

控制 worker 数量

max_workers 会限制同时运行的线程数。对于 I/O 密集型工作，您可以使用比 CPU 核心数更多的 worker。

from concurrent.futures import ThreadPoolExecutor
import time

def io_task(n):
    time.sleep(0.05)
    return n

start = time.perf_counter()
with ThreadPoolExecutor(max_workers=4) as ex:
    print(list(ex.map(io_task, range(4))))
print('ran concurrently')

as_completed

as_completed(futures) 会在每个结果对象完成的瞬间生成它们，而不考虑提交顺序。这非常适合在结果到达后立即处理。

from concurrent.futures import ThreadPoolExecutor, as_completed
import time

def work(n):
    time.sleep(0.01 * (3 - n))
    return n

with ThreadPoolExecutor() as ex:
    futures = [ex.submit(work, i) for i in range(3)]
    for f in as_completed(futures):
        print('finished', f.result())

处理异常

如果某个任务引发异常，异常会存储在其结果对象中，并在您调用 .result() 时再次引发。请将该调用放在 try/except 中。

from concurrent.futures import ThreadPoolExecutor

def risky(n):
    if n == 0:
        raise ValueError('cannot be zero')
    return 10 // n

with ThreadPoolExecutor() as ex:
    f = ex.submit(risky, 0)
    try:
        print(f.result())
    except ValueError as e:
        print('caught:', e)

使用多个参数进行映射

map 接受多个可迭代对象，并将它们像内置 map 一样组合成参数。

from concurrent.futures import ThreadPoolExecutor

def add(a, b):
    return a + b

with ThreadPoolExecutor() as ex:
    print(list(ex.map(add, [1, 2, 3], [10, 20, 30])))

将结果收集到字典中

一种常见模式是将每个结果对象映射回其输入，这样使用 as_completed 时，您就能知道每个结果对应的输入。

from concurrent.futures import ThreadPoolExecutor, as_completed

def length(word):
    return len(word)

words = ['cat', 'tiger', 'ox']
with ThreadPoolExecutor() as ex:
    future_to_word = {ex.submit(length, w): w for w in words}
    out = {}
    for f in as_completed(future_to_word):
        out[future_to_word[f]] = f.result()
print(sorted(out.items()))

上下文管理器关闭

在 with 代码块中使用执行器时，会自动调用 shutdown()，并等待所有待处理任务完成。

from concurrent.futures import ThreadPoolExecutor

def job(n):
    return n * 2

with ThreadPoolExecutor() as ex:
    futures = [ex.submit(job, i) for i in range(3)]
print('all tasks done after the with block')
print([f.result() for f in futures])

检查结果对象状态

结果对象提供了 .done() 和 .running()，因此您可以在不阻塞的情况下检查进度。

from concurrent.futures import ThreadPoolExecutor

def quick(n):
    return n + 1

with ThreadPoolExecutor() as ex:
    f = ex.submit(quick, 41)
    result = f.result()
    print('done?', f.done())
    print('value:', result)

何时使用它

ThreadPoolExecutor 非常适合处理许多小型 I/O 密集型任务，例如 HTTP 调用、文件读取和数据库查询。对于 CPU 密集型工作，请改用 ProcessPoolExecutor。

from concurrent.futures import ThreadPoolExecutor

urls = ['a', 'b', 'c']

def fetch(u):
    return 'fetched ' + u

with ThreadPoolExecutor(max_workers=3) as ex:
    for r in ex.map(fetch, urls):
        print(r)

快速检查

请测试您对 ThreadPoolExecutor 的理解。

总结

您已学会 ThreadPoolExecutor：

submit() 返回一个异步结果对象；map() 按顺序处理可迭代对象。
as_completed() 会在各个异步结果完成时依次返回它们。
异常会通过 result() 显现。
with 代码块会处理关闭操作。

下一步：使用多进程实现真正的并行。

常见问题解答

「ThreadPoolExecutor」课时是免费的吗？

是的 — 「ThreadPoolExecutor」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 Python Academy 课程的其余内容，请升级到 CoddyKit PRO。 Python Academy 课程共包含 4 节课。

「ThreadPoolExecutor」这节课中我会学到什么？

并发运行任务你通过在浏览器中直接运行的动手代码来练习 Python Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 Python Academy 需要有经验吗？

无需任何先前经验。CoddyKit 上的 Python Academy 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。这是第 2 节课，共 4 节。

「ThreadPoolExecutor」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 Python Academy 课中编写并运行代码吗？

能。每节 Python Academy 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

← 返回 Python Academy