在asyncio 中跳出正在执行的task

需求描述

代码在asyncio的框架中运行, 但是一旦一个task出现了长时间的堵塞,我们要跳过这个task(代码可能是用户输入的,例如用户编写的插件)

代码如下

(其中大部分代码出自官方的 asyncio 以及 signal)

重点的函数在于 asyncio.gather 详情查看上面的链接

import asyncio

import os

import signal

import time

async def factorial(name, number):

    f = 1

    for i in range(2, number + 1):

        print(f"Task {name}: Compute factorial({i})...")

        await asyncio.sleep(1)

        f *= i

    print(f"Task {name}: factorial({number}) = {f}")

async def test():

    for i in range(100):

        print("sleep--", i)

        time.sleep(1)

def handler(signum, frame):

    print('Signal handler called with signal', signum)

    raise OSError("Couldn't open device!")

signal.signal(signal.SIGTERM, handler)

print(os.getpid())

async def main():

    # Schedule three calls *concurrently*:

    await asyncio.gather(

        test(),

        factorial("A", 2),

        factorial("B", 3),

        factorial("C", 4),

        return_exceptions=True

    )

asyncio.run(main())

运行结果如下

18733

sleep-- 0

sleep-- 1

sleep-- 2

sleep-- 3

sleep-- 4

sleep-- 5

sleep-- 6

Signal handler called with signal 15

Task A: Compute factorial(2)...

Task B: Compute factorial(2)...

Task C: Compute factorial(2)...

Task A: factorial(2) = 2

Task B: Compute factorial(3)...

Task C: Compute factorial(3)...

Task B: factorial(3) = 6

Task C: Compute factorial(4)...

Task C: factorial(4) = 24

Process finished with exit code 0

这样就可以跳过被长时间堵塞的task,或者长时间执行的task了