[Java知识库]asyncio异步io并发编程

一、关于asyncio

asyncio是解决异步io高并发编程的核心模块，python3.4后开始引用，可以说是python中最具野心的一个模块，无论是高并发web服务器还是高并发爬虫都可以胜任。

asyncio提供了异步IO编程的一整套方案，包括:

• 包含各种特定系统都能够兼容的模块化事件循环。
• 传输和协议抽象。
• 实现了对TCP、UDP、SSL、子进程、延时调用等的具体支持。
• 模仿futures模块但适用于事件循环使用的Future类。
• 基于yield from的协议和任务，可以使用同步编码的方式编写异步并发编码
• 当我们必须使用一个将产生阻塞IO的调用时，可以把这个事件转移到线程池。虽然协程是在单线程内进行任务调度，但是也可以把多线程和多进程联系起来，进一步提高性能。
• 模仿threading模块中的同步原语，可以用在单线程内的协程之间。

asyncio异步IO高并发编程依然离不开三要素，即事件循环、回调模式(协程模式中又叫做驱动生成器/协程)和IO多路复用(epoll)。有很多优秀的框架都是基于asyncio模块的，比如说Tornado、gevent、twisted（scrapy、django channels）

二、低层级API

2.1 事件循环(event loop)

事件循环是每个 asyncio 应用的核心。 事件循环会运行异步任务和回调，执行网络 IO 操作，以及运行子进程。一个线程只能有一个事件循环。它实现了管理事件的所有功能。asyncio提供用于管理事件循环的方法如下：

2.1.1 获取事件循环:

• asyncio.get_running_loop()：返回当前 OS 线程中正在运行的事件循环。如果没有正在运行的事件循环则会引发 RuntimeError。 此函数只能由协程或回调来调用。
• asyncio.get_event_loop()：获取当前事件循环。如果当前OS线程为主线程，没有设置事件循环，并且没有调用set_event_loop()，asyncio将自动创建一个新的事件循环并将其设置为当前事件循环。
• asyncio.set_event_loop(loop)：将 loop 设置为当前 OS 线程的当前事件循环。
• asyncio.new_event_loop()：创建一个新的事件循环。

# 使用asyncio
import asyncio
import time


async def get_html(url):
    print("start get url")
    await asyncio.sleep(2)  # 返回一个future对象
    print("end get url")


if __name__ == '__main__':
    start_time = time.time()
    loop = asyncio.get_event_loop()
    tasks = [get_html("123") for i in range(10)]
    loop.run_until_complete(asyncio.wait(tasks))
    print(time.time() - start_time)

# 获取协程的返回值

import asyncio
import time
# 返回函数
from functools import partial


async def get_html(url):
    print("start get url")
    await asyncio.sleep(2)
    return "zzh"

def callback(url, future):
    print("send email to zzh {}".format(url))


if __name__ == '__main__':
    start_time = time.time()
    loop = asyncio.get_event_loop()
    # 使用ensure_future获取协程返回值
    get_future = asyncio.ensure_future(get_html("123"))
    """
    源码:
    if coroutines.iscoroutine(coro_or_future):
        if loop is None:
            loop = events.get_event_loop()
        task = loop.create_task(coro_or_future)
    一个线程只有一个loop。asyncio.ensure_future没有传入loop源码中会获取当前loop。
    然后在使用create_task获取返回值。
    """
    # loop.run_until_complete(get_future)
    # print(get_future.result())

    # 使用create_task获取值
    task = loop.create_task(get_html("123"))
    task.add_done_callback(partial(callback, "456"))
    loop.run_until_complete(task)
    print(task.result())

    # print(time.time() - start_time)

2.2 事件循环方法集

2.2.1 运行和停止循环

• loop.run_forever()：一直运行事件循环直到 stop() 被调用。
• loop.stop()： 停止事件循环。loop对象会传入到Task/Future对象中，所以通过任意一个Task/Future都可以停止loop。
• loop.is_running()：如果事件循环当前正在运行返回 True 。
• loop.is_closed()：如果事件循环已经被关闭，返回 True 。
• loop.close()：关闭事件循环。 当这个函数被调用的时候，循环必须处于非运行状态。pending状态的回调将被丢弃。此方法清除所有的队列并立即关闭执行器，不会等待执行器完成。
• loop.run_until_complete(future)：
  - 阻塞运行直到 future ( Future 的实例 ) 被完成。方法内部调用run_forever()，在future执行完毕后调用stop()
  - 如果参数是协程 ，将被隐式调度为asyncio.task来运行。
  - 返回Future的结果或者引发相关异常。

