Python中最简单的异步/等待示例

Question

我在Python 3.5+中读过许多关于asyncio/ async/的博客文章,问题/答案await,很多都是复杂的,我发现最简单的可能是这个.它仍然使用ensure_future,并且为了学习Python中的异步编程,我想看看是否有更简单的例子是可能的(即,执行基本异步/等待示例所需的最小工具是什么).

问题:为了学习Python中的异步编程,是否可以通过仅使用这两个关键字+ + +其他Python代码但没有其他函数来提供一个显示如何async/ await工作的简单示例？asyncio.get_event_loop()run_until_completeasyncio

示例:类似这样的事情:

import asyncio

async def async_foo():
    print("async_foo started")
    await asyncio.sleep(5)
    print("async_foo done")

async def main():
    asyncio.ensure_future(async_foo())  # fire and forget async_foo()
    print('Do some actions 1')
    await asyncio.sleep(5)
    print('Do some actions 2')

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

但没有ensure_future,仍然演示了await/async的工作原理.

Answer 1

是否可以 通过仅使用这两个关键字+ + +其他Python代码但没有其他函数给出一个显示如何async/ await工作的简单示例？asyncio.get_event_loop()run_until_completeasyncio

这样就可以编写有效的代码:

import asyncio


async def main():
    print('done!')


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

但这样就无法证明为什么需要asyncio.

那么,为什么你需要asyncio,而不仅仅是简单的代码？答案是 - asyncio允许您在并行化I/O阻塞操作(如读/写网络)时获得性能优势.要编写有用的示例,您需要使用这些操作的异步实现.

请阅读此答案以获得更详细的解释.

UPD:

好的,这是asyncio.sleep用于模仿I/O阻塞操作的示例,asyncio.gather它显示了如何同时运行多个阻塞操作:

import asyncio


async def io_related(name):
    print(f'{name} started')
    await asyncio.sleep(1)
    print(f'{name} finished')


async def main():
    await asyncio.gather(
        io_related('first'),
        io_related('second'),
    )  # 1s + 1s = over 1s


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

输出:

first started
second started
first finished
second finished
[Finished in 1.2s]

注意两者都是io_related在两秒钟之后开始的.

Answer 2

为了回答您的问题，我将为同一问题提供3种不同的解决方案。

情况1：普通的python

import time

def sleep():
    print(f'Time: {time.time() - start:.2f}')
    time.sleep(1)

def sum(name, numbers):
    total = 0
    for number in numbers:
        print(f'Task {name}: Computing {total}+{number}')
        sleep()
        total += number
    print(f'Task {name}: Sum = {total}\n')

start = time.time()
tasks = [
    sum("A", [1, 2]),
    sum("B", [1, 2, 3]),
]
end = time.time()
print(f'Time: {end-start:.2f} sec')

输出：

Task A: Computing 0+1
Time: 0.00
Task A: Computing 1+2
Time: 1.00
Task A: Sum = 3

Task B: Computing 0+1
Time: 2.01
Task B: Computing 1+2
Time: 3.01
Task B: Computing 3+3
Time: 4.01
Task B: Sum = 6

Time: 5.02 sec

情况2：异步/等待做错了

import asyncio
import time

async def sleep():
    print(f'Time: {time.time() - start:.2f}')
    time.sleep(1)

async def sum(name, numbers):
    total = 0
    for number in numbers:
        print(f'Task {name}: Computing {total}+{number}')
        await sleep()
        total += number
    print(f'Task {name}: Sum = {total}\n')

start = time.time()

loop = asyncio.get_event_loop()
tasks = [
    loop.create_task(sum("A", [1, 2])),
    loop.create_task(sum("B", [1, 2, 3])),
]
loop.run_until_complete(asyncio.wait(tasks))
loop.close()

end = time.time()
print(f'Time: {end-start:.2f} sec')

输出：

Task A: Computing 0+1
Time: 0.00
Task A: Computing 1+2
Time: 1.00
Task A: Sum = 3

Task B: Computing 0+1
Time: 2.01
Task B: Computing 1+2
Time: 3.01
Task B: Computing 3+3
Time: 4.01
Task B: Sum = 6

Time: 5.01 sec

