本文深入探讨Python在I/O密集型Web API数据抓取中，多进程性能反而下降的常见问题。文章分析了手动创建进程和进程间通信（IPC）带来的高昂开销，并指出I/O密集型任务的特性。教程提供了多线程、异步编程及`multiprocessing.Pool`等优化方案，强调了选择合适并发模型和连接复用的关键性，旨在帮助开发者高效处理网络请求。
在处理需要从Web API抓取大量数据的场景时，开发者常常面临性能瓶颈。当单次API请求耗时较长（例如，由于后端查询复杂或网络延迟），自然会想到利用并发来加速。Python提供了多种并发机制，如多线程、多进程和异步编程。然而，在某些情况下，盲目地使用多进程可能会导致性能不升反降，甚至比单进程串行执行还要慢。本教程将深入分析这一现象，并提供针对I/O密集型任务的优化策略。
首先，我们来看一个典型的单进程串行数据抓取实现。这种方法简单直观，但效率较低，因为它必须等待每个API请求完成后才能发起下一个请求。
import requests
import time
def 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 None
def 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_results
if __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 requests
import time
from multiprocessing import Process, Queue
def 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_results
if __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):
进程间通信 (IPC) 序列化/反序列化开销:
I/O密集型任务的特点:
对于I/O密集型任务，我们应该优先考虑多线程或异步编程，如果必须使用多进程，也应采用更高效的multiprocessing.Pool。
由于Web API请求是I/O密集型的，多线程是Python中一个非常合适的并发模型。concurrent.futures.ThreadPoolExecutor提供了一个高级接口来管理线程池，简化了多线程编程。
关键优化点：
import requests
import time
from concurrent.futures import ThreadPoolExecutor
def 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 None
def 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_results
if __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)对于需要处理大量并发连接的场景，异步编程（如asyncio配合aiohttp）是最高效的选择。它通过事件循环和协程（coroutine）实现单线程下的高并发，避免了线程/进程切换的开销。
import asyncio
import aiohttp
import time
async 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 None
async 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_results
if __name__ == '__main__':
sample_rows = [f'SN{i:03d}' for i in range(10)]
# 运行异步主函数
asyncio.run(async_main(sample_rows))如果确实需要使用多进程（例如，任务涉及一些CPU密集型计算，或者需要绕过某些库的GIL限制），那么应该使用multiprocessing.Pool而不是手动创建进程。Pool会创建一组固定的工作进程，并将任务分发给它们，从而摊销进程创建的开销。
import requests
import time
from 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密集型任务：多进程为何慢于单进程及高效并发策略的详细内容，更多请关注php中文网其它相关文章！
每个人都需要一台速度更快、更稳定的 PC。随着时间的推移，垃圾文件、旧注册表数据和不必要的后台进程会占用资源并降低性能。幸运的是，许多工具可以让 Windows 保持平稳运行。
Copyright 2014-2025 https://www.php.cn/ All Rights Reserved | php.cn | 湘ICP备2023035733号
111
收藏