# loop会被放到future中
import asyncio
loop = asyncio.get_event_loop()
loop.run_forever()
# 源码 asyncio/base_events.py
# 运行完会执行_run_until_complete_cb方法
future.add_done_callback(_run_until_complete_cb)
def _run_until_complete_cb(fut):
    if not fut.cancelled():
        exc = fut.exception()
        if isinstance(exc, BaseException) and not isinstance(exc, Exception):
            # Issue #22429: run_forever() already finished, no need to
            # stop it.
            return
    futures._get_loop(fut).stop()

2.2.2调度回调

loop.call_soon(callback, *args, context=None)：

• 安排callback在事件循环的下一次循环时立即被调用。
• 回调按其注册顺序被调用。每个回调仅被调用一次。返回一个能用来取消回调的 asyncio.Handle 实例。
• 这个方法不是线程安全的。

loop.call_soon_threadsafe(callback, *args, context=None)：

• call_soon() 的线程安全变体。必须被用于安排来自其他线程 的回调。

2.2.3 调度延迟回调

loop.call_later(delay, callback, *args, context=None)：

• 安排 callback 在给定的延迟delay秒（可以是 int 或者 float）后被调用。
• callback 只被调用一次。如果两个回调被安排在同样的时间点，执行顺序未限定。

loop.call_at(when, callback, *args, context=None)：

• 安排 callback 在给定的绝对时间戳的时间 （一个 int 或者 float）被调用
• 使用与 loop.time() 同样的时间参考。这个函数的行为与 call_later() 相同。

import asyncio


def callback(sleep_time, loop):
    print("success time {}".format(sleep_time))


def stop_loop(loop):
    loop.stop()


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop_time = loop.time()
    loop.call_at(loop_time+2, callback, 2, loop)
    loop.call_at(loop_time+1, callback, 1, loop)
    loop.call_at(loop_time+3, callback, 3, loop)
    loop.call_soon(callback, 4, loop)

    # loop.call_soon(callback, 2, loop)
    # loop.call_soon(stop_loop, loop)
    loop.run_forever()

"""
success time 4
success time 2
success time 1
success time 2
success time 3
"""

2.2.4 创建 Futures 和 Tasks

• loop.create_future()：创建一个附加到事件循环中的 asyncio.Future对象。
• loop.create_task(coro, *, name=None)：安排一个协程的执行。返回一个 Task 对象。

2.2.5 在多线程或者多进程中执行代码

awaitable loop.run_in_executor(executor, func, *args)： 协程是单线程任务调度方案，一般不要在协程中加入阻塞代码，如果一定需要阻塞代码，可以和多线程和多进程结合起来完成整套解决方案，把费时的阻塞IO操作通过协程调度到线程池或进程池中。

• 安排在指定的executor中调用 func ，比如多线程池的ThreadPoolExecutor和多进程的ProcessPoolExecutor。
• 这个方法将线程池中的Future封装成 asyncio.Future对象并返回。

#使用多线程：在协程中集成阻塞io
import asyncio
from concurrent.futures import ThreadPoolExecutor
import socket
from urllib.parse import urlparse


def get_url(url):
    #通过socket请求html
    url = urlparse(url)
    host = url.netloc
    path = url.path
    if path == "":
        path = "/"

    #建立socket连接
    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    # client.setblocking(False)
    client.connect((host, 80)) #阻塞不会消耗cpu

    #不停的询问连接是否建立好， 需要while循环不停的去检查状态
    #做计算任务或者再次发起其他的连接请求

    client.send("GET {} HTTP/1.1\r\nHost:{}\r\nConnection:close\r\n\r\n".format(path, host).encode("utf8"))

    data = b""
    while True:
        d = client.recv(1024)
        if d:
            data += d
        else:
            break

    data = data.decode("utf8")
    html_data = data.split("\r\n\r\n")[1]
    print(html_data)
    client.close()


