Python 3.5中的协同程序和未来/任务之间的区别？

Question

假设我们有一个虚函数:

async def foo(arg):
    result = await some_remote_call(arg)
    return result.upper()

有什么区别:

coros = []
for i in range(5):
    coros.append(foo(i))

loop = get_event_loop()
loop.run_until_complete(wait(coros))

和:

from asyncio import ensure_future

futures = []
for i in range(5):
    futures.append(ensure_future(foo(i)))

loop = get_event_loop()
loop.run_until_complete(wait(futures))

注意:该示例返回结果,但这不是问题的焦点.当返回值很重要时,请使用gather()而不是wait().

无论回报价值如何,我都在寻求清晰度ensure_future().wait(coros)并且wait(futures)都运行协同程序,所以何时以及为什么要将协程包装进去ensure_future？

基本上,使用Python 3.5运行一堆非阻塞操作的正确方法(tm)是async什么？

如果我想批量通话,如果需要额外的积分？例如,我需要调用some_remote_call(...)1000次,但我不想粉碎Web服务器/数据库/等1000个同时连接.这对于线程或进程池是可行的,但有没有办法做到这一点asyncio？

Answer 1

协程是一个生成器函数,它既可以产生值,也可以接受来自外部的值.使用协程的好处是我们可以暂停函数的执行并在以后恢复它.在网络操作的情况下,在我们等待响应时暂停执行功能是有意义的.我们可以利用时间来运行其他一些功能.

未来就像Promise来自Javascript 的对象.它就像是一个将在未来实现的价值的占位符.在上述情况下,在等待网络I/O时,一个函数可以给我们一个容器,一个承诺,它将在操作完成时用容器填充容器.我们坚持未来的对象,当它完成时,我们可以调用一个方法来检索实际的结果.

直接回答:你并不需要ensure_future,如果你不想要的结果.如果您需要结果或检索发生的异常,它们就很好.

额外积分:我会选择run_in_executor并传递一个Executor实例来控制最大工人的数量.

说明和示例代码

在第一个示例中,您正在使用协同程序.该wait函数需要一堆协同程序并将它们组合在一起.所以wait()当所有协程都耗尽时完成(完成/完成返回所有值).

loop = get_event_loop() # 
loop.run_until_complete(wait(coros))

该run_until_complete方法将确保循环处于活动状态,直到执行完成.请注意在这种情况下如何获得异步执行的结果.

在第二个示例中,您正在使用该ensure_future函数来包装协程并返回一个Task对象Future.当您调用时,协程计划在主事件循环中执行ensure_future.返回的future/task对象还没有值,但随着时间的推移,当网络操作完成时,future对象将保存操作的结果.

from asyncio import ensure_future

futures = []
for i in range(5):
    futures.append(ensure_future(foo(i)))

loop = get_event_loop()
loop.run_until_complete(wait(futures))

所以在这个例子中,我们正在做同样的事情,除了我们使用的是期货而不仅仅是使用协同程序.

我们来看一个如何使用asyncio/coroutines /期货的例子:

import asyncio


async def slow_operation():
    await asyncio.sleep(1)
    return 'Future is done!'


def got_result(future):
    print(future.result())

    # We have result, so let's stop
    loop.stop()


loop = asyncio.get_event_loop()
task = loop.create_task(slow_operation())
task.add_done_callback(got_result)

# We run forever
loop.run_forever()

在这里,我们create_task在loop对象上使用了该方法.ensure_future将在主事件循环中安排任务.这种方法使我们能够在我们选择的循环上安排一个协同程序.

我们还看到了add_done_callback在任务对象上使用该方法添加回调的概念.

A Task是done协程返回值,引发异常或被取消的时候.有方法可以检查这些事件.

我写了一些关于这些主题的博客文章可能有所帮助:

• http://masnun.com/2015/11/13/python-generators-coroutines-native-coroutines-and-async-await.html
• http://masnun.com/2015/11/20/python-asyncio-future-task-and-the-event-loop.html
• http://masnun.com/2015/12/07/python-3-using-blocking-functions-or-codes-with-asyncio.html

当然,您可以在官方手册中找到更多详细信息:https://docs.python.org/3/library/asyncio.html

Answer 2

简单的答案是

• 调用croutine函数(async def)不运行它.它只返回协程对象,就像生成器函数返回生成器对象一样.
• await 从协程中检索值,即调用协程
• eusure_future/create_task 安排协程在下一次迭代时在事件循环上运行(尽管没有等待它们完成,就像一个守护程序线程).

一些代码示例

我们先来说清楚一些术语:

• 协程功能,你async def是谁
• coroutine,当你调用corotine函数时你得到了什么

似乎下面的评论.

案例1,await关于协程

我们创建了两个协同程序,await一个,并使用create_task运行另一个协同程序.

import asyncio
import time

# coroutine function
async def p(word):
    print(f'{time.time()} - {word}')


async def main():
    loop = asyncio.get_event_loop()
    coro = p('await')  # coroutine
    task2 = loop.create_task(p('create_task'))  # <- runs in next iteration
    await coro  # <-- run directly
    await task2

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

你会得到结果:

1539486251.7055213 - await
1539486251.7055705 - create_task

说明:

task1直接执行,task2在下一次迭代中执行.

情况2,产生对事件循环的控制

如果我们替换main函数,我们可以看到不同的结果:

async def main():
    loop = asyncio.get_event_loop()
    coro = p('await')
    task2 = loop.create_task(p('create_task'))  # scheduled to next iteration
    await asyncio.sleep(1)  # loop got control, and runs task2
    await coro  # run coro
    await task2

你会得到结果:

-> % python coro.py
1539486378.5244057 - create_task
1539486379.5252144 - await  # note the delay

说明:

调用时create_task,控件返回到事件循环,循环检查要运行的任务,然后运行由创建的任务asyncio.sleep(1).

请注意,我们首先调用corotine函数,但不是create_task它,因此我们只创建了一个corotine,而不是让它运行.然后,我们再次调用corotine函数,并将其包装在一个await调用中,creat_task将按照惯例安排协程在下一次迭代时运行.所以,在结果中,create_task之前执行create task.

实际上,这里的重点是将控制权交还给循环,您可以使用它await来查看相同的结果.

在引擎盖下

asyncio.sleep(0)实际上是电话loop.create_task,会打电话asyncio.tasks.Task().并将loop.call_soon完成任务loop.call_soon.在循环的每次迭代期间,它会检查loop._ready中的每个回调并运行它.

loop._ready,asyncio.wait并直接或间接地asyncio.ensure_future打电话asyncio.gather.

另请注意文档:

回调按其注册顺序调用.每个回调将被调用一次.

Answer 3

文森特的评论链接到https://github.com/python/asyncio/blob/master/asyncio/tasks.py#L346,它显示wait()了ensure_future()为你包装协同程序!

换句话说,我们确实需要一个未来,协同作用将被默默地转化为它们.

当我找到关于如何批量协程/期货的明确解释时,我会更新这个答案.

Answer 4

来自 BDFL [2013]

任务

• 这是一个包裹在 Future 中的协程
• Task 类是 Future 类的子类
• 所以它也适用于await！

• 它与裸协程有何不同？
• 无需等待就能取得进展
  • 只要你等待别的事情，即
    • 等待[其他东西]

考虑到这一点，作为创建任务的名称是有意义的，因为无论您是否等待它（只要您等待某些东西），ensure_future都会计算 Future 的结果。这允许事件循环在您等待其他事情时完成您的任务。请注意，Python 3.7 中是确保 future 的首选方式。create_task

注意：为了现代化，我将 Guido 幻灯片中的“yield from”更改为“await”。