本文深入探讨了Python中多进程处理网络API请求时可能出现的性能瓶颈，特别是当多进程方案反而比单进程更慢的原因。我们将分析进程创建、进程间通信（IPC）的开销，并提出针对I/O密集型任务的优化策略，包括使用`multiprocessing.Pool`、`requests.Session`，以及探讨多线程和异步编程作为更高效的替代方案。
在处理需要从Web API拉取数据的场景中，开发者常倾向于使用多进程来加速数据获取。然而，有时会发现多进程方案反而比单进程方案耗时更长，这通常是由于对并发模型选择和其底层开销理解不足所致。
创建新的操作系统进程是一项资源密集型操作。它涉及到内存分配、复制父进程的数据结构（在某些操作系统上，如Linux，会使用写时复制技术），以及为新进程设置执行环境。当每个任务都需要创建一个新进程时，这些重复的创建和销毁开销会迅速累积，吞噬掉并行处理带来的潜在收益。在初始的慢速多进程代码中，为每个API请求手动创建Process实例，正是导致性能下降的一个主要原因。
多进程之间无法直接共享内存，需要通过进程间通信（IPC）机制来交换数据，例如使用Queue。数据在发送方需要被序列化（如Python的pickle模块），通过管道或共享内存传输到接收方，再进行反序列化。如果传输的数据量大或结构复杂，序列化和反序列化的过程将带来显著的CPU和内存开销，这进一步降低了整体性能。原始多进程代码中使用Queue来收集每个进程的API响应，也增加了不必要的IPC负担。
立即学习“Python免费学习笔记（深入）”；
理解任务类型是选择正确并发模型的关键。
对于I/O密集型任务，多进程的优势并不明显，因为进程在等待I/O时，其他进程同样可能在等待I/O。而进程创建和IPC的开销，反而可能使得多进程的效率低于单进程。
针对I/O密集型API请求的场景，我们可以采用以下优化策略：
multiprocessing.Pool提供了一种更高级、更高效的方式来管理一组工作进程。它会在启动时创建固定数量的进程，并将任务分发给这些进程，从而摊销了进程创建的开销，避免了频繁的进程创建与销毁。
以下是使用multiprocessing.Pool优化API请求的示例代码：
from multiprocessing import Pool
import requests
import time
# 假设这是一个模拟的API请求函数
def pull_data(row_data):
"""
模拟从Web API拉取数据的函数。
row_data 可以是序列号、ID或其他请求参数。
"""
url_api = 'https://jsonplaceholder.typicode.com/posts/1' # 替换为实际的API地址
# 实际应用中，post_json应根据row_data构建
# 这里为了示例简化，假设row_data直接是post_json的一部分或用于构建它
# 或者直接使用GET请求，如果API支持
try:
# 模拟复杂的查询，可能需要1.5秒
# time.sleep(1.5)
# 实际的POST请求
post_json = {"data": row_data} # 假设row_data是字典或可序列化对象
x = requests.post(url_api, json=post_json) # 使用json参数发送JSON数据
x.raise_for_status() # 检查HTTP请求是否成功
return x.json()
except requests.exceptions.RequestException as e:
print(f"Error fetching data for {row_data}: {e}")
return None
def optimized_api_requests(rows_list, max_workers=5):
"""
使用multiprocessing.Pool并行处理API请求。
"""
start_time = time.time()
with Pool(processes=max_workers) as pool:
# pool.map 会将rows_list中的每个元素作为参数传递给pull_data函数
# 并并行执行，然后收集所有结果。
results = pool.map(pull_data, rows_list)
end_time = time.time()
print(f"\n所有请求完成，耗时: {format(end_time - start_time, '.2f')}秒")
return results
if __name__ == '__main__':
# 模拟一系列需要请求的数据行
sample_rows = [f"SN_{i}" for i in range(10)] # 假设有10个请求
print("开始使用 multiprocessing.Pool 进行API请求...")
api_results = optimized_api_requests(sample_rows, max_workers=3) # 设置3个工作进程
# 打印部分结果
for i, res in enumerate(api_results):
if res:
print(f"请求 {sample_rows[i]} 结果示例: {res['id'] if 'id' in res else res}")
else:
print(f"请求 {sample_rows[i]} 失败。")
if i >= 2: # 只打印前3个结果
break代码说明：
在进行大量HTTP请求时，每次请求都建立新的TCP连接（包括DNS解析、TCP三次握手、SSL/TLS握手等）会带来显著的开销。requests.Session对象可以跨多个请求重用TCP连接，从而减少这些重复的开销，尤其是在与同一个API服务器进行多次交互时效果显著。
为了在多进程/多线程环境中有效利用requests.Session，通常的做法是在每个工作进程或线程内部创建一个Session实例，而不是在全局共享一个。
import requests
import time
from multiprocessing import Pool, current_process
# 为每个进程创建一个Session
# 注意：requests.Session对象不是线程安全的，也不是进程安全的，
# 所以每个worker（进程或线程）都应该有自己的Session实例。
def pull_data_with_session(row_data):
"""
使用requests.Session从Web API拉取数据的函数。
每个进程（或线程）应该有自己的Session。
"""
# 在每个进程的执行上下文中创建Session
# 对于Pool，这个函数会在每个worker进程中被调用
session = requests.Session()
url_api = 'https://jsonplaceholder.typicode.com/posts/1'
try:
post_json = {"data": row_data, "process": current_process().name}
x = session.post(url_api, json=post_json)
x.raise_for_status()
return x.json()
except requests.exceptions.RequestException as e:
print(f"Error fetching data for {row_data} in {current_process().name}: {e}")
return None
finally:
session.close() # 显式关闭Session，释放资源
def optimized_api_requests_with_session(rows_list, max_workers=5):
start_time = time.time()
with Pool(processes=max_workers) as pool:
results = pool.map(pull_data_with_session, rows_list)
end_time = time.time()
print(f"\n所有请求（带Session）完成，耗时: {format(end_time - start_time, '.2f')}秒")
return results
if __name__ == '__main__':
sample_rows = [f"SN_{i}" for i in range(10)]
print("\n开始使用 multiprocessing.Pool 和 requests.Session 进行API请求...")
api_results_session = optimized_api_requests_with_session(sample_rows, max_workers=3)
for i, res in enumerate(api_results_session):
if res:
print(f"请求 {sample_rows[i]} 结果示例: {res.get('id')} (由 {res.get('process')} 处理)")
else:
print(f"请求 {sample_rows[i]} 失败。")
if i >= 2:
break对于I/O密集型任务，除了multiprocessing.Pool，更推荐使用以下两种并发模型：
多线程（Multithreading）： Python的GIL对CPU密集型任务有性能限制，但对I/O密集型任务影响较小。当一个线程在等待I/O时，GIL会被释放，允许其他线程执行。因此，多线程是处理I/O密集型任务的有效且开销较小的方案。concurrent.futures.ThreadPoolExecutor是Python标准库中用于管理线程池的强大工具。
from concurrent.futures import ThreadPoolExecutor
import requests
import time
def pull_data_threaded(row_data):
session = requests.Session() # 每个线程创建自己的Session
url_api = 'https://jsonplaceholder.typicode.com/posts/1'
try:
post_json = {"data": row_data, "thread": current_process().name} # 线程也可以获取当前进程名
x = session.post(url_api, json=post_json)
x.raise_for_status()
return x.json()
except requests.exceptions.RequestException as e:
print(f"Error fetching data for {row_data}: {e}")
return None
finally:
session.close()
def threaded_api_requests(rows_list, max_workers=10):
start_time = time.time()
with ThreadPoolExecutor(max_workers=max_workers) as executor:
# executor.map 与 pool.map 类似，但使用线程池
results = list(executor.map(pull_data_threaded, rows_list))
end_time = time.time()
print(f"\n所有请求（多线程）完成，耗时: {format(end_time - start_time, '.2f')}秒")
return results
if __name__ == '__main__':
sample_rows = [f"SN_{i}" for i in range(10)]
print("\n开始使用 ThreadPoolExecutor 进行API请求...")
api_results_threaded = threaded_api_requests(sample_rows, max_workers=5)
for i, res in enumerate(api_results_threaded):
if res:
print(f"请求 {sample_rows[i]} 结果示例: {res.get('id')}")
else:
print(f"请求 {sample_rows[i]} 失败。")
if i >= 2:
break异步编程（Asynchronous Programming）： 对于需要极高并发量的I/O密集型任务，asyncio配合异步HTTP客户端（如aiohttp）是最高效的解决方案。它通过事件循环（event loop）以非阻塞的方式处理I/O，避免了线程/进程切换的开销，可以同时管理成千上万个并发连接。
import asyncio
import aiohttp
import time
async def fetch_data_async(session, row_data):
"""
使用aiohttp异步从Web API拉取数据。
"""
url_api = 'https://jsonplaceholder.typicode.com/posts/1'
try:
post_json = {"data": row_data}
async with session.post(url_api, json=post_json) as response:
response.raise_for_status()
return await response.json()
except aiohttp.ClientError as e:
print(f"Error fetching data for {row_data}: {e}")
return None
async def async_api_requests(rows_list):
start_time = time.time()
async with aiohttp.ClientSession() as session: # 创建一个异步Session
tasks = [fetch_data_async(session, row) for row in rows_list]
results = await asyncio.gather(*tasks) # 并行执行所有任务
end_time = time.time()
print(f"\n所有请求（异步）完成，耗时: {format(end_time - start_time, '.2f')}秒")
return results
if __name__ == '__main__':
sample_rows = [f"SN_{i}" for i in range(10)]
print("\n开始使用 asyncio 和 aiohttp 进行API请求...")
api_results_async = asyncio.run(async_api_requests(sample_rows))
for i, res in enumerate(api_results_async):
if res:
print(f"请求 {sample_rows[i]} 结果示例: {res.get('id')}")
else:
print(f"请求 {sample_rows[i]} 失败。")
if i >= 2:
break在进行任何优化之前，始终建议对代码进行性能剖析（profiling）。使用Python内置的cProfile模块或第三方工具可以帮助识别代码中真正的性能瓶颈，例如：
通过数据分析，可以避免盲目优化，将精力集中在最能提升性能的区域。
客户端的并发策略只是影响整体性能的一部分。以下外部因素也可能显著影响API请求的耗时：
以上就是解析Python多进程API调用慢的原因及优化策略的详细内容，更多请关注php中文网其它相关文章！
每个人都需要一台速度更快、更稳定的 PC。随着时间的推移，垃圾文件、旧注册表数据和不必要的后台进程会占用资源并降低性能。幸运的是，许多工具可以让 Windows 保持平稳运行。
Copyright 2014-2025 https://www.php.cn/ All Rights Reserved | php.cn | 湘ICP备2023035733号
206
收藏
