Python Library Async Notes

asyncio

Use async IO when you can; use threading when you must

example

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16


import asyncio

async def count():
    print("One")
    await asyncio.sleep(1)
    print("Two")

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

if __name__ == "__main__":
    import time
    s = time.perf_counter()
    asyncio.run(main())
    elapsed = time.perf_counter() - s
    print(f"{__file__} executed in {elapsed:0.2f} seconds.")

在 async 函数main的里面，asyncio.gather() 方法将多个异步任务（三个 count()）包装成一个新的异步任务，必须等到内部的多个异步任务都执行结束，这个新的异步任务才会结束。

三个 count() 依次执行，打印完 One，就休眠1秒钟，把执行权交给下一个 count()，所以先连续打印出三个 One。等到1秒钟休眠结束，执行权重新交回第一个 count()，开始执行 await 命令下一行的语句，所以会接着打印出三个Two。脚本总的运行时间是1秒。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19


import asyncio

async def func():
    cnt = 0
    for i in range(1000000):
        cnt += i
    print("i = ", cnt)

async def count():
    print("One")
    # await asyncio.sleep(1)
    await func()
    print("Two")

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

asyncio.run(main())

以上并没有表现出异步的执行，因为一直有计算在执行。只有一个协程由于 IO 卡住了就会异步的去执行其其它的任务进行计算。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16


import asyncio
import time

async def say_after(delay, what):
    await asyncio.sleep(delay)
    print(what)

async def main():
    print(f"started at {time.strftime('%X')}")

    await say_after(1, 'hello')
    await say_after(2, 'world')

    print(f"finished at {time.strftime('%X')}")

asyncio.run(main())

输出为

started at 10:51:11
hello
world
finished at 10:51:14

可看出以上代码是按顺序执行。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23


import asyncio
import time

async def say_after(delay, what):
    await asyncio.sleep(delay)
    print(what)

async def main():
    task1 = asyncio.create_task(
        say_after(1, 'hello'))

    task2 = asyncio.create_task(
        say_after(2, 'world'))

    print(f"started at {time.strftime('%X')}")

    # Wait until both tasks are completed (should take
    # around 2 seconds.)
    await task1
    await task2

    print(f"finished at {time.strftime('%X')}")
asyncio.run(main())

输出为

started at 10:52:50
hello
world
finished at 10:52:52

可看出这样才会异步的在执行。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21


import asyncio
import random

async def myCoroutine(id):
    process_time = random.randint(1,5)
    await asyncio.sleep(process_time)
    print("Coroutine: {}, has successfully completed after {} seconds".format(id, process_time))

async def main():
    tasks = []
    for i in range(10):
        tasks.append(asyncio.ensure_future(myCoroutine(i)))

    await asyncio.gather(*tasks)


loop = asyncio.get_event_loop()
try:
    loop.run_until_complete(main())
finally:
    loop.close()

Awaitables

可以使用 await 的三种对象: coroutines, Tasks, Futures.

coroutines

A coroutine is a specialized version of a Python generator function. A coroutine is a function that can suspend its execution before reaching return, and it can indirectly pass control to another coroutine for some time.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15


import asyncio

async def nested():
    return 42

async def main():
    # Nothing happens if we just call "nested()".
    # A coroutine object is created but not awaited,
    # so it *won't run at all*.
    nested()

    # Let's do it differently now and await it:
    print(await nested())  # will print "42".

asyncio.run(main())

Tasks

Tasks are used to schedule coroutines concurrently.

When a coroutine is wrapped into a Task with functions like asyncio.create_task() the coroutine is automatically scheduled to run soon:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15


import asyncio

async def nested():
    return 42

async def main():
    # Schedule nested() to run soon concurrently
    # with "main()".
    task = asyncio.create_task(nested())

    # "task" can now be used to cancel "nested()", or
    # can simply be awaited to wait until it is complete:
    await task

asyncio.run(main())

Futures

1
2
3
4
5
6
7
8


async def main():
    await function_that_returns_a_future_object()

    # this is also valid:
    await asyncio.gather(
        function_that_returns_a_future_object(),
        some_python_coroutine()
    )

uvloop

uvloop is a fast, drop-in replacement of the built-in asyncio event loop. uvloop is implemented in Cython and uses libuv under the hood.

To make asyncio use uvloop, you can install the uvloop event loop policy:

1
2
3


import asyncio
import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

1
2


loop = uvloop.new_event_loop()
asyncio.set_event_loop(loop)

pyuv

Python interface for libuv, a high performance asynchronous networking and platform abstraction library.