python - 在长时间运行的代码中使用 asyncio.sleep() 将异步函数划分为多个较小的代码部分是否合适？

Question

如果我有一些函数需要进行大量计算，并且可能需要一段时间，那么在计算部分之间使用 asyncio.sleep() 来释放事件循环是否合适(以防止阻塞事件循环)？

import asyncio


async def long_function(a, b, c):
    # some calculations
    await asyncio.sleep(0)  # release event loop
    # some another calculations
    await asyncio.sleep(0)  # release event loop

还有其他更好的方法来解决此类问题吗？也许是一些最佳实践？

Answer 1

TL;DR 只需使用loop.run_in_executor 来执行阻塞工作

<小时/>

要理解为什么它没有帮助，让我们首先创建一个使用事件循环执行某些操作的类。喜欢:

class CounterTask(object):
    def __init__(self):
        self.total = 0
    
    async def count(self):
        while True:
            try:
                self.total += 1
                await asyncio.sleep(0.001)  # Count ~1000 times a second
            except asyncio.CancelledError:
                return

如果事件循环完全开放的话，每秒只会计数大约 1000 次。

天真

为了演示最糟糕的方法，让我们开始计数器任务并天真地运行一个昂贵的函数，而不考虑后果:

async def long_function1():
    time.sleep(0.2)  # some calculations


async def no_awaiting():
    counter = CounterTask()
    task = asyncio.create_task(counter.count())
    await long_function1()
    task.cancel()
    print("Counted to:", counter.total)


asyncio.run(no_awaiting())

输出:

Counted to: 0

好吧，这没有进行任何计数!请注意，我们根本没有等待。这个函数只是做同步阻塞的工作。如果计数器能够在事件循环中自行运行，我们当时应该已经数到了 200 左右。嗯，那么也许如果我们将其拆分并利用 asyncio 将控制权交还给事件循环，它就可以计数了？让我们尝试一下...

拆分

async def long_function2():
    time.sleep(0.1)  # some calculations
    await asyncio.sleep(0)  # release event loop
    time.sleep(0.1)  # some another calculations
    await asyncio.sleep(0)  # release event loop


async def with_awaiting():
    counter = CounterTask()
    task = asyncio.create_task(counter.count())
    await long_function2()
    task.cancel()
    print("Counted to:", counter.total)


asyncio.run(with_awaiting())

输出:

Counted to: 1

嗯，我想这在技术上更好。但最终这表明了这一点:asyncio 事件循环不应该执行任何阻塞处理。它并不是为了解决这些问题。事件循环正在无助地等待您的下一个 await。但 run_in_executor 确实为此提供了解决方案，同时使我们的代码保持 asyncio 风格。

执行者

def long_function3():
    time.sleep(0.2)  # some calculations


async def in_executor():
    counter = CounterTask()
    task = asyncio.create_task(counter.count())
    await asyncio.get_running_loop().run_in_executor(None, long_function3)
    task.cancel()
    print("Counted to:", counter.total)


asyncio.run(in_executor())

输出:

Counted to: 164

好多了!当我们的阻塞函数也通过良好的老式线程方式执行操作时，我们的循环能够继续进行。