Python 3.5 async/await 与真实代码示例

Posted 2023-03-07

【中文标题】Python 3.5 async/await 与真实代码示例

【英文标题】：Python 3.5 async/await with real code example

【发布时间】：2017-02-18 13:14:43

【问题描述】：

我已经阅读了大量关于 Python 3.5 async/await 的文章和教程。我不得不说我很困惑，因为有些使用 get_event_loop() 和 run_until_complete()，有些使用 ensure_future()，有些使用 asyncio.wait()，有些使用 call_soon()。

似乎我有很多选择，但我不知道它们是否完全相同，或者在某些情况下您使用循环并且在某些情况下您使用 wait()。

但问题是所有示例都与asyncio.sleep() 一起工作，模拟真正的慢速操作，返回一个等待对象。一旦我尝试将这一行换成一些真实的代码，整个事情就会失败。上面写的方法和我应该如何运行还没有准备好异步/等待的第三方库之间的区别到底是什么。我确实使用 Quandl 服务来获取一些股票数据。

 import asyncio
 import quandl

 async def slow_operation(n):
     # await asyncio.sleep(1) # Works because it's await ready.
     await quandl.Dataset(n) # Doesn't work because it's not await ready.


 async def main():
     await asyncio.wait([
         slow_operation("SIX/US9884981013EUR4"),
         slow_operation("SIX/US88160R1014EUR4"),
     ])

 # You don't have to use any code for 50 requests/day.
 quandl.ApiConfig.api_key = "MY_SECRET_CODE"

 loop = asyncio.get_event_loop()
 loop.run_until_complete(main())

我希望你明白我的失落感以及我希望并行运行的事情是多么简单。

【参考方案1】：

如果第三方库与async/await 不兼容，那么显然你不能轻易使用它。有两种情况：

假设库中的函数是异步的，它会给你一个回调，例如

def fn(..., clb):
    ...

所以你可以这样做：

def on_result(...):
    ...

fn(..., on_result)

在这种情况下，您可以像这样将这些函数包装到 asyncio 协议中：

from asyncio import Future

def wrapper(...):
    future = Future()
    def my_clb(...):
        future.set_result(xyz)
    fn(..., my_clb)
    return future

（在异常情况下使用future.set_exception(exc)）

然后你可以简单地在一些 async 函数中使用 await 调用该包装器：

value = await wrapper(...)

请注意，await 适用于任何 Future 对象。您不必将wrapper 声明为async。

如果库中的函数是同步的，那么您可以在单独的线程中运行它（可能您会为此使用一些线程池）。整个代码可能如下所示：

import asyncio
import time
from concurrent.futures import ThreadPoolExecutor

# Initialize 10 threads
THREAD_POOL = ThreadPoolExecutor(10)

def synchronous_handler(param1, ...):
    # Do something synchronous
    time.sleep(2)
    return "foo"

# Somewhere else
async def main():
    loop = asyncio.get_event_loop()
    futures = [
        loop.run_in_executor(THREAD_POOL, synchronous_handler, param1, ...),
        loop.run_in_executor(THREAD_POOL, synchronous_handler, param1, ...),
        loop.run_in_executor(THREAD_POOL, synchronous_handler, param1, ...),
    ]
    await asyncio.wait(futures)
    for future in futures:
        print(future.result())

with THREAD_POOL:
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

如果您出于某种原因不能使用线程，那么使用这样的库只会使整个异步代码变得毫无意义。

但是请注意，将同步库与异步一起使用可能不是一个好主意。你不会得到太多，但你却使代码复杂了很多。

【讨论】：

@grafa 我的错，我没有测试代码。我现在已经修复并简化了很多。实际上 Python 有这个简洁的 .run_in_executor 函数，它负责返回值和异常处理。无需手动操作。

@grafa 请注意，如果您调度的任务超过线程池的大小，那么它们将被排队等待线程。

@grafa 另请记住，之后您必须关闭执行程序。在此处阅读更多信息：docs.python.org/3/library/concurrent.futures.html

@grafa 首先您安排通话，然后您可以使用asyncio.wait() 等待所有通话。最后，您可以通过调用future.result() 循环访问期货并检索结果。请参阅更新的代码。注意.run_in_executor已经返回了一个future对象，不需要包装它。

非常感谢。每个想法都有效，我想我知道整个事情。您是这个星球上第一个提供了如何将 Python 3.5 async/await 与经典非异步任务一起使用的复杂示例的人。再次感谢！

【参考方案2】：

您可以查看here 中的以下简单工作示例。顺便说一句，它返回一个值得一读的字符串:-)

import aiohttp
import asyncio

async def fetch(client):
  async with client.get('https://docs.aiohttp.org/en/stable/client_reference.html') as resp:
    assert resp.status == 200
    return await resp.text()

async def main():
  async with aiohttp.ClientSession() as client:
    html = await fetch(client)
    print(html)

loop = asyncio.get_event_loop()
loop.run_until_complete(main())