Python 3.10下如何强制终止运行超20秒的异步函数

问题分析 #

你当前的代码用asyncio.wait_for设置了20秒超时但未生效，核心原因是Broker.historical_data是同步阻塞函数——它会持续占用asyncio的事件循环，导致事件循环无法处理超时检测的回调，自然触发不了超时终止逻辑。

方案一：用线程池包装同步API（推荐，轻量） #

Python 3.9+提供的asyncio.to_thread可以将同步函数放到线程池执行，不会阻塞事件循环，这样asyncio.wait_for的超时机制就能正常生效。

修改后的FMinData函数：

import asyncio
import datetime
import os
import sys

async def FMinData():
    print(datetime.datetime.now())
    try:        
        for strike in strikelistlimit: 
            try:
                # 将同步API调用委托给线程池执行
                data = await asyncio.to_thread(
                    Broker.historical_data,
                    strike, lastBusDay, datetime.datetime.now(), 'minute', oi=False
                )
                # 数据处理逻辑
            except Exception:
                os._exit(0)                        
        print(datetime.datetime.now())
    except asyncio.TimeoutError:        
        os._exit(0)
    except Exception:        
        os._exit(0)
    finally:        
        os._exit(0)

调用部分保持原有逻辑即可：

if str(sys.argv[1]) == "1": 
    # 其他函数调用
    try:        
        asyncio.run(asyncio.wait_for(FMinData(), timeout=20))
    except asyncio.TimeoutError:
        os._exit(0)
    os._exit(0)

方案二：多进程强制终止（应对完全无响应的场景） #

如果线程池方案仍无法终止程序（比如同步API调用完全卡住、不响应中断），可以用多进程实现——Windows下线程无法强制终止，但进程可以直接杀死。

实现思路：把FMinData的逻辑放到子进程执行，主进程监控子进程运行时长，超过20秒就强制杀死子进程。

示例代码：

import datetime
import os
import sys
import multiprocessing
import time

def FMinData_process():
    print(datetime.datetime.now())
    try:        
        for strike in strikelistlimit: 
            try:
                data = Broker.historical_data(
                    strike, lastBusDay, datetime.datetime.now(), 'minute', oi=False
                )
                # 数据处理逻辑
            except Exception:
                os._exit(0)                        
        print(datetime.datetime.now())
    except Exception:        
        os._exit(0)
    finally:        
        os._exit(0)

if __name__ == "__main__":
    if str(sys.argv[1]) == "1": 
        # 其他函数调用
        process = multiprocessing.Process(target=FMinData_process)
        process.start()
        # 等待20秒
        process.join(timeout=20)
        # 若进程仍在运行，强制杀死
        if process.is_alive():
            process.kill()
            os._exit(0)
        os._exit(0)

注意事项 #

• 方案一中，asyncio.to_thread将同步任务放到后台线程，事件循环可正常处理超时回调，超时触发时会抛出asyncio.TimeoutError，捕获后即可终止程序。
• 方案二中，多进程的缺点是进程间数据共享需要额外处理（比如用队列），如果你的数据处理涉及大量数据传递，需注意这一点。
• 若非必要，尽量用sys.exit()替代os._exit(0)终止程序，后者会直接终止进程，不执行清理逻辑。

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