if __name__ == "__main__":
    import time
    start_time = time.time()
    loop = asyncio.get_event_loop()
    executor = ThreadPoolExecutor(3)
    tasks = []
    for url in range(20):
        url = "http://shop.projectsedu.com/goods/{}/".format(url)
        task = loop.run_in_executor(executor, get_url, url)
        tasks.append(task)
    loop.run_until_complete(asyncio.wait(tasks))
    print("last time:{}".format(time.time()-start_time))

2.2.6 Asyncio模拟HTTP请求

import asyncio
from urllib.parse import urlparse


async def get_url(url):
    url = urlparse(url)
    host = url.netloc
    path = url.path
    if path == "":
        path = "/"

    reader, writer = await asyncio.open_connection(host, 80)
    writer.write("GET {} HTTP/1.1\r\nHost:{}\r\nConnection:close\r\n\r\n".format(path, host).encode("utf8"))
    all_lines = []
    async for raw_line in reader:
        data = raw_line.decode("utf8")
        all_lines.append(data)
    html = "\n".join(all_lines)
    return html

async def main():
    tasks = []
    for url in range(20):
        url = "http://shop.projectsedu.com/goods/{}/".format(url)
        tasks.append(asyncio.ensure_future(get_url(url)))
    for task in asyncio.as_completed(tasks):
        result = await task
        print(result)

if __name__ == "__main__":
    import time
    start_time = time.time()
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())
    print('last time:{}'.format(time.time()-start_time))

2.3 Futures

2.3.1 Future相关函数

asyncio.isfuture(obj) 如果 obj 为下面任意对象，返回 True：

• 一个 asyncio.Future类的实例.
• 一个 asyncio.Task 类的实例.
• 带有 _asyncio_future_blocking 属性的类似 Future 的对象。
  asyncio.ensure_future(obj, *, loop=None) 返回：
• 如果 obj 是 Future、 Task 或 类似 Future 的对象( isfuture() 用于测试。)，返回obj对象并保持原样
• 如果 obj 是一个协程 (使用 iscoroutine() 进行检测)；在此情况下该协程将通过 ensure_future() 加入执行计划，返回一个封装了 obj 的 Task 对象。
• 如果 obj 是一个可等待对象( inspect.isawaitable() 用于测试)，返回obj 的 Task 对象。
  如果 obj 不是上述对象会引发一个 TypeError 异常。

import asyncio


async def get_html(url):
    print("start get url")
    await asyncio.sleep(2)
    return "success"

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    get_future = asyncio.ensure_future(get_html("liuchongyu.com"))
    task = loop.create_task(get_html("liuchongyu.com"))

    loop.run_until_complete(task)
    print(get_future.result())
    # print(task.result())

使用 asyncio.ensure_future()和loop.create_task()都可以创建一个task，我们查看asyncio.ensure_future()的源码中可以发现，如果没有传入loop，会自动获取事件循环一个线程只能有一个事件循环，所以和在外部手动获取事件循环是一样的，然后通过loop.create_task()创建一个task。

#  ###源码 asyncio/tasks.py
def ensure_future(coro_or_future, *, loop=None):
    if coroutines.iscoroutine(coro_or_future):
        if loop is None:
            loop = events.get_event_loop()
        task = loop.create_task(coro_or_future)
        if task._source_traceback:
            del task._source_traceback[-1]
        return task
    elif futures.isfuture(coro_or_future):
        if loop is not None and loop is not futures._get_loop(coro_or_future):
            raise ValueError('loop argument must agree with Future')
        return coro_or_future
    elif inspect.isawaitable(coro_or_future):
        return ensure_future(_wrap_awaitable(coro_or_future), loop=loop)
    else:
        raise TypeError('An asyncio.Future, a coroutine or an awaitable is '
                        'required')

所以，一般使用create_task()函数，它是创建新task的首选途径。

2.3.2 Future相关函数

class asyncio.Future(*, loop=None)： 一个 Future 代表一个异步运算的最终结果。非线程安全。Future 是一个 awaitable 对象。协程可以等待 Future 对象直到它们有结果或异常集合或被取消。

