当前位置：首页 > news >正文

Python面试题：如何在 Python 中实现一个线程池？

news 2025/8/2 7:50:34

在 Python 中，实现线程池可以使用内置的 concurrent.futures 模块，该模块提供了一个高层次的接口来管理并发任务。ThreadPoolExecutor 类是实现线程池的主要工具。以下是一些使用示例，展示如何在 Python 中实现和使用线程池：

1. 基本使用

首先，创建一个线程池，并使用 submit 方法将任务提交给线程池：

from concurrent.futures import ThreadPoolExecutor, as_completed
import timedef task(n):print(f"Task {n} is running")time.sleep(2)return f"Task {n} completed"# 创建一个线程池，最大线程数为3
with ThreadPoolExecutor(max_workers=3) as executor:# 提交多个任务futures = [executor.submit(task, i) for i in range(5)]# 等待任务完成并获取结果for future in as_completed(futures):print(future.result())

在这个示例中，线程池最多可以同时运行3个任务，所有5个任务会被提交到线程池，并通过 as_completed 函数等待它们的完成。

2. 使用 `map` 方法

ThreadPoolExecutor 提供了一个 map 方法，可以用来将一个函数应用到一个可迭代对象的每个元素上，类似于内置的 map 函数，但它会并行地执行任务：

from concurrent.futures import ThreadPoolExecutor
import timedef task(n):print(f"Task {n} is running")time.sleep(2)return f"Task {n} completed"# 创建一个线程池，最大线程数为3
with ThreadPoolExecutor(max_workers=3) as executor:# 使用 map 方法并行执行任务results = executor.map(task, range(5))# 打印结果for result in results:print(result)

3. 处理异常

在提交任务时，如果任务中发生异常，ThreadPoolExecutor 可以捕获并处理这些异常：

from concurrent.futures import ThreadPoolExecutordef task(n):if n == 2:raise ValueError("An error occurred in task 2")return f"Task {n} completed"# 创建一个线程池
with ThreadPoolExecutor(max_workers=3) as executor:futures = [executor.submit(task, i) for i in range(5)]for future in as_completed(futures):try:result = future.result()except Exception as e:print(f"Task raised an exception: {e}")else:print(result)

4. 使用 `shutdown` 方法

线程池可以使用 shutdown 方法来显式关闭。在 with 语句块中使用 ThreadPoolExecutor 时，它会自动调用 shutdown 方法，但你也可以显式调用它：

from concurrent.futures import ThreadPoolExecutordef task(n):print(f"Task {n} is running")return f"Task {n} completed"# 创建一个线程池
executor = ThreadPoolExecutor(max_workers=3)
futures = [executor.submit(task, i) for i in range(5)]# 等待所有任务完成
for future in as_completed(futures):print(future.result())# 显式关闭线程池
executor.shutdown()