在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了