情况3：异步/等待正确完成（与情况2相同，除了sleep功能不同）

import asyncio
import time

async def sleep():
    print(f'Time: {time.time() - start:.2f}')
    await asyncio.sleep(1)

async def sum(name, numbers):
    total = 0
    for number in numbers:
        print(f'Task {name}: Computing {total}+{number}')
        await sleep()
        total += number
    print(f'Task {name}: Sum = {total}\n')

start = time.time()

loop = asyncio.get_event_loop()
tasks = [
    loop.create_task(sum("A", [1, 2])),
    loop.create_task(sum("B", [1, 2, 3])),
]
loop.run_until_complete(asyncio.wait(tasks))
loop.close()

end = time.time()
print(f'Time: {end-start:.2f} sec')

输出：

Task A: Computing 0+1
Time: 0.00
Task B: Computing 0+1
Time: 0.00
Task A: Computing 1+2
Time: 1.00
Task B: Computing 1+2
Time: 1.00
Task A: Sum = 3

Task B: Computing 3+3
Time: 2.00
Task B: Sum = 6

Time: 3.01 sec

case 1与case 2给相同5 seconds，而case 3公正3 seconds。因此，async/await done right速度更快。

差异的原因在于sleep功能的实现内。

# case 1
def sleep():
    print(f'Time: {time.time() - start:.2f}')
    time.sleep(1)

# case 2
async def sleep():
    print(f'Time: {time.time() - start:.2f}')
    time.sleep(1)

# case 3
async def sleep():
    print(f'Time: {time.time() - start:.2f}')
    await asyncio.sleep(1)

sleep在case 1和case 2中是“相同”的。他们“睡觉”而不允许其他人使用资源。而case 3允许在睡眠状态下访问资源。

在case 2我们添加async到正常功能。但是，事件循环将在不中断的情况下运行它。为什么？因为我们没有说允许循环在哪里中断您的函数以运行另一个任务。

在case 3我们告诉事件循环中，确切的位置中断了该函数以运行另一个任务。到底在哪里？

# case 3
async def sleep():
    print(f'Time: {time.time() - start:.2f}')
    await asyncio.sleep(1) # <-- Right here!

更多关于此阅读这里

Answer 3

Python 3.7+ 现在有一个更简单的 API（在我看来）和更简单的措辞（比“ensure_future”更容易记住）：您可以使用create_taskwhich 返回一个 Task 对象（如果需要，这在以后取消任务时很有用）。

基本示例 1

import asyncio

async def hello(i):
    print(f"hello {i} started")
    await asyncio.sleep(4)
    print(f"hello {i} done")

async def main():
    task1 = asyncio.create_task(hello(1))  # returns immediately, the task is created
    await asyncio.sleep(3)
    task2 = asyncio.create_task(hello(2))
    await task1
    await task2

asyncio.run(main())  # main loop

结果：

你好1开始
你好2开始
你好1完成
你好2完成

基本示例 2

如果需要获取这些异步函数的返回值，那么gather很有用。以下示例的灵感来自文档，但不幸的是，该文档没有显示gather真正有用的内容：获取返回值！

import asyncio

async def factorial(n):
    f = 1
    for i in range(2, n + 1):
        print(f"Computing factorial({n}), currently i={i}...")
        await asyncio.sleep(1)
        f *= i
    return f

async def main():
    L = await asyncio.gather(factorial(2), factorial(3), factorial(4))
    print(L)  # [2, 6, 24]

asyncio.run(main())

预期输出：

计算阶乘（2），当前 i=2...
计算阶乘（3），当前 i=2...
计算阶乘（4），当前 i=2...
计算阶乘（3），当前 i=3。 ..
计算阶乘(4), 当前 i=3...
计算阶乘 (4), 当前 i=4...
[2, 6, 24]

PS：即使你使用asyncio，而不是trio，后者的教程对我理解 Python 异步编程很有帮助。

Answer 4

既然一切都得到了很好的解释，那么让我们运行一些带有事件循环的示例，将同步代码与异步代码进行比较。

同步代码：

import time

def count():
    time.sleep(1)
    print('1')
    time.sleep(1)
    print('2')
    time.sleep(1)
    print('3')

def main():
    for i in range(3):
        count()

