如何实现第三方API调用函数超时5秒或失败时自动重试？

我通过第三方官方原生库调用其API的函数，正常情况下响应速度良好，仅需1-2秒，但偶尔该函数需要30-50秒才能获取响应。我希望实现这样的逻辑：若函数执行时间超过5秒，则取消调用并重试该函数。

我尝试过tenacity与concurrent futures的多种组合，但均未达到预期效果。

注意：第三方服务响应缓慢时不会显示为挂起状态，因此需要一种能等待5秒后终止调用并重试的方案。

尝试的代码 #

@retry(wait=wait_random_exponential(min=3, max=5), stop=stop_after_attempt(3), retry_error_callback=lambda x: logger.info("Retrying getting vectors..."))
def get_vectors(namespace):
    def query():
        ####SOME CODE for 3rd party API###
        return some_output

    with concurrent.futures.ThreadPoolExecutor() as executor:
        future = executor.submit(query)
        try:
            return future.result(timeout=5)
        except concurrent.futures.TimeoutError:
            logger.info('5 secs timeout happened...')
            future.cancel()
            raise Exception("Function took too long to complete. Retrying...")   

解决方案 #

核心问题分析 #

你写的代码里，future.cancel()只能取消还未开始执行的任务。一旦线程已经进入第三方API的阻塞调用状态，cancel()完全无法终止这个慢请求，它会在后台继续运行，同时重试逻辑又发起新的请求，反而造成资源浪费。

正确实现方案 #

由于Python线程无法被强制终止，改用multiprocessing（多进程）来实现超时终止逻辑，再结合tenacity完成重试。进程可以被强制杀死，能确保超时后彻底停止慢请求。

最终代码 #

from tenacity import retry, wait_random_exponential, stop_after_attempt
import multiprocessing
import logging

logger = logging.getLogger(__name__)

def query_api(namespace, result_queue):
    """在子进程中执行第三方API调用，将结果存入队列"""
    try:
        # 替换为实际的第三方API调用代码
        ####SOME CODE for 3rd party API###
        api_result = some_output
        result_queue.put(('success', api_result))
    except Exception as e:
        result_queue.put(('error', e))

@retry(
    wait=wait_random_exponential(min=3, max=5),
    stop=stop_after_attempt(3),
    retry_error_callback=lambda _: logger.info("重试获取向量...")
)
def get_vectors(namespace):
    result_queue = multiprocessing.Queue()
    # 创建子进程执行API调用
    api_process = multiprocessing.Process(target=query_api, args=(namespace, result_queue))
    api_process.start()
    
    # 等待5秒，超时则终止进程
    api_process.join(timeout=5)
    
    if api_process.is_alive():
        # 强制终止仍在运行的进程
        api_process.terminate()
        api_process.join()  # 等待进程彻底退出
        logger.info("API调用超时5秒，已终止进程并重试")
        raise Exception("API调用超时，触发重试")
    
    # 获取子进程返回的结果
    status, data = result_queue.get()
    if status == 'success':
        return data
    else:
        # 抛出API调用中的异常，触发tenacity重试
        raise data

关键说明 #

• 用multiprocessing.Process替代线程：进程可以被terminate()强制终止，确保超时后彻底停止慢请求，避免资源泄漏。
• 通过Queue传递子进程的执行结果：多进程间无法直接共享变量，队列是安全的通信方式。
• tenacity重试逻辑：当超时异常或API调用本身的异常抛出时，自动触发重试，最多重试3次，每次间隔3-5秒的随机指数退避时间。

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