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

[fangwenxue]

    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
• 有没有更好的实现？

[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 等方式，把取消事件传递给副线程的函数