递归函数上的 Asyncio loop.run_until_complete()

Question

我正在尝试自学 asyncio，因此我构建了一个 web scraper 作为练习。主要工作部分是以下递归函数...

async def get_gids(base_url='http://gd2.mlb.com/components/game/mlb/'):
    soup = await get_soup(base_url)

    for link in soup.find_all('a'):
        dest = link.get('href')

        fut = asyncio.Future()

        def set_result(out):
            try:
                fut.set_result(out.result())
            except:
                pass

        if dest.startswith('year_') and dest[YEAR] in VALID_YEARS:
            in_fut = asyncio.ensure_future(get_gids(base_url + dest))
            in_fut.add_done_callback(set_result)
        elif dest.startswith('month_') and dest[MONTH] in VALID_MONTHS:
            in_fut = asyncio.ensure_future(get_gids(base_url + dest))
            in_fut.add_done_callback(set_result)
        elif dest.startswith('day_'):
            in_fut = asyncio.ensure_future(get_gids(base_url + dest))
            in_fut.add_done_callback(set_result)
        elif dest.startswith('gid_'):
            if await is_regular_season(base_url + dest):
                print('gid: {}'.format(dest))
                gids.append(dest)
        else:
            pass

在run_forever 循环中，这很有效。但是，希望它运行并完成而无需我杀死它，例如run_unitl_complete(get_gids())，但这似乎不起作用，因为我只得到第一页。我试过这个...

f = asyncio.wait(get_gids())  
loop.run_until_complete(f)  

再次引发 TypeError 没有成功...TypeError: expect a list of futures, not coroutine

在原始函数调用从调用堆栈中弹出之前，我该怎么做才能运行循环？

Answer 1

您应该收到来自asyncio 的警告，因为您当前的代码创建了一个任务，但没有await 它或将其返回以执行其他操作。

import asyncio

def callback(f):
    print("Callback: {}".format(f.result()))


async def non_awaiting_coro():
    print("Executing non awaiting coroutine")
    fut = asyncio.ensure_future(asyncio.sleep(0.25))
    fut.add_done_callback(callback)
    print("Done executing coroutine")


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

返回以下内容：

Executing non awaiting coroutine
Done executing coroutine
Task was destroyed but it is pending!
task: <Task pending coro=<sleep() running at ...>

注意回调没有被触发。如果你等待fut，协程应该完成：

async def awaiting_coro():
    print("Executing coroutine")
    fut = asyncio.ensure_future(asyncio.sleep(0.25))
    fut.add_done_callback(callback)
    await fut
    print("Done executing coroutine")

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

返回：

Executing coroutine
Callback: None
Done executing coroutine

Answer 2

你尝试了一件棘手的事情——异步递归。您可能会注意到有一个警告：

RuntimeWarning：从未等待协程“get_gids”

这意味着你递归创建的协程不是在循环中开始的，也不是await。您可以在文档中找到该行为的描述：

调用协程不会启动其代码运行——调用返回的协程对象在您安排其执行之前不会做任何事情。启动它有两种基本方法：调用 await 协程或从另一个协程的协程中产生（假设另一个协程已经在运行！），或使用 ensure_future() 函数或 AbstractEventLoop.create_task() 方法安排其执行。

虽然您只使用run_until_complete 安排了第一个创建的协程。如果循环正在运行，则可以调度其他协程。

可以在协程内部使用awaiting 解决这种情况，但这是一种阻塞方式，所以有点无意义：

async def recursive(arg=0):
if arg < 10:
    fut = asyncio.ensure_future(recursive(arg+1))
    fut.add_done_callback(lambda out: print(out))
    await fut

    # Simulation of extra ections                                 
    count = 3                                    
    while count:                                 
        await asyncio.sleep(random.random())     
        count -= 1                               
return arg


def main():
    loop = asyncio.get_event_loop()
    loop.run_until_complete(recursive())


if __name__ == "__main__":
    main()

您还可以使用全局循环来运行所有协程：

LOOP = asyncio.get_event_loop()                      
TASKS = []                                           

async def recursive(arg=0):                                                                 
    if arg < 10:                                     
        fut = asyncio.ensure_future(recursive(arg+1))
        TASKS.append(fut)                            
        fut.add_done_callback(lambda out: print(out))

        # Simulation of extra actions.               
        count = 3                           
        while count:                                 
            count -= 1                               
            await asyncio.sleep(random.random())     
    return arg                                       


# Stop the loop when everything is done                                
async def shutdown():                                
    while not all(i.done() for i in TASKS):          
        await asyncio.sleep(.1)                      
    LOOP.stop()                                      


def main():                                          
    fut = asyncio.ensure_future(recursive())         
    TASKS.append(fut)                                
    asyncio.ensure_future(shutdown())                
    LOOP.run_forever()                               


if __name__ == "__main__":                           
    main()             

这种方法与具有所有全局变量的良好架构相去甚远，但它抓住了本质。它也符合您的要求

希望它在我不必杀死它的情况下运行并完成

要做到这一点，您需要有一个协程，它在一切完成后关闭一个循环。