如何使函数异步运行？

Posted 2023-03-30

【中文标题】如何使函数异步运行？

【英文标题】：How to make a function behave asynchronously?

【发布时间】：2021-02-21 07:05:26

【问题描述】：

我最近在 Python 中使用了异步函数，我想知道如何将同步函数变成异步函数。

例如，有通过 google api pygoogletranslation 进行翻译的库。人们很可能想知道，如何异步翻译许多不同的单词。当然，你可以把它放在一个请求中，但是google api会认为它是一个文本并进行相应的处理，这会导致错误的结果。

这个代码怎么可能转：

from pygoogletranslation import Translator
translator = Translator()
translations = []
words = ['partying', 'sightseeing', 'sleeping', 'catering']
for word in words:
    translations.append(translator.translate(word, src='en', dest='es'))
print(translations)

进入这个：

from pygoogletranslation import Translator
import asyncio
translator = Translator()
translation_tasks = []
words = ['partying', 'sightseeing', 'sleeping', 'catering']
for word in words:
    asyncio.create_task(translator.translate(word, src='en', dest='es'))
translations = asyncio.run(
    asyncio.gather(translation_tasks, return_exceptions=True) 
)
print(translations)

考虑到函数translate 没有内置的async 实现？

【问题讨论】：

我的 javascript 大脑搜索 Promise.all 等价物：***.com/questions/34377319/…。基本上将每个对谷歌的请求包装在一个异步函数中，并在异步函数数组完成后使用结果。

改用多线程或多处理

@MattWilde 如答案中所述，asyncio.gather() 与Promise.all 最接近。关键区别在于其他地方：在 JavaScript 中，所有可能阻塞的东西默认都是异步的，而在 Python 中不是这种情况。

【参考方案1】：

您必须创建一个async 函数然后运行它。尽管如果 translate 没有内置 async 支持或被阻塞，使用 async 不会使其更快。按照 cmets 的建议，使用多线程/多处理可能会更好。

async def main():
    async def one_iteration(word):
        output.append(translator.translate(word, src='en', dest='es'))
    coros = [one_iteration(word) for word in words]
    await asyncio.gather(*coros)
asyncio.run(main())

【讨论】：

对不起，事实上我给出的例子只是为了展示我想对任何类型的函数做什么。我只是想知道，如何实现一个异步函数，它是如何在内部工作的，以及为什么一个函数可以异步而另一个不能。

异步函数需要用async def定义，它本质上是一个非阻塞函数，可以让其他函数在等待结果的同时运行。因此，如果您在异步函数中放置一些阻塞调用，它仍然会阻止其他函数运行。一个简单的例子是在异步函数中你想使用async.sleep 而不是time.sleep，因为异步版本是非阻塞的。有关详细信息，请参阅 python 文档或realpython.com/async-io-python。

【参考方案2】：

正如其他答案中提到的，调用阻塞函数对 ayncio 毫无用处。在这种特殊情况下，我建议您使用google-cloud-translate，这是 Google 的官方翻译库。

你可以在你当前的库中做这样的事情：

async def do_task(word):
    return translator.translate(word, ...)

def main():
    # Create translator
    ...
    asyncio.gather(do_task(word) for word in [])

但这只会在没有 asyncio 的情况下以相同的方式运行任务。 asyncio 的真正好处是，当有事情挂起或等待时，它可以做其他事情。例如，在等待服务器响应时，它可以发送另一个请求。

Python 如何知道某些工作正在等待处理？仅当函数（此处为协程）通过 await 关键字通知事件循环时。所以你肯定需要使用原生支持异步操作的库。上面提到的google-cloud-translate就是这样一个库。你可以这样做：

from google.cloud import translate


async def main():
    # Async-supported google translator client
    client = translate.TranslationServiceAsyncClient()
    words = ['partying', 'sightseeing', 'sleeping', 'catering']
    results = await asyncio.gather(*[client.translate_text(parent=f"projects/project_name", contents=[word], source_language_code="en", target_language_code="es") for word in words])
    print(results)

asyncio.run(main())

您可以看到这个客户端实际上将字符串列表作为输入，因此您可以在此处直接传递字符串列表。根据docs，这个限制是1024。所以如果你的列表更大，你必须使用这个for循环。

您可能需要为此客户端设置凭据等，这超出了本问题的范围。

【参考方案3】：

要使函数异步，您需要使用async def 对其进行定义，并将其更改为使用其他异步函数来处理任何可能阻塞的情况——例如，您将使用aiohttp 代替requests，等等在。努力的重点是该函数可以与其他此类函数一起由事件循环执行。每当异步函数需要等待某事时，正如await 关键字所指示的，它会挂起到事件循环并给其他人执行的机会。事件循环将无缝协调可能大量此类异步函数的并发执行。参见例如this answer 了解更多详情。

如果您依赖的关键阻塞函数没有异步实现，您可以使用run_in_executor（或者，从 Python 3.9 开始，asyncio.to_thread）使其异步。但是请注意，此类解决方案是“作弊”，因为它们在后台使用线程，因此它们不会提供通常与 asyncio 相关的好处，例如扩展线程池中线程数量的能力，或取消执行的能力协程。