Python 异步任务如何检测运行时间

資深大佬 : fangwenxue 4

    async def run_task(self, task_name, task_fun):         self.logger.info(f'{task_name}')         begin_time = int(time.time())         self.task_state = False         # t = threading.Thread(target=self.check_task, args=(begin_time, task_name))         # t.start()         # t.join()             try:             await getattr(self, task_fun)()             await asyncio.sleep(1)             self.logger.warning(f'{task_name} -> DONE.')             self.task_state = True         except Exception as e:             self.logger.error(e)             self.task_state = True         finally:             await self.close_page()             await asyncio.sleep(1)      def check_task(self, begin_time, task_name):         while True:             if self.task_state:                 return True              now_time = int(time.time())             n = now_time - begin_time             if n > 120:                 self.logger.warning(f'{task_name} runtime {n}s.')                 return False             time.sleep(1)

想要实现的是 task_fun 执行时间超过 N 秒直接退出 run_task
自己想了笨方法，写了个线程检测运行时间，但不知道怎么抛异常退出 task_fun
有没有更好的实现？

大佬有話說 (3)

資深大佬 : ysc3839

我不了解 Python 的 async，如果是 js 的话可以用 Promise.race，然后取消 task 运行。

資深大佬 : Vegetable

想得倒是挺多，不过你没感觉这是一个很常见的需求吗？

https://docs.python.org/zh-cn/3/library/asyncio-task.html#timeouts

async def main():
# Wait for at most 1 second
try:
await asyncio.wait_for(eternity(), timeout=1.0)
except asyncio.TimeoutError:
print(‘timeout!’)

asyncio.run(main())

資深大佬 : linw1995

如果不是同步堵塞住的话，用 `asyncio.wait_for` 就好了。同步堵塞的话，建议修改这个函数，通过 `loop.run_in_executor` 用副线程去跑，这样就不会堵塞住了。同时在外部用 `asyncio.wait_for`，且通过在 coro 内部去 try-except 这个超时而抛出的 `asyncio.CancelError` ，再通过 theading.Event 等方式，把取消事件传递给副线程的函数