• result()：返回 Future 的结果。
• set_result(result)：将 Future 标记为 完成 并设置结果。
• set_exception：将 Future 标记为 完成 并设置一个异常。
• done()：如果 Future 为已 完成 则返回 True 。
• cancelled()：如果 Future 已 取消 则返回 True
• add_done_callback(callback, *, context=None)：添加一个在 Future 完成 时运行的回调函数。
• remove_done_callback(callback)：从回调列表中移除 callback 。
• cancel(msg=None)：取消 Future 并调度回调函数。
• exception()：返回 Future 已设置的异常。
• get_loop()：返回 Future 对象已绑定的事件循环。

三、高层级API

3.1 协程与任务

3.1.1 可等待对象(awaitable)

await语句中只能使用可等待对象awaitable。可等待对象有三种主要类型: 协程, Task和Future.

3.1.2 运行 asyncio 程序

asyncio.run(coro, *, debug=False) 执行 coroutine coro 并返回结果。 此函数会运行传入的协程，负责管理 asyncio 事件循环，终结异步生成器，并关闭线程池。 当有其他 asyncio 事件循环在同一线程中运行时，此函数不能被调用。 如果 debug 为 True，事件循环将以调试模式运行。 此函数总是会创建一个新的事件循环并在结束时关闭之。它应当被用作 asyncio程序的主入口点，理想情况下应当只被调用一次。

3.1.3 创建任务

asyncio.create_task(coro, *, name=None) 将 coro 协程 打包为一个 Task 排入协程准备执行。返回 Task 对象。 name 不为 None，它将使用Task.set_name()来设为任务的名称。 该任务会在 get_running_loop()返回的循环中执行，如果当前线程没有在运行的事件循环则会引发 RuntimeError。

3.1.4 休眠

coroutine asyncio.sleep(delay, result=None, *, loop=None)

• 阻塞 delay 指定的秒数。
• 如果指定了 result，则当协程完成时将其返回给调用者。
• sleep() 总是会挂起当前任务，以允许其他任务运行。

3.1.5 并发运行任务

asyncio.gather(*aws, loop=None, return_exceptions=False) 并发运行aws序列中的可等待对象。

• 如果 aws 中的某个可等待对象为协程，它将自动作为一个任务加入日程。
• 如果所有可等待对象都成功完成，结果将是一个由所有返回值聚合而成的列表。结果值的顺序与 aws 中可等待对象的顺序一致。
• 如果 return_exceptions 为 False (默认)，所引发的首个异常会立即传给等待 gather() 的任务。aws 序列中的其他可等待对象不会被取消并将继续运行。
• 如果 return_exceptions为 True，异常会和成功的结果一块处理，并聚合至结果列表。
• 如果 gather() 被取消，所有被提交 (尚未完成) 的可等待对象也会被取消。
• 如果 aws 序列中的任一 Task 或 Future 对象 被取消，它将被当作引发了 CancelledError一样处理 ——在此情况下 gather()调用不会被取消。这是为了防止一个已提交的 Task/Future 被取消的情况下，导致其他Tasks/Future也被取消。

和wait()方法类似，区别是gather更高级，传入的是awaitable的序列，还可以将分组后的序列传入，并且可以取消序列中的某个Task/gather。

3.1.6 屏蔽取消操作

awaitable asyncio.shield(aw, *, loop=None)

• 保护一个 awaitable 防止其被取消。
• 如果aw是一个协程，它将自动作为任务加入日程。

3.1.7 超时

coroutine asyncio.wait_for(aw, timeout, *, loop=None)：

• 等待 aw 完成，指定 timeout 秒数后超时。
• 如果 aw 是一个协程，它将自动作为Task加入日程。
• timeout 可以为 None，也可以为 float 或 int 型数值表示的等待秒数。如果 timeout 为 None，则等待直到完成。
• 如果发生超时，任务将取消并引发 asyncio.TimeoutError.
• 要避免任务 取消，可以加上 shield() 。
• 此函数将等待直到 Future 确实被取消，所以总等待时间可能超过 timeout。 如果在取消期间发生了异常，异常将会被传播。
• 如果等待被取消，则 aw 指定的对象也会被取消。

3.1.8 简单等待(Wait)

asyncio.wait(aws, *, loop=None, timeout=None, return_when=ALL_COMPLETED)

• 并发运行aws指定的awaitable，并阻塞线程直到满足 return_when指定的条件。
• aws集必须不为空。
• 返回两个 Task/Future 集合: (done, pending)。
  done, pending = await asyncio.wait(aws)

