在Spyder IDE中使用asyncio.create_task获取结果遇问题求助

我正在Spyder IDE中学习Python异步调用的运行机制，编写了如下代码：

import asyncio
from concurrent.futures import ThreadPoolExecutor

async def factorial(number):
    f = 1
    for i in range(2, number + 1):
        await asyncio.sleep(1)
        f *= i
    print(f"Task {number}: factorial({number}) = {f}")
    return f

async def main():
    # Schedule three calls *concurrently*:
    executor = ThreadPoolExecutor(max_workers=10)
    loop = asyncio.get_event_loop()
    L =  [await loop.run_in_executor(executor, factorial, i) for i in range(2,5)]
    L = await asyncio.gather(*L)
        

    print(L)
    return(L)

if __name__ == '__main__':
    a = asyncio.create_task(main())
    if a.done():
        res = a.result()
        print("RESULT:",res)
    print("AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA")

Spyder控制台输出如下：

In[1] :
AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA
Task 2: factorial(2) = 2
Task 3: factorial(3) = 6
Task 4: factorial(4) = 24
[2, 6, 24]

可见print("AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA")在asyncio.create_task的结果输出前执行，且print("RESULT:",res)的内容并未显示。

问题原因分析 #

1. 异步任务未被等待：asyncio.create_task(main())仅创建异步任务并将其加入事件循环，但主线程不会主动等待任务完成，会直接执行后续代码，所以那段长串"A"的输出会先出现。
2. 任务状态判断时机错误：创建任务后立刻调用a.done()，此时任务刚被调度，还未开始执行，必然返回False，因此print("RESULT:",res)永远不会执行。
3. 线程池使用错误：factorial是异步函数，loop.run_in_executor会把它放到线程池执行，但线程中没有运行asyncio事件循环，异步函数无法按预期执行（实际能运行是Spyder的环境特性，属于错误用法）。异步任务应该直接交给asyncio事件调度，线程池适合处理阻塞的同步函数。

修正后的代码 #

方案1：使用asyncio.run（Python3.7+推荐）

import asyncio

async def factorial(number):
    f = 1
    for i in range(2, number + 1):
        await asyncio.sleep(1)
        f *= i
    print(f"Task {number}: factorial({number}) = {f}")
    return f

async def main():
    # 直接用asyncio.gather调度异步任务
    tasks = [factorial(i) for i in range(2,5)]
    L = await asyncio.gather(*tasks)
    print(L)
    return L

if __name__ == '__main__':
    res = asyncio.run(main())
    print("RESULT:", res)
    print("AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA")

方案2：保留create_task并等待任务完成

import asyncio

async def factorial(number):
    f = 1
    for i in range(2, number + 1):
        await asyncio.sleep(1)
        f *= i
    print(f"Task {number}: factorial({number}) = {f}")
    return f

async def main():
    tasks = [factorial(i) for i in range(2,5)]
    L = await asyncio.gather(*tasks)
    print(L)
    return L

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    a = loop.create_task(main())
    # 等待任务完成
    res = loop.run_until_complete(a)
    print("RESULT:", res)
    print("AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA")

修正说明 #

• 移除了不必要的ThreadPoolExecutor，改用asyncio.gather直接调度异步任务，符合asyncio的使用规范。
• 使用asyncio.run或loop.run_until_complete等待异步任务执行完毕，确保主线程不会提前结束，且能正确获取任务返回值。
• 修正后输出顺序会变为：任务执行输出→[2,6,24]→RESULT: [2,6,24]→长串"A"，符合预期。

内容的提问来源于stack exchange，提问作者Leonardo Lima Ferreira