Python Gil 任务优化

Question

我在 Python 中开发了一个简单的脚本，使用 asyncio 和 tasks 同时对多个目标执行一些随机工作。我有几个关于并发和 Python 解释器对任务管理的优化的问题，因为我对并发编程有点陌生。

我的主要问题是，Python 在什么时候决定同时运行任务？这是 Python 解释器此时进行的优化吗？或者我应该使用 threads 或 futures 显式编写根据代码？

非常感谢任何指向有趣的帖子、经验或文档的指针。

我附上了几个代码 sn-ps 用于比较它们：

    # A Python3 example that uses tasks for concurrency
 
    import asyncio
    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 main():
    
        print(f"started main at {time.strftime('%X')}")
    
        # Schedule three calls *concurrently*:
        await asyncio.gather(
            factorial("A", 2),
            factorial("B", 3),
            factorial("C", 4),
        )
    
        print(f"started main at {time.strftime('%X')}")
    
    asyncio.run(main())

    # A second example using threads.
    
    import asyncio
    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 main():
        print(f"started main at {time.strftime('%X')}")
    
        try: 
            await asyncio.gather(
                asyncio.to_thread(await factorial("A", 2)),
                asyncio.to_thread(await factorial("B", 3)),
                asyncio.to_thread(await factorial("C", 4)),
            )
        except:
            print("Main routine finished.")
    
        print(f"started main at {time.strftime('%X')}")
    
    
    asyncio.run(main())

谢谢！

Answer 1

Python 在什么时候决定并发运行任务？

我认为存在一些术语混乱。

我怀疑你的意思是：

Python 什么时候决定在单独的线程中运行任务？

从来没有。 Python 不会隐式地将 asyncio 任务作为单独的线程运行。

asyncio 代码通常是在假设单线程的情况下编写的。在工作线程中隐式运行该代码会破坏事情。

如果您想在单独的线程或进程中完成大量工作，请使用concurrent.futures 中的ThreadPoolExecutor 或ProcessPoolExecutor。

如果你想在一个单独的线程或进程中做一堆工作，并且你恰好在一个协程中，使用类似asyncio.loop.run_in_executor()的东西。

Answer 2

首先，您应该知道 asyncio 是单线程的，并且设计用于并行化 IO 密集型任务，而不是 CPU 密集型任务。如果您需要 CPU 密集型任务的并行性，您应该使用线程，或者更好的是进程。

其次，您的第二个示例完全不正确 - 如果您使用线程，factorial 应该是一个普通函数，而不是一个异步函数。您的gather() 表达式相当于：

result1 = await factorial("A", 2)
result2 = await factorial("B", 3)
result3 = await factorial("C", 4)
await asyncio.gather(
    asyncio.to_thread(result1),
    asyncio.to_thread(result2),
    asyncio.to_thread(result3),
)

这样改写，显然是错误的。代码正在按顺序而不是按预期并行等待阶乘，并且还因为它正在调用 to_thread，返回值为 factorial，这是 None 因为 factorial 不返回任何内容。 to_thread 接受一个 函数（以及可选的参数），因此定义它的正确方法是：

def factorial(name, number):  # note: def, not async def
    f = 1
    for i in range(2, number + 1):
        time.sleep(1)  # note: time.sleep(), not asyncio.sleep()
        f *= i

# ...

# note: don't call `factorial` here, just pass it to `to_thread`
await asyncio.gather(
    asyncio.to_thread(factorial, "A", 2)),
    asyncio.to_thread(factorial, "B", 3)),
    asyncio.to_thread(factorial, "C", 4)),
)

如果您没有使用 try: ... excpet: ...（这是一种反模式）抑制所有异常，您会注意到“NoneType is not callable”异常。

或者我应该使用线程或期货显式编写根据代码？

是的，您应该使用concurrent.futures 模块来编写您的代码。参见例如here 回顾各种库之间的差异。