3.1.9 Task

class asyncio.Task(coro, *, loop=None, name=None) Task是futures.Future 的子类，被用来在事件循环中运行协程，是Future和协程之间的一个桥梁，封装了一些之前操作协程来生成Future的一些方法。

比如说，我们在定义一个协程后，在驱动这个协程前需要自己使用next()或者send(None)预激协程，Task中则将send(None)方法封装起来自动调用。

再比如说，在Future中，需要捕捉StopIteration异常，并将异常值用set_result()方法放到Future中，在线程池中是submit方法实现这个功能的，而Task则将这个过程封装进来自动调用。

# 源码				
		def __step(self, exc=None):
    # ...
    try:
        if exc is None:
            result = coro.send(None)						
        else:
            result = coro.throw(exc)
    except StopIteration as exc:
            if self._must_cancel:
                self._must_cancel = False
                super().set_exception(futures.CancelledError())
            else:
                super().set_result(exc.value)

3.1.10Task的状态：

• Pending：创建future，还未执行
• Running：事件循环正在调用执行任务
• Done：任务执行完毕
• Cancelled：Task被取消后的状态

asyncio.Task从Future 继承了其除Future.set_result()和Future.set_exception()以外的所有 API。

cancel(msg=None) 取消一个正在运行的 Task 对象可使用 cancel() 方法。 下一轮事件循环中抛出一个 CancelledError 异常给被封包的协程。。如果取消期间一个协程正在等待一个 Future 对象，该Future对象也将被取消。

async def delay(sleep_times):
    print("waiting")
    await asyncio.sleep(sleep_times)
    print("done after {}s".format(sleep_times))

if __name__ == '__main__':
    task1 = delay(1)
    task2 = delay(2)
    task3 = delay(3)
    tasks = [task1, task2, task3]

    loop = asyncio.get_event_loop()
    try:
        loop.run_until_complete(asyncio.wait(tasks))
    except KeyboardInterrupt as e:
        all_task = asyncio.Task.all_tasks()
        for task in all_task:
            print("cancel task")
            print(task.cancel())
        # run_until_complete()方法在所有task执行完成后会自动调用stop，
        # 但是如果使用cancel取消掉task，就不能自动执行stop()方法，所以需要手动运行
        loop.stop()
        # 如果线程中没有时间循环处于panding状态会报错
        loop.run_forever()
    finally:
        loop.close()

• add_done_callback(callback, *, context=None) 添加一个回调，将在 Task 对象 完成 时被运行。此方法应该仅在低层级的基于回调的代码中使用。callback传入的是函数名，如果想要传入参数，可以使用偏函数partial将回调函数和参数封装。

3.2 Steam

Steam是用于处理网络连接的高级 async/await-ready原语。Steam允许发送和接收数据，而不需要使用回调或低级协议和传输。

3.2.1 Stream 函数

asyncio.open_connection(host=None, port=None, *, loop=None, limit=None, ssl=None, family=0, proto=0, flags=0, sock=None, local_addr=None, server_hostname=None, ssl_handshake_timeout=None)

• asyncio.open_connection()是一个协程，用来建立网络连接并返回一对 (reader, writer) 对象。
• 返回的 reader和writer 对象是 StreamReader 和 StreamWriter 类的实例。
• loop 参数是可选的，当从协程中等待该函数时，总是可以自动确定。
• limit 确定返回的 StreamReader 实例使用的缓冲区大小限制。默认情况下，limit 设置为 64 KiB 。
• 其余的参数直接传递到 loop.create_connection() 。

reader, writer = await asyncio.open_connection(host, 80)

使用asyncio.open_connection()我们依然需要像使用回调模式中那样，建立socket连接 > 设置为非阻塞IO > register到epoll/select中监听其IO状态，这在asyncio.open_connection()都实现了。

coroutine asyncio.start_server(client_connected_cb, host=None, port=None, *, loop=None, limit=None, family=socket.AF_UNSPEC, flags=socket.AI_PASSIVE, sock=None, backlog=100, ssl=None, reuse_address=None, reuse_port=None, ssl_handshake_timeout=None, start_serving=True)

