Python 并发编程 —— 线程池

摘要Python并发编程中线程池(ThreadPoolExecutor)能有效解决多线程任务开销问题。通过预先创建并管理一组线程线程池可以复用线程资源避免频繁创建销毁线程的开销。concurrent.futures模块提供了高层接口其中Executor抽象类有两个具体实现ThreadPoolExecutor(线程池)和ProcessPoolExecutor(进程池)。ThreadPoolExecutor支持submit()方法提交任务返回Future对象以及map()方法批量处理可迭代对象。示例展示了如何创建线程池、提交任务、获取结果以及使用上下文管理器简化代码。相比直接创建线程线程池能显著提高程序性能特别是在处理大量短时任务时。目录Python 并发编程 —— 线程池Python 模块concurrent.futuresExecutor 类ThreadPoolExecutorExecutor 的具体子类如何创建 ThreadPoolExecutor示例输出结果以上下文管理器方式实例化 ThreadPoolExecutor示例输出结果Executor.map () 函数的使用示例输出结果Python 并发编程 —— 线程池如果我们需要为多线程任务创建大量线程会产生极高的计算开销过多的线程还会引发诸多性能问题其中一个主要问题就是吞吐量受限。我们可以通过创建线程池来解决这一问题。线程池可定义为一组预先实例化、处于空闲状态且随时准备接收任务的线程。当需要执行大量任务时创建线程池远比为每个任务都实例化新线程更优。线程池可通过以下方式管理大量线程的并发执行线程池中的线程完成执行后可被重新利用若某个线程终止会自动创建新线程来替代它。Python 模块concurrent.futuresPython 标准库中包含concurrent.futures模块该模块在 Python 3.2 版本中被引入为开发者提供了启动异步任务的高层级接口。它是构建在 Python 线程和多进程模块之上的抽象层能为通过线程池或进程池运行任务提供统一接口。在后续小节中我们将学习concurrent.futures模块的不同类。Executor 类Executor是concurrent.futures模块中的一个抽象类无法直接使用需要借助它的以下两个具体子类ThreadPoolExecutor线程池执行器ProcessPoolExecutor进程池执行器ThreadPoolExecutorExecutor 的具体子类该类是 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()输出结果plaintextFalse True Completed在上述示例中我们构造了一个包含 5 个线程的ThreadPoolExecutor实例然后向其提交了一个任务 —— 该任务会等待 2 秒后返回指定信息。从输出结果可以看到任务在 2 秒内未完成因此第一次调用done()方法返回False2 秒后任务执行完毕再次调用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 脚本的输出结果如下plaintexthttp://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 脚本的输出结果如下plaintext4 9 16 25