Python中如何用asyncio获取当前时间?异步代码异常求助
异步代码故障排查与解决办法
你没贴出故障代码,但异步代码无法运行基本逃不过以下几种常见问题,对应解决方式如下:
调用异步函数未加
await
直接调用异步函数只会返回一个协程对象,不会实际执行逻辑。
错误示例:async def bad_call(): get_hour() # 缺少await,函数不执行修正:
async def good_call(): await get_hour()未正确启动事件循环
所有异步任务必须在事件循环中运行,没有调用启动方法的话代码不会执行。
错误示例:async def no_loop(): await get_hour() no_loop() # 直接调用,未启动事件循环修正:
asyncio.run(no_loop())同步阻塞代码卡死事件循环
异步环境中如果存在长时间运行的同步代码(比如time.sleep()、密集计算),会阻塞整个事件循环,导致所有异步任务挂起。
错误示例:import time async def blocked_loop(): time.sleep(60) # 同步sleep,事件循环被锁死 await get_hour()修正方案:
# 方案1:用异步sleep替代同步sleep async def fixed_loop1(): await asyncio.sleep(60) await get_hour() # 方案2:将同步代码放到线程池执行 async def fixed_loop2(): await asyncio.get_event_loop().run_in_executor(None, time.sleep, 60) await get_hour()异步任务未被等待
用asyncio.create_task()创建任务后,如果不await或者不将任务关联到事件循环,任务不会执行。
错误示例:async def unawaited_task(): asyncio.create_task(get_hour()) # 创建任务但未等待执行修正:
async def awaited_task(): task = asyncio.create_task(get_hour()) result = await task异常未被捕获导致静默失败
异步任务中的异常如果未被捕获,可能会导致任务悄悄终止,看起来像是代码没运行。
解决方式:添加异常捕获逻辑,或在批量任务中启用异常返回:async def catch_error(): try: await faulty_async_func() except Exception as e: print(f"执行出错:{e}") # 批量任务时查看异常 async def gather_tasks(): results = await asyncio.gather( task1(), task2(), return_exceptions=True # 捕获异常并返回,避免整个gather失败 ) print(results)
内容的提问来源于stack exchange,提问作者SapeNeCo
相关产品推荐
相关产品推荐