• 启动套接字服务。
• 当一个新的客户端连接被建立时，回调函数 client_connected_cb 会被调用。该函数会接收到一对参数(reader, writer)，reader是类 StreamReader的实例，而writer是类 StreamWriter的实例。
• client_connected_cb即可以是普通的可调用对象也可以是一个 协程函数; 如果它是一个协程函数，它将自动作为 Task 被调度。
• loop 参数是可选的。当在一个协程中await该方法时，该参数始终可以自动确定。
• limit 确定返回的 StreamReader 实例使用的缓冲区大小限制。默认情况下，limit 设置为 64 KiB 。
• 余下的参数将会直接传递给 loop.create_server().

3.2.2 StreamReader

asyncio.StreamReader 这个类表示一个读取器对象，该对象提供api以便于从IO流中读取数据。不推荐直接实例化 StreamReader 对象，建议使用 open_connection() 和 start_server()来获取 StreamReader 实例 reader 。

• reader.read(n=-1)

  • 至多读取 n 个byte。 如果没有设置 n , 则自动置为 -1 ， -1时表示读至 EOF 并返回所有读取的byte。
  • 如果读到EOF，且内部缓冲区为空，则返回一个空的 bytes 对象。

• readline()

  • 读取一行，其中“行”指的是以 \n 结尾的字节序列。
  • 如果读到EOF而没有找到 \n ，该方法返回部分读取的数据。
  • 如果读到EOF，且内部缓冲区为空，则返回一个空的 bytes 对象。

• readexactly(n)

  • 精确读取 n 个 bytes，不会超过也不能少于。
  • 如果在读取完 n 个byte之前读取到EOF，则会引发 IncompleteReadError 异常。使用 IncompleteReadError.partial 属性来获取到达流结束之前读取的 bytes 字符串。

• readuntil(separator=b'\n')

  • 从流中读取数据直至遇到 separator
  • 成功后，数据和指定的separator将从内部缓冲区中删除(或者说被消费掉)。返回的数据将包括在末尾的指定separator。
  • 如果读取的数据量超过了配置的流限制，将引发 LimitOverrunError 异常，数据将留在内部缓冲区中并可以再次读取。
  • 如果在找到完整的separator之前到达EOF，则会引发 IncompleteReadError 异常，并重置内部缓冲区。 IncompleteReadError.partial属性可能包含指定separator的一部分。

3.3 StreamWriter

asyncio.StreamWriter 这个类表示一个写入器对象，该对象提供api以便于写数据至IO流中。不推荐直接实例化 StreamReader 对象，建议使用 open_connection() 和 start_server() 来获取 StreamReader 实例 reader 。

• write(data)
  • 该方法尝试立即将数据写入基础套接字。 如果失败，则数据将在内部写缓冲区中排队，直到可以发送为止。
  • write(data)方法中封装了send()方法，即可以直接传入http请求头
  • 该方法应与drain()方法一起使用：

writer.write("GET {} HTTP/1.1\r\nHost:{}\r\nConnection:close\r\n\r\n".format(path, host).encode("utf8"))
stream.write(data)
await stream.drain()

• writelines(data)
  • 该方法立即将字节列表（或任何可迭代的字节）写入基础套接字。 如果失败，则数据将在内部写缓冲区中排队，直到可以发送为止。
  • 该方法应与drain()方法一起使用：

stream.writelines(lines)
await stream.drain()

• close()
  • 该方法关闭steam和基础套接字。
  • 该方法应与wait_closed()方法一起使用：

stream.close()
await stream.wait_closed()

• can_write_eof()：如果基础传输支持write_eof()方法，则返回True，否则返回False。
  -write_eof()：刷新缓冲的写入数据后关闭writer。
• transport：返回基础异步传输。
• get_extra_info(name, default=None)：访问可选的传输信息；
  -coroutine drain()：等待，直到适合继续写入流为止。

writer.write(data)
await writer.drain()

这是一种与基础IO写缓冲区交互的流控制方法。 当缓冲区的大小达到高水平线时，drain()会阻塞，直到缓冲区的大小排空到低水平线为止，然后才能恢复写入。 当没有什么可等待的时，drain()立即返回。

• is_closing()：如果steam已关闭或正在关闭，则返回True。
• coroutine wait_closed()：等待，直到steam关闭。应该在close()之后调用，以等待基础连接关闭。

