本文深入探讨Python在I/O密集型Web API数据抓取中，多进程性能反而下降的常见问题。文章分析了手动创建进程和进程间通信（IPC）带来的高昂开销，并指出I/O密集型任务的特性。教程提供了多线程、异步编程及`multiprocessing.Pool`等优化方案，强调了选择合适并发模型和连接复用的关键性，旨在帮助开发者高效处理网络请求。
在处理需要从Web API抓取大量数据的场景时，开发者常常面临性能瓶颈。当单次API请求耗时较长（例如，由于后端查询复杂或网络延迟），自然会想到利用并发来加速。Python提供了多种并发机制，如多线程、多进程和异步编程。然而，在某些情况下，盲目地使用多进程可能会导致性能不升反降，甚至比单进程串行执行还要慢。本教程将深入分析这一现象，并提供针对I/O密集型任务的优化策略。
单进程基准实现
首先，我们来看一个典型的单进程串行数据抓取实现。这种方法简单直观，但效率较低，因为它必须等待每个API请求完成后才能发起下一个请求。
import requestsimport timedef pull_data(row_data): """ 模拟从Web API拉取数据的函数。 """ url_api = 'https://api.example.com/data' # 替换为实际API地址 post_json = {'query_param': row_data} # 假设row_data是查询参数 try: x = requests.post(url_api, json=post_json, timeout=10) # 增加超时设置 x.raise_for_status() # 检查HTTP请求是否成功 return x.json() except requests.exceptions.RequestException as e: print(f"API请求失败: {e}") return Nonedef single_process_pull(rows): """ 单进程串行拉取数据。 """ all_results = [] for i, row in enumerate(rows): t_s = time.time() result = pull_data(row) if result: all_results.append(result) t_e = time.time() print(f"处理第 {i+1} 条数据完成，耗时 {format(t_e - t_s, '.2f')} 秒") return all_resultsif __name__ == '__main__': # 示例数据 sample_rows = [f'SN{i:03d}' for i in range(5)] # 模拟5个SN print("--- 开始单进程数据抓取 ---") start_total_time = time.time() results = single_process_pull(sample_rows) end_total_time = time.time() print(f"n单进程总耗时: {format(end_total_time - start_total_time, '.2f')} 秒") # print("结果:", results)
在上述代码中，pull_data函数模拟了对Web API的调用。由于每个API请求可能涉及复杂的后端查询或网络延迟，单次调用耗时约1.5秒。对于少量数据，这种方法尚可接受，但当数据量增大时，总耗时将线性增长，效率低下。
立即学习“Python免费学习笔记（深入）”；
多进程尝试及其性能瓶颈分析
为了提高效率，开发者可能会尝试使用Python的multiprocessing模块来并行处理任务。以下是一个常见的多进程实现方式，它尝试手动创建进程并使用Queue来收集结果。
import requestsimport timefrom multiprocessing import Process, Queuedef pull_data_with_queue(row_data, q): """ 在多进程中执行API请求，并将结果放入队列。 """ url_api = 'https://api.example.com/data' # 替换为实际API地址 post_json = {'query_param': row_data} try: x = requests.post(url_api, json=post_json, timeout=10) x.raise_for_status() q.put(x.json()) except requests.exceptions.RequestException as e: print(f"进程 {row_data} API请求失败: {e}") q.put(None) # 放入None表示失败def manual_multiprocess_pull(rows, processes_per_batch=3): """ 手动创建进程进行数据抓取。 """ all_results = [] q = Queue() # 在主进程中创建队列 print(f"--- 开始手动多进程数据抓取 (每批次 {processes_per_batch} 进程) ---") start_total_time = time.time() for i in range(0, len(rows), processes_per_batch): jobs = [] batch_start_time = time.time() # 为当前批次创建并启动进程 for j in range(processes_per_batch): if i + j < len(rows): row_data = rows[i + j] p = Process(target=pull_data_with_queue, args=(row_data, q)) jobs.append(p) p.start() # 等待所有进程完成 for proc in jobs: proc.join() # 从队列中收集结果 batch_results = [] while not q.empty(): result = q.get() if result: batch_results.append(result) all_results.extend(batch_results) batch_end_time = time.time() print(f"处理批次 {i//processes_per_batch + 1} 完成，耗时 {format(batch_end_time - batch_start_time, '.2f')} 秒") end_total_time = time.time() print(f"n手动多进程总耗时: {format(end_total_time - start_total_time, '.2f')} 秒") return all_resultsif __name__ == '__main__': sample_rows = [f'SN{i:03d}' for i in range(10)] # 模拟10个SN results_mp = manual_multiprocess_pull(sample_rows, processes_per_batch=3) # print("多进程结果:", results_mp)
然而，这种手动创建和管理进程的方式，在实际运行中常常会发现其性能甚至比单进程还要差。原因主要有以下几点：
进程创建开销 (Process Creation Overhead):
创建新进程是一个“昂贵”的操作，它涉及到操作系统级别的资源分配，如内存空间、文件句柄等。每次循环都创建新的Process对象并启动它们，会产生显著的启动时间开销。相比之下，线程的创建开销要小得多。
进程间通信 (IPC) 序列化/反序列化开销:
当子进程通过Queue将结果返回给主进程时，数据需要进行序列化（pickling）才能跨进程传输，主进程接收后还需要反序列化。如果API返回的JSON数据结构复杂或数据量大，这种序列化和反序列化的开销会非常高，成为性能瓶颈。
I/O密集型任务的特点:
Web API请求是典型的I/O密集型任务，即大部分时间都在等待网络响应，而不是执行CPU计算。Python的全局解释器锁（GIL）在I/O操作（如网络请求、文件读写）期间会被释放，这意味着多线程在I/O密集型任务中可以实现真正的并发，因为它们在等待I/O时不会互相阻塞。而多进程虽然绕过了GIL，但其固有的进程创建和IPC开销，对于I/O密集型任务来说往往得不偿失。
I/O密集型任务的高效并发策略
对于I/O密集型任务，我们应该优先考虑多线程或异步编程，如果必须使用多进程，也应采用更高效的multiprocessing.Pool。
1. 推荐方案一：多线程 (Multithreading)
由于Web API请求是I/O密集型的，多线程是Python中一个非常合适的并发模型。concurrent.futures.ThreadPoolExecutor提供了一个高级接口来管理线程池，简化了多线程编程。
关键优化点：
requests.Session的使用： 对于多次向同一域名发起请求，使用requests.Session可以复用底层的TCP连接，显著减少每次请求建立连接的开销，提高效率。
import requestsimport timefrom concurrent.futures import ThreadPoolExecutordef pull_data_with_session(row_data, session): """ 在多线程中使用requests.Session拉取数据。 """ url_api = 'https://api.example.com/data' post_json = {'query_param': row_data} try: # 使用传入的session对象 x = session.post(url_api, json=post_json, timeout=10) x.raise_for_status() return x.json() except requests.exceptions.RequestException as e: print(f"线程 {row_data} API请求失败: {e}") return Nonedef threaded_pull(rows, max_workers=5): """ 使用线程池拉取数据。 """ all_results = [] print(f"--- 开始多线程数据抓取 (最大 {max_workers} 线程) ---") start_total_time = time.time() # 为每个线程池任务创建一个session，或者在ThreadPoolExecutor外部创建一个session并传递给每个worker # 更推荐在worker函数内部创建或使用线程局部存储的session with requests.Session() as session: # 在主线程创建并传递给每个future，或在每个worker函数中创建 with ThreadPoolExecutor(max_workers=max_workers) as executor: # 使用partial函数传递session对象给pull_data_with_session # from functools import partial # func = partial(pull_data_with_session, session=session) # results_iterator = executor.map(func, rows) # 更简单的做法是让每个worker自己创建session，或使用requests默认的连接池 # 但为了连接复用，最好是每个线程一个session # 这里的示例简单起见，直接使用map，如果pull_data_with_session需要session，需要调整 # 最佳实践是在每个线程启动时创建一个Session对象，并存储在thread-local storage # 为了演示，我们假设pull_data_with_session可以处理session的创建或传递 # 这里为了简化，我们直接在map中传递rows，并假设pull_data_with_session内部处理session或requests自动管理连接 # 实际生产代码中，应确保每个线程的请求都通过一个Session对象。 # 一个更实际的Session管理方式是： futures = [executor.submit(pull_data_with_session, row, session) for row in rows] for i, future in enumerate(futures): result = future.result() # 阻塞直到任务完成 if result: all_results.append(result) print(f"处理第 {i+1} 条数据完成") # 可以根据需要打印进度 end_total_time = time.time() print(f"n多线程总耗时: {format(end_total_time - start_total_time, '.2f')} 秒") return all_resultsif __name__ == '__main__': sample_rows = [f'SN{i:03d}' for i in range(10)] results_threaded = threaded_pull(sample_rows, max_workers=5) # print("多线程结果:", results_threaded)
2. 推荐方案二：异步编程 (Asynchronous Programming)
对于需要处理大量并发连接的场景，异步编程（如asyncio配合aiohttp）是最高效的选择。它通过事件循环和协程（coroutine）实现单线程下的高并发，避免了线程/进程切换的开销。
import asyncioimport aiohttpimport timeasync def async_pull_data(row_data, session): """ 异步拉取数据。 """ url_api = 'https://api.example.com/data' post_json = {'query_param': row_data} try: async with session.post(url_api, json=post_json, timeout=10) as response: response.raise_for_status() return await response.json() except aiohttp.ClientError as e: print(f"异步请求 {row_data} 失败: {e}") return Noneasync def async_main(rows): """ 异步主函数，管理并发请求。 """ all_results = [] print("--- 开始异步数据抓取 ---") start_total_time = time.time() async with aiohttp.ClientSession() as session: # 创建一个异步会话 tasks = [async_pull_data(row, session) for row in rows] # 等待所有任务完成，并收集结果 results = await asyncio.gather(*tasks, return_exceptions=True) # return_exceptions=True 可以捕获单个任务的异常 for i, res in enumerate(results): if not isinstance(res, Exception) and res: all_results.append(res) print(f"处理第 {i+1} 条数据完成") end_total_time = time.time() print(f"n异步总耗时: {format(end_total_time - start_total_time, '.2f')} 秒") return all_resultsif __name__ == '__main__': sample_rows = [f'SN{i:03d}' for i in range(10)] # 运行异步主函数 asyncio.run(async_main(sample_rows))
3. 多进程池的优化使用 (multiprocessing.Pool)
如果确实需要使用多进程（例如，任务涉及一些CPU密集型计算，或者需要绕过某些库的GIL限制），那么应该使用multiprocessing.Pool而不是手动创建进程。Pool会创建一组固定的工作进程，并将任务分发给它们，从而摊销进程创建的开销。
import requestsimport timefrom multiprocessing import Pool# pull_data函数不需要Queue，因为Pool会通过返回值收集结果def pull_data_for_pool(row_data): """ 供multiprocessing.Pool使用的API拉取函数。 注意：requests.Session对象不能直接跨进程共享， 因此每个子进程需要自行创建或使用requests默认的连接池。 对于Pool，通常是让每个worker进程在首次调用时创建其自己的Session。 """ url_api = 'https://api.example.com/data' post_json = {'query_param': row_data} try: # 在每个进程中独立发起请求，requests会自动管理连接池 # 如果需要更精细的Session控制，可以考虑在worker初始化函数中创建Session x = requests.post(url_api
以上就是Python I/O密集型任务：多进程为何慢于单进程及高效并发策略的详细内容，更多请关注创想鸟其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 chuangxiangniao@163.com 举报，一经查实，本站将立刻删除。
发布者：程序猿，转转请注明出处：https://www.chuangxiangniao.com/p/1378842.html
微信扫一扫 
支付宝扫一扫 
