摘要：Python并发编程中，线程池(ThreadPoolExecutor)能有效解决多线程任务开销问题。通过预先创建并管理一组线程，线程池可以复用线程资源，避免频繁创建销毁线程的开销。concurrent.futures模块提供了高层接口，其中Executor抽象类有两个具体实现：ThreadPoolExecutor(线程池)和ProcessPoolExecutor(进程池)。ThreadPoolExecutor支持submit()方法提交任务返回Future对象，以及map()方法批量处理可迭代对象。示例展示了如何创建线程池、提交任务、获取结果，以及使用上下文管理器简化代码。相比直接创建线程，线程池能显著提高程序性能，特别是在处理大量短时任务时。

目录

Python 并发编程 —— 线程池

Python 模块：concurrent.futures

Executor 类

ThreadPoolExecutor：Executor 的具体子类

如何创建 ThreadPoolExecutor？

示例

输出结果

以上下文管理器方式实例化 ThreadPoolExecutor

示例

输出结果

Executor.map () 函数的使用

示例

输出结果

---

Python 并发编程 —— 线程池

如果我们需要为多线程任务创建大量线程，会产生极高的计算开销，过多的线程还会引发诸多性能问题，其中一个主要问题就是吞吐量受限。我们可以通过创建线程池来解决这一问题。线程池可定义为一组预先实例化、处于空闲状态且随时准备接收任务的线程。当需要执行大量任务时，创建线程池远比为每个任务都实例化新线程更优。线程池可通过以下方式管理大量线程的并发执行：

• 线程池中的线程完成执行后，可被重新利用；
• 若某个线程终止，会自动创建新线程来替代它。

Python 模块：concurrent.futures

Python 标准库中包含concurrent.futures模块，该模块在 Python 3.2 版本中被引入，为开发者提供了启动异步任务的高层级接口。它是构建在 Python 线程和多进程模块之上的抽象层，能为通过线程池或进程池运行任务提供统一接口。

在后续小节中，我们将学习 concurrent.futures 模块的不同类。

Executor 类

Executor是 concurrent.futures 模块中的一个抽象类，无法直接使用，需要借助它的以下两个具体子类：

• ThreadPoolExecutor（线程池执行器）
• ProcessPoolExecutor（进程池执行器）

ThreadPoolExecutor：Executor 的具体子类

该类是 Executor 抽象类的具体子类之一，基于多线程实现，会为任务提交提供一个线程池。线程池会将任务分配给可用线程，并调度线程执行。

如何创建 ThreadPoolExecutor？

借助 concurrent.futures 模块及其具体子类 ThreadPoolExecutor，我们可以轻松创建线程池。首先需要根据需求的线程数量构造一个 ThreadPoolExecutor 实例（默认线程数为 5），随后即可向线程池提交任务。

调用submit()方法提交任务时，会返回一个Future对象。该对象包含 **done ()** 方法，可用于判断对应的异步任务是否执行完成（即 Future 对象是否已解析出结果）。当任务完成时，线程池执行器会为该 Future 对象设置执行结果。

示例

from concurrent.futures import ThreadPoolExecutor
from time import sleep
def task(message):
 sleep(2)
 return message

def main():
 executor = ThreadPoolExecutor(5)
 future = executor.submit(task, ("Completed"))
 print(future.done())
 sleep(2)
 print(future.done())
 print(future.result())
if __name__ == '__main__':
 main()

输出结果

plaintext

False
True
Completed

在上述示例中，我们构造了一个包含 5 个线程的 ThreadPoolExecutor 实例，然后向其提交了一个任务 —— 该任务会等待 2 秒后返回指定信息。从输出结果可以看到，任务在 2 秒内未完成，因此第一次调用 done() 方法返回 False；2 秒后任务执行完毕，再次调用 done() 方法返回 True，此时调用 result() 方法即可获取该 Future 对象对应的任务结果。

以上下文管理器方式实例化 ThreadPoolExecutor

实例化 ThreadPoolExecutor 还有另一种方式 —— 使用上下文管理器，其功能与上述方式一致，核心优势是语法更简洁、可读性更高。

通过以下代码即可实现上下文管理器方式的实例化：

with ThreadPoolExecutor(max_workers = 5) as executor:

示例

以下示例源自 Python 官方文档。首先需要导入 concurrent.futures 模块，然后定义一个 load_url() 函数，用于加载指定的 URL 地址。随后创建包含 5 个线程的 ThreadPoolExecutor 实例，并将其作为上下文管理器使用，最终通过调用 Future 对象的 result() 方法获取任务结果。

import concurrent.futures
import urllib.request

URLS = ['http://www.foxnews.com/',
 'http://www.cnn.com/',
 'http://europe.wsj.com/',
 'http://www.bbc.co.uk/',
 'http://some-made-up-domain.com/']

def load_url(url, timeout):
 with urllib.request.urlopen(url, timeout = timeout) as conn:
 return conn.read()

with concurrent.futures.ThreadPoolExecutor(max_workers = 5) as executor:
 # 提交所有URL的加载任务，构建Future对象与URL的映射
 future_to_url = {executor.submit(load_url, url, 60): url for url in URLS}
 # 遍历已完成的Future对象
 for future in concurrent.futures.as_completed(future_to_url):
 url = future_to_url[future]
 try:
 data = future.result()
 except Exception as exc:
 print('%r 执行出错：%s' % (url, exc))
 else:
 print('%r 页面大小为 %d 字节' % (url, len(data)))

输出结果

上述 Python 脚本的输出结果如下：

plaintext

'http://some-made-up-domain.com/' 执行出错：<urlopen error [Errno 11004] getaddrinfo failed>
'http://www.foxnews.com/' 页面大小为 229313 字节
'http://www.cnn.com/' 页面大小为 168933 字节
'http://www.bbc.co.uk/' 页面大小为 283893 字节
'http://europe.wsj.com/' 页面大小为 938109 字节

Executor.map () 函数的使用

Python 中的map()函数被广泛用于各类任务，其中一个典型用途是将某个函数应用于可迭代对象的每一个元素。同理，我们可以使用 Executor.map() 函数，将迭代器的所有元素映射到指定函数，并将这些映射后的任务作为独立作业提交给ThreadPoolExecutor。

通过以下 Python 脚本示例，可理解该函数的工作原理。

示例

在本示例中，map() 函数被用于将square()（平方）函数应用于 values 数组中的每一个值。

from concurrent.futures import ThreadPoolExecutor
values = [2,3,4,5]
def square(n):
 return n * n
def main():
 with ThreadPoolExecutor(max_workers = 3) as executor:
 results = executor.map(square, values)
 for result in results:
 print(result)
if __name__ == '__main__':
 main()

输出结果

上述 Python 脚本的输出结果如下：

plaintext

4
9
16
25