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

Python使用多线程和异步调用

news 2025/9/16 14:51:03

概述

在 Python 中，多线程和异步调用是处理并发任务的两种常用方式，适用于不同场景。

多线程（threading 模块）

多线程适合处理 I/O 密集型任务（如网络请求、文件读写），因为这类任务大部分时间在等待，线程可以在等待时切换到其他任务。

import threading
import timedef task(name, delay):"""线程执行的任务"""print(f"任务 {name} 开始")time.sleep(delay)  # 模拟I/O操作（如网络请求）print(f"任务 {name} 完成")# 创建线程
t1 = threading.Thread(target=task, args=("A", 2))
t2 = threading.Thread(target=task, args=("B", 3))# 启动线程
t1.start()
t2.start()# 等待所有线程完成
t1.join()
t2.join()print("所有任务完成")

特点

优势：适合 I/O 密集型任务，线程切换开销小。
限制：受 GIL（全局解释器锁）影响，CPU 密集型任务（如大量计算）无法真正并行，多线程可能比单线程更慢。
适用场景：爬虫、API 调用、文件读写等。

线程池（ThreadPoolExecutor）

在 Python 中，线程池（ThreadPoolExecutor）是管理线程的高效方式，它可以预先创建一定数量的线程，避免频繁创建和销毁线程带来的开销，特别适合处理大量短期的 I/O 密集型任务（如网络请求、文件读写等）。

ThreadPoolExecutor 位于 concurrent.futures 模块中，使用起来非常简洁。

基本使用步骤

导入 ThreadPoolExecutor 类
创建线程池实例，指定最大线程数
提交任务到线程池（使用 submit() 或 map() 方法）
获取任务结果（可选）
关闭线程池（通常使用 with 语句自动管理）

示例 1：使用 submit() 提交单个任务

import time
from concurrent.futures import ThreadPoolExecutordef task(task_id, sleep_time):"""线程执行的任务：休眠指定时间后返回结果"""print(f"任务 {task_id} 开始执行，将休眠 {sleep_time} 秒")time.sleep(sleep_time)  # 模拟I/O操作result = f"任务 {task_id} 完成，休眠了 {sleep_time} 秒"return resultdef main():# 创建线程池，最大线程数为3with ThreadPoolExecutor(max_workers=3) as executor:# 提交3个任务到线程池futures = [executor.submit(task, 1, 2),  # 任务1，休眠2秒executor.submit(task, 2, 3),  # 任务2，休眠3秒executor.submit(task, 3, 1)   # 任务3，休眠1秒]# 获取所有任务的结果for future in futures:print(future.result())  # result()方法会阻塞直到任务完成if __name__ == "__main__":start_time = time.time()main()print(f"总耗时：{time.time() - start_time:.2f}秒")

运行结果说明：

3 个任务并发执行，总耗时接近最长任务的耗时（3 秒）
with 语句会自动关闭线程池，无需手动调用 shutdown()

使用 map() 批量处理任务

当需要对多个参数执行相同任务时，map() 方法更简洁：

import time
from concurrent.futures import ThreadPoolExecutordef process_item(item):"""处理单个项目的函数"""item_id, sleep_time = itemprint(f"处理项目 {item_id}，休眠 {sleep_time} 秒")time.sleep(sleep_time)return f"项目 {item_id} 处理完成"def main():# 待处理的项目列表items = [(1, 2), (2, 3), (3, 1), (4, 2)]# 创建线程池，最大线程数为2with ThreadPoolExecutor(max_workers=2) as executor:# 使用map批量处理任务，返回结果顺序与输入顺序一致results = executor.map(process_item, items)# 遍历结果for result in results:print(result)if __name__ == "__main__":start_time = time.time()main()print(f"总耗时：{time.time() - start_time:.2f}秒")

map() 特点：

结果返回顺序与输入列表顺序一致
无需手动收集结果，直接迭代即可

示例 3：处理异常

线程池中的任务异常需要在获取结果时捕获：

import time
from concurrent.futures import ThreadPoolExecutor, as_completeddef risky_task(task_id):"""可能抛出异常的任务"""print(f"任务 {task_id} 开始执行")if task_id == 2:raise ValueError(f"任务 {task_id} 发生错误")time.sleep(1)return f"任务 {task_id} 成功完成"def main():with ThreadPoolExecutor(max_workers=2) as executor:futures = [executor.submit(risky_task, i) for i in range(1, 4)]# 使用as_completed获取已完成的任务（顺序不确定）for future in as_completed(futures):try:result = future.result()print(result)except ValueError as e:print(f"捕获到异常：{e}")if __name__ == "__main__":main()

as_completed() 特点：

按任务完成顺序返回结果（与提交顺序可能不同）
便于实时处理已完成的任务

线程池关键参数与方法

1.max_workers：线程池最大线程数

对于 I/O 密集型任务，通常设置为 CPU核心数 * 5 左右
对于 CPU 密集型任务，建议设置为 CPU核心数 + 1

2.常用方法：

submit(func, *args, **kwargs)：提交单个任务，返回 Future 对象
map(func, *iterables)：批量提交任务，返回结果迭代器
shutdown(wait=True)：关闭线程池，wait=True 表示等待所有任务完成

3.Future 对象方法：

result(timeout=None)：获取任务结果，可设置超时时间
add_done_callback(fn)：任务完成后执行回调函数
cancel()：取消未执行的任务

适用场景

网络爬虫（批量请求网页）
API 接口测试（并发调用多个接口）
日志处理（批量读取多个日志文件）
任何需要并发处理的 I/O 密集型任务

线程池避免了频繁创建线程的开销，同时通过限制最大线程数防止系统资源耗尽，是 Python 并发编程中非常实用的工具。

异步调用（asyncio 模块）

异步编程通过事件循环实现并发，更适合处理高并发 I/O 任务（如大量网络请求），效率通常高于多线程。

import asyncio
import timeasync def task(name, delay):"""异步任务（必须用async定义）"""print(f"任务 {name} 开始")# 模拟I/O操作（必须用await调用异步函数）await asyncio.sleep(delay)print(f"任务 {name} 完成")async def main():# 创建任务列表tasks = [asyncio.create_task(task("A", 2)),asyncio.create_task(task("B", 3))]# 等待所有任务完成await asyncio.gather(*tasks)# 启动事件循环
asyncio.run(main())
print("所有任务完成")

关键概念