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

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

self task_name task_state task_fun

3 条回复 • 2021-01-21 14:47:44 +08:00

1

ysc3839

2021 年 1 月 20 日 via Android

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

2

Vegetable

2021 年 1 月 20 日

1

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

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())

3

linw1995

2021 年 1 月 21 日

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