如何优化Python程序等待第三方API响应时的资源占用？

你的核心问题是同步调用耗时极长的API，导致程序和对象长时间挂起，浪费系统资源。以下是几种可行的改进方案，优先推荐异步编程或后台线程的方式：

方案一：使用 asyncio 实现异步非阻塞调用 #

将同步的API调用改为异步模式，让程序在等待API响应期间可以处理其他任务，而非完全挂起。

代码示例 #

若第三方API仅提供同步客户端，可通过 asyncio.to_thread 将阻塞调用包装为异步操作：

import asyncio

class Circle():
    def __init__(self, id):
        self.id = id
        self.radius = None

async def fetch_radius(circle):
    # 将同步API调用放到线程池执行，避免阻塞事件循环
    radius = await asyncio.to_thread(fetch_from_api, circle.id)
    circle.radius = radius
    # 获取半径后执行后续处理逻辑
    await process_circle(circle)

async def process_circle(circle):
    if circle.radius:
        print(f"Circle {circle.id} 半径为 {circle.radius}")

async def main():
    circle = Circle(id=2)
    # 启动异步任务，主线程无需等待即可执行其他操作
    fetch_task = asyncio.create_task(fetch_radius(circle))
    # 模拟主线程处理其他任务
    await asyncio.sleep(1)
    # 可选：等待异步任务完成后再结束程序
    await fetch_task

if __name__ == "__main__":
    asyncio.run(main())

关键点 #

• 用 asyncio.create_task 包装API调用，主线程不会被阻塞
• 针对同步API，asyncio.to_thread 能将阻塞逻辑转移到后台线程，不影响事件循环
• 可同时启动多个异步任务，最大化利用系统资源

方案二：使用后台线程处理阻塞调用 #

如果不想引入异步编程的复杂度，可通过 threading 模块将耗时操作放到单独线程，主线程可继续运行或直接退出。

代码示例 #

import threading

class Circle():
    def __init__(self, id):
        self.id = id
        self.radius = None

def fetch_radius(circle):
    radius = fetch_from_api(circle.id)
    circle.radius = radius
    process_circle(circle)

def process_circle(circle):
    if circle.radius:
        print(f"Circle {circle.id} 半径为 {circle.radius}")

def main():
    circle = Circle(id=2)
    # 创建并启动后台线程
    thread = threading.Thread(target=fetch_radius, args=(circle,))
    thread.daemon = True  # 设置为守护线程，主线程退出时自动终止
    thread.start()
    
    # 主线程可继续执行其他任务，或调用 thread.join() 等待线程完成
    # thread.join()

if __name__ == "__main__":
    main()

关键点 #

• 后台线程独立执行API调用，主线程无需挂起
• 若需等待结果，调用 thread.join()；否则主线程可直接退出，线程在后台继续运行
• 注意线程安全：若多个线程操作同一Circle对象，需用 threading.Lock 避免竞争条件

方案三：分离任务为独立进程（无Pickle版） #

如果程序无需长时间运行，可将半径获取任务拆分为独立worker进程，用数据库或文件存储Circle状态，替代对象序列化：

1. 主程序创建Circle对象后，将id存入数据库/文件，随即退出
2. Worker进程定期扫描待处理任务，调用API获取半径后更新存储状态
3. 后续处理Circle对象时，从存储介质读取最新数据

简化示例（文件存储） #

import time

# 主程序：提交待处理任务
def main():
    circle_id = 2
    with open("pending_circles.txt", "a") as f:
        f.write(f"{circle_id}\n")

# Worker进程：处理耗时任务
def worker():
    while True:
        # 读取待处理任务
        with open("pending_circles.txt", "r") as f:
            pending_ids = [int(line.strip()) for line in f if line.strip()]
        
        if not pending_ids:
            time.sleep(3600)  # 每小时扫描一次
            continue
        
        # 处理每个任务
        for circle_id in pending_ids:
            radius = fetch_from_api(circle_id)
            with open("circle_results.txt", "a") as f:
                f.write(f"{circle_id},{radius}\n")
        
        # 清空待处理列表
        open("pending_circles.txt", "w").close()

关键点 #

• 无需保持主程序运行，避免资源占用
• 用文件/数据库存储状态，比Pickle更可靠、易维护

额外注意事项 #

• 原代码错误：main() 中调用 fetch_radius(circle.id)，但函数参数需传入Circle对象，应改为 fetch_radius(circle)
• 针对耗时数小时的任务，建议添加日志记录和重试机制，避免API调用失败导致任务丢失
• 异步/线程方案中，需确保依赖radius的逻辑在radius赋值完成后执行（可通过回调、事件通知实现）

内容的提问来源于stack exchange，提问作者Clej