四、协程嵌套

import asyncio

async def computer(x, y):
    print("Computer %s + %s ..." % (x, y))
    await asyncio.sleep(1)
    return x + y

async def print_sum(x, y):
    result = await computer(x, y)
    print("%s + %s = %s" % (x, y, result))

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(print_sum(1, 2))
    loop.close()

task可以看作是调用方，print_num是委托生成器，computer是子生成器，Task 对象被用来在事件循环中运行调度协程。

• asyncio.get_event_loop()：创建一个事件循环
• loop.run_until_complete(print_sum(1, 2))：将print_sum(1, 2)这个协程注册到事件循环中，因为参数是一个协程对象，所以创建一个asyncio.Task来运行协程。
• Task进入挂起(pending)状态，协程print_sum(1, 2)开始运行(running)，运行到await computer(1, 2)时协程暂停(suspended)
• 协程computer(x, y)开始运行(running)，运行到await asyncio.sleep(1)协程暂停(suspended)
• 执行await asyncio.sleep(1)，获得资源后Future对象直接返回给调用方Task，Task再返回给事件循环
• 事件循环接收到Task返回的Future对象后，执行这个Future，即asyncio.sleep(1)
• 事件循环再次启动Task，因为Task已经和协程computer(x, y)建立双向通道，computer(x, y)继续运行(running)await asyncio.sleep(1)后面的代码return x + y，执行完后computer(x, y)停止(done)，返回给print_sum(1, 2)
• print_sum(1, 2)继续运行(running)，接收子生成器抛出的StopIteration(3)异常，并提取异常值3，然后执行await下面的代码print("%s + %s = %s" % (x, y, result))，执行完后print_sum(1, 2)停止(done)，返回给Task
• Task接收委托生成器返回的StopIteration()异常，Task(停止)，返回给事件循环
• 事件循环接收后，停止(stopped)

五、asyncio模拟http请求

使用IO多路复用+回调模式+事件循环完成http请求，为了解决

• 回调模式编码复杂度高
• 同步编程并发性低
• 多线程编程需要线程同步，即需要加入锁导致性能下降

为了解决回调之痛，引出了协程，我们现在尝试用基于协程的asyncio，即IO多路复用+协程+事件循环的模式来模拟http请求。

原生asyncio目前没有提供http协议的接口，提供的是更底层的tcp/udp，基于aiohttp框架是专门用来作http请求的。

import asyncio
import time

async def get_url(url):
    # 解析url
    url = urlparse(url)
    host = url.netloc
    path = url.path
    if path == "":
        path = "/"

    # 建立socket连接
    reader, writer = await asyncio.open_connection(host, 80)
    writer.write("GET {} HTTP/1.1\r\nHost:{}\r\nConnection:close\r\n\r\n".format(path, host).encode("utf8"))

    all_lines = []
    async for raw_line in reader:
        data = raw_line.decode("utf-8")
        all_lines.append(data)

    html = "\n".join(all_lines)
    # print(html)
    return html


async def main():
    # 将所有任务打包成Future/Task对象放入tasks中
    tasks = []
    for url in range(20):
        url = "http://shop.projectsedu.com/goods/{}/".format(url)
        tasks.append(asyncio.ensure_future(get_url(url)))
    # 调用as_completed()返回一个生成器，首先找出调用此方法时就已经执行完成或者取消的Future/Task
    for task in asyncio.as_completed(tasks):
        result = await task
        print(result)

if __name__ == '__main__':
    start_time = time.time()

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

    print("last time:{}".format(time.time()-start_time))

六、asyncio的同步

asyncio是基于单线程的协程，是不涉及到GIL锁的问题，在协程中，除await语句外，其他语句都是在CPU中计算，在当前协程一旦开始执行就会顺序执行完成后才会切换到另一个协程中，因此一般不需要担心同步问题。

但是有时候会遇到两个协程都要await到同一个子协程中，向这个子协程请求一个结果，如果子协程是一个高耗时的IO操作，就会导致这两个协程都要各自请求一次这个高耗时的IO操作，这就涉及到协程间的同步问题。

