python异步编程之 async await

python异步编程之 async await

本文代码采用python3.6运行.

发展史

- 3.3: The yield from expression allows for generator delegation.
- 3.4: asyncio was introduced in the Python standard library with provisional API status.
- 3.5: async and await became a part of the Python grammar, used to signify and wait on coroutines. They were not yet reserved keywords. (You could still define functions or variables named async and await.)
- 3.6: Asynchronous generators and asynchronous comprehensions were introduced. The API of asyncio was declared stable rather than provisional.
- 3.7: async and await became reserved keywords. (They cannot be used as identifiers.) They are intended to replace the asyncio.coroutine() decorator. asyncio.run() was introduced to the asyncio package, among a bunch of other features.

本质上是使用了协程,当调用await时让渡CPU,有结果返回时再切换回来.相比使用回调来协调执行顺序来说,await编程方式在每个协程中代码是顺序执行的,对代码编写来说更为友好.

语法

async def

调用一个async func返回了什么?

async def fun1():
    return

async def fun2():
    yield

print(type(fun1()))
print(type(fun2()))

# <class 'coroutine'>
# <class 'async_generator'>

await

• await只能用于async函数中

  async def f(x):
      y = await z(x)  # OK - `await` and `return` allowed in coroutines
      return y
  
  async def g(x):
      yield x  # OK - this is an async generator
  
  async def m(x):
      yield from gen(x)  # No - SyntaxError
  
  def m(x):
      y = await z(x)  # Still no - SyntaxError (no `async def` here)
      return y
  

• await后面只能跟一个awaitable对象(一般情况下是一个Coroutine对象)

调用(启用)方式

import time
import asyncio


async def coro(seq) -> list:
    """'IO' wait time is proportional to the max element."""
    await asyncio.sleep(max(seq))
    return list(reversed(seq))


async def main():
    """
    ensure_future():创建任务,Python 3.7+后被create_task()取代.
    t.done(): 返回任务执行状态:Bool类型.

    执行结果:
    ------------------------------------------
    t: type <class '_asyncio.Task'>
    t done: False
    t result: [1, 2, 3]
    t done: True
    ------------------------------------------
    """
    # t = asyncio.create_task(coro([3, 2, 1]))  # Python 3.7+
    t = asyncio.ensure_future(coro([3, 2, 1]))  # Python 3.*

    print(f't: type {type(t)}')
    print(f't done: {t.done()}')
    a = await t
    print(f't result: {a}')
    print(f't done: {t.done()}')

async def main2():
    """
    tasks = gather():收集多个任务
    await tasks:任务全部完成一起返回

    执行结果:
    ------------------------------------------
    Start: 11:11:29
    [[1, 2, 3], [0, 5, 6]]
    End: 11:11:35
    Both tasks done: True
    ------------------------------------------
    """
    t = asyncio.ensure_future(coro([3, 2, 1]))
    t2 = asyncio.ensure_future(coro([6, 5, 0]))
    print('Start:', time.strftime('%X'))
    a = await asyncio.gather(t, t2)
    print(a)
    print('End:', time.strftime('%X'))  # Should be 10 seconds
    print(f'Both tasks done: {all((t.done(), t2.done()))}')


async def main3():
    """
    as_completed 每个任务完成即返回该任务的结果.

    执行结果:
    ------------------------------------------
    Start: 11:08:13
    res: [1, 2, 3] completed at 11:08:16
    # 三秒后
    res: [0, 5, 6] completed at 11:08:19
    End: 11:08:19
    Both tasks done: True
    ------------------------------------------

    """
    t = asyncio.ensure_future(coro([3, 2, 1]))
    t2 = asyncio.ensure_future(coro([6, 5, 0]))
    print('Start:', time.strftime('%X'))
    for res in asyncio.as_completed((t, t2)):
        compl = await res
        print(f'res: {compl} completed at {time.strftime("%X")}')
    print('End:', time.strftime('%X'))
    print(f'Both tasks done: {all((t.done(), t2.done()))}')


if __name__ == '__main__':
    # python < 3.7
    loop = asyncio.get_event_loop()
    # loop.run_until_complete(main())
    # loop.run_until_complete(main2())
    loop.run_until_complete(main3())
    loop.close()

    # python 3.7
    # asyncio.run(main())

DEMO

并发http请求

该Demo使用aiohttp发送异步请求.

"""
并发爬取36kr数据,并保存为本地文件
"""
import asyncio
import aiohttp
import json


async def get_json(client, url):
    async with client.get(url) as response:
        assert response.status == 200
        return await response.read()


async def get_36_kr(page, client):
    data1 = await get_json(client, 'https://36kr.com/api/search-column/mainsite?per_page=20&page={}'.format(page))
    j = json.loads(data1.decode('utf-8'))

    with open(f"./content/{page}.json", "w", encoding='utf-8') as f:
        f.write(json.dumps(j, indent=2, ensure_ascii=False))

    print('Done with page:', page)


async def main():
    client = aiohttp.ClientSession(loop=loop)
    tasks = [asyncio.ensure_future(get_36_kr(i, client)) for i in range(100)]
    await asyncio.gather(*tasks)
    await client.close()

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

精简下(需python3.7下运行)

import asyncio
import aiohttp
import json


async def get_36_kr(page, client):
    url = f'https://36kr.com/api/search-column/mainsite?per_page=1&page={page}'
    async with client.get(url, ssl=False) as response:
        assert response.status == 200
        return await response.json()


async def main():
    async with aiohttp.ClientSession() as client:
        tasks = [asyncio.ensure_future(get_36_kr(i, client))
                 for i in range(2)]
        res = await asyncio.gather(*tasks)
        print(json.dumps(res,indent=2))

if __name__ == '__main__':
    asyncio.run(main())

异步IO库

async/await可使用的库.

• From aio-libs:

  • aiohttp: Asynchronous HTTP client/server framework
  • aioredis: Async IO Redis support
  • aiopg: Async IO PostgreSQL support
  • aiomcache: Async IO memcached client
  • aiokafka: Async IO Kafka client
  • aiozmq: Async IO ZeroMQ support
  • aiojobs: Jobs scheduler for managing background tasks
  • async_lru: Simple LRU cache for async IO

• From magicstack:

  • uvloop: Ultra fast async IO event loop
  • asyncpg: (Also very fast) async IO PostgreSQL support

• From other hosts:

  • trio: Friendlier asyncio intended to showcase a radically simpler design
  • aiofiles: Async file IO
  • asks: Async requests-like http library
  • asyncio-redis: Async IO Redis support
  • aioprocessing: Integrates multiprocessing module with asyncio
  • umongo: Async IO MongoDB client
  • unsync: Unsynchronize asyncio
  • aiostream: Like itertools, but async