if __name__ == "__main__":
    t = time.perf_counter()
    main()
    t2 = time.perf_counter()
    
    print(f'Total time elapsed: {t2:0.2f} seconds')

输出：

1
2
3
1
2
3
1
2
3
Total time elapsed: 9.00 seconds

我们可以看到，在下一个循环开始之前，每个循环计数运行到完成。

异步代码：

import asyncio
import time

async def count():
    await asyncio.sleep(1)
    print('1')
    await asyncio.sleep(1)
    print('2')
    await asyncio.sleep(1)
    print('3')

async def main():
    await asyncio.gather(count(), count(), count())

if __name__ == "__main__":
    t = time.perf_counter()
    asyncio.run(main())
    t2 = time.perf_counter()

    print(f'Total time elapsed: {t2:0.2f} seconds')

输出：

1
1
1
2
2
2
3
3
3
Total time elapsed: 3.00 seconds

另一方面，异步等价物看起来像这样需要 3 秒才能运行，而不是 9 秒。第一个计数周期开始了，一旦它达到awaits sleep one Python 可以自由地做其他工作，例如开始第二个计数周期，然后是第三个计数周期。这就是为什么我们拥有所有的管而不是所有的管，然后是所有三个。在输出中并发编程是一个非常有价值的工具。多处理让操作完成所有的多任务工作，在 Python 中，它是多核并发的唯一选择，让您的程序在 CPU 的多个内核上执行。如果使用线程，则操作系统仍在执行所有多任务工作，并且在 cpython 中，全局内部互操作器锁可防止异步编程中的多核并发。没有操作系统干预，一个进程有一个线程，所以正在发生的事情可以在等待期间释放 CPU，以便其他任务可以使用它。

import asyncio

loop = asyncio.get_event_loop()


async def greeter(name):
    print(f"Hi, {name} you're in a coroutine.")

try:
    print('starting coroutine')
    coro = greeter('LP')
    print('entering event loop')
    loop.run_until_complete(coro)
finally:
    print('closing event loop')
    loop.close()

输出：

starting coroutine
entering event loop
Hi, LP you're in a coroutine.
closing event loop

异步框架需要一个通常称为事件循环的调度器。这个事件循环跟踪所有正在运行的任务，当一个函数挂起时，它会将控制权返回给事件循环，然后事件循环会找到另一个函数来启动或恢复，这称为协作多任务。Async IO 提供了一个框架，一个以这个事件循环为中心的异步框架，它有效地处理输入/输出事件，一个应用程序与事件循环显式交互，它注册要运行的代码，然后它让事件循环调度程序进行必要的调用资源可用时的应用程序代码。因此，如果网络服务器打开套接字然后注册它们以在输入事件发生时被告知事件循环将在有新的传入连接或有新的传入连接时提醒服务器代码 要读取的数据。如果没有比服务器更多的数据要从套接字读取，则将控制权交还给事件循环。

从让出控制回到事件循环的机制取决于协同程序协同程序是一种为并发操作而设计的语言结构。协程可以使用唤醒关键字与另一个协程暂停执行，当它暂停时，协程状态被保持，允许它从停止的地方恢复，一个协程可以启动另一个协程，然后等待结果，这可以更轻松地将任务分解为可重用的部分。

import asyncio

loop = asyncio.get_event_loop()

async def outer():
    print('in outer')
    print('waiting for result 1')
    result1 = await phase1()
    print('waiting for result 2')
    result2 = await phase2(result1)
    return result1, result2


async def phase1():
    print('in phase1')
    return 'phase1 result'

async def phase2(arg):
    print('in phase2')
    return 'result2 derived from {}'.format(arg)

asyncio.run(outer())

输出：

in outer
waiting for result 1
in phase1
waiting for result 2
in phase2

此示例要求必须按顺序执行但可以与其他操作并发运行的两个阶段。使用awake关键字而不是向循环添加新的协同程序，因为控制流已经在循环管理的协同程序内部。没有必要告诉循环来管理新的协程。

Answer 5

我不知道为什么，但关于这个主题的所有解释都太复杂，或者他们使用的示例没有无用的 asyncio.sleep()...到目前为止，我发现的最好的代码示例是这样的： https: //codeflex.co /python3-async-await-示例/