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

我读过大量关于 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())

我希望你明白我有多么失落，以及我希望并行运行是多么简单。

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

1. 假设库中的函数是异步的，并且它为您提供回调，例如

   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
   

   (use future.set_exception(exc)例外情况）

   然后你可以简单地在某些地方调用该包装器async功能与await:

   value = await wrapper(...)
   

   注意await适用于任何Future目的。你不必声明wrapper as async.

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

   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())
   

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

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