将小函数转换为协程

Question

我觉得我对异步 IO 的理解存在差距：在较大的协程范围内将小函数包装到协程中是否有好处？ 正确地发出事件循环信号有好处吗？这种好处的程度是否取决于包装的函数是 IO 还是 CPU 密集型？

示例：我有一个协程，download()其中：

1. 通过 . 从 HTTP 端点下载 JSON 序列化字节aiohttp。
2. 通过 - 压缩这些字节bz2.compress()- 这本身不是可等待的
3. 通过以下方式将压缩字节写入 S3aioboto3

因此，第 1 部分和第 3 部分使用这些库中的预定义协程；默认情况下，第 2 部分没有。

简化的例子：

import bz2
import io
import aiohttp
import aioboto3

async def download(endpoint, bucket_name, key):
    async with aiohttp.ClientSession() as session:
        async with session.request("GET", endpoint, raise_for_status=True) as resp:
            raw = await resp.read()  # payload (bytes)
            # Yikes - isn't it bad to throw a synchronous call into the middle
            # of a coroutine?
            comp = bz2.compress(raw)
            async with (
                aioboto3.session.Session()
                .resource('s3')
                .Bucket(bucket_name)
            ) as bucket:
                await bucket.upload_fileobj(io.BytesIO(comp), key)

正如上面的评论所暗示的，我的理解一直是，将同步函数（如协程）扔进bz2.compress()协程中可能会弄乱它。（即使可能bz2.compress()更多的是 IO 限制而不是 CPU 限制。）

那么，这种类型的样板通常有什么好处吗？

async def compress(*args, **kwargs):
    return bz2.compress(*args, **kwargs)

（现在comp = await compress(raw)在 内download()。）

哇啦，这现在是一个可等待的协程，因为鞋底return在本机协程中是有效的。是否有使用此功能的案例？

根据这个答案，我听说过以类似方式随机抛出的理由asyncio.sleep(0)- 只是为了单备份到调用协程想要中断的事件循环。这是正确的吗？

Answer 1

那么，这种类型的样板通常有什么好处吗？

async def compress(*args, **kwargs):
    return bz2.compress(*args, **kwargs)

这样做没有任何好处。与预期相反，添加 anawait 并不能保证控制将传递到事件循环 - 仅当等待的协程实际挂起时才会发生这种情况。由于compress不等待任何内容，因此它永远不会挂起，因此它只是名义上的协程。

请注意，添加await asyncio.sleep(0)协程并不能解决问题；请参阅此答案以获取更详细的讨论。如果您需要运行阻塞函数，请使用run_in_executor：

async def compress(*args, **kwargs):
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(None, lambda: bz2.compress(*args, **kwargs))