如下面这个例子，协程get_stuff()是请求一个url获得返回值的高耗时IO操作。当parse_stuff()协程向get_stuff()发出请求解析某个url的请求后，get_stuff()开始执行这个url的下载请求，在stuff = await aiohttp.request(‘GET’, url)处暂停，继续其他协程的执行，这时parse_stuff()也请求了同一个url的获取请求，因为上一次对这个url的请求还没有返回，所以cache中还没有返回值，所以会再次发起一次stuff = await aiohttp.request(‘GET’, url) 请求。这样会耗费大量时间。如果在get_stuff(url)中加锁，一个协程请求后，只能在执行完释放锁后，另一个协程才可以再次请求。

import asyncio
import aiohttp
from asyncio import Lock, Queue

cache = {}
lock = Lock()

async def get_stuff(url):
    # with await lock:
    # await lock.acquire()
    async with lock:
        if url in cache:
            return cache[url]
        stuff = await aiohttp.request('GET', url)
        cache[url] = stuff
        return stuff

async def parse_stuff():
    stuff = await get_stuff()
    # do some parsing

async def use_stuff():
    stuff = await get_stuff()
    # use stuff to do something interesting

if __name__ == '__main__':
    tasks = [parse_stuff(), use_stuff()]
    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.wait(tasks))

asyncio也为我们提供了一系列同步机制，是程序员级别的锁，不深入到操作系统中去。

• Lock： await lock.acquire()锁定并执行这个协程，lock.release()解锁，因为acquire()方法肩负执行协程的任务，必须异步执行，所以是一个协程。

# 源码
class Lock(_ContextManagerMixin):
    # ...
    async def acquire(self):
        if not self._locked and all(w.cancelled() for w in self._waiters):
            self._locked = True
            return True
        # 创建Future
        fut = self._loop.create_future()
        # 把Future加入执行队列中
        self._waiters.append(fut)

        try:
            try:
                # 异步等待fut的完成
                await fut
            finally:
                # 完成后将fut从队列中移除
                self._waiters.remove(fut)
        except futures.CancelledError:
            if not self._locked:
                self._wake_up_first()
            raise

        # 将判断是否获取锁的标志_locked置为True
        self._locked = True
        return True

    def release(self):
        if self._locked:
            # 将判断是否获取锁的标志_locked置为False
            self._locked = False
            self._wake_up_first()
        else:
            raise RuntimeError('Lock is not acquired.')
    # ...

七、asyncio的通信

多线程中的Queue内部使用Condition会发生阻塞，当队列已满put()就会阻塞，当队列已空，get()就会阻塞。异步编程中不能存在阻塞，所以不能使用多线程中的Queue完成通信，需要使用asyncio中提供的Queue。 asyncio中的Queue中的接口和多线程中的是一样的，但是其中put和get方法实现了协程。 asyncio中的Queue可以控制最大长度，即限流，如果没有限流的需求，可以在单线程中申请一个全局的List完成通信

# 源码 asyncio/queues.py
class Queue:
    async def put(self, item):
            """
            将项目放入队列。 如果队列已满，请等到空闲插槽可用后再添加项目。
            """
            while self.full():
                putter = self._loop.create_future()
                self._putters.append(putter)
                try:
                    await putter
                except:
                    putter.cancel()  # Just in case putter is not done yet.
                    try:
                        # Clean self._putters from canceled putters.
                        self._putters.remove(putter)
                    except ValueError:
                        # The putter could be removed from self._putters by a
                        # previous get_nowait call.
                        pass
                    if not self.full() and not putter.cancelled():
                        # We were woken up by get_nowait(), but can't take
                        # the call.  Wake up the next in line.
                        self._wakeup_next(self._putters)
                    raise
            return self.put_nowait(item)


    async def get(self):
            """
            如果队列为空，请等待直到有一个项目可用。
            """
            while self.empty():
                getter = self._loop.create_future()
                self._getters.append(getter)
                try:
                    await getter
                except:
                    getter.cancel()  # Just in case getter is not done yet.
                    try:
                        # Clean self._getters from canceled getters.
                        self._getters.remove(getter)
                    except ValueError:
                        # The getter could be removed from self._getters by a
                        # previous put_nowait call.
                        pass
                    if not self.empty() and not getter.cancelled():
                        # We were woken up by put_nowait(), but can't take
                        # the call.  Wake up the next in line.
                        self._wakeup_next(self._getters)
                    raise
            return self.get_nowait()