Python 3.5异步/等待与真实代码示例

Question

我读过吨的文章和教程有关Python的3.5异步/等待的事情。我不得不说，我很困惑，因为一些使用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()) 

我希望你明白我的感受是多么的迷茫，我想要平行运行多么简单的事情。

Answer 1

如果第三方库与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 
   

   （上异常使用future.set_exception(exc)）

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

   value = await wrapper(...) 
   

   注意await适用于任何Future对象。您不必声明wrapper为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()) 
   

如果你不能使用线程无论出于何种原因，然后使用这些库只是使整个异步代码毫无意义。

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