python协程

协程本质上就是一个线程，不过它是协作式的非抢占式的程序，面向的是IO操作。python有GIL的限制，不能充分利用多线程的实现高并发。进程和线程都是通过cpu的调度实现不同任务的有序执行，协程则要在写代码的时候确定执行顺序。由于协程在一个线程中，所以协程不能阻塞。

优缺点：

无需线程上下文切换的开销
在一个线程中，不需要加锁
无法利用多核资源：协程的本质是单线程，需要和进程配合才能运行在多CPU上
进行阻塞(Blocking)操作(如IO时)会阻塞掉整个程序
python协程的发展时间较长：

python2.5 为生成器引用.send()、.throw()、.close()方法
python3.3 为引入yield from，可以接收返回值，可以使用yield from定义协程
Python3.4 加入了asyncio模块
Python3.5 增加async、await关键字，在语法层面的提供支持
python3.7 使用async def + await的方式定义协程
此后asyncio模块更加完善和稳定，对底层的API进行的封装和扩展
python将于 3.10版本中移除 以yield from的方式定义协程 （目前版本是3.9.1）
由于asyncio每个版本都会新增功能，对一些旧的底层的API进行封装，极大地方便的使用者，但正因为此，网上有很多教程使用的接口官方已经不建议直接使用，应该改而使用更加高级的API，所以在这里记录一下如何使用这些API。

简单例子
要点：

使用async def的形式定义
在协程中可以使用await关键字，注意其后跟的是"可等待对象"(协程, 任务 和 Future)
协程不能直接执行，需要在asyncio.run()中执行，也可以跟在await后面
async和await这两个关键字只能在协程中使用

import asyncio


async def foo(name):

	await asyncio.sleep(1)      # 这是一个不会阻塞的sleep，是一个协程
	print(f"name = {name}")


async def main():
	# 协程本身就是一个可等待对象
	await foo("lczmx")  # 执行协程
	print("done")

if __name__ == '__main__':
	# 使用asyncio.run运行
	asyncio.run(main())
asyncio.run(main, *, debug=False)方法就是对run_until_complete进行了封装：
loop = events.new_event_loop()
return loop.run_until_complete(main)

关于可等待对象
可等待对象(awaitable)是能在 await 表达式中使用的对象。可以是 协程 或是具有__await__() 方法的对象。

那么协程是如何成为可等待对象的呢？

collections.abc.Awaitable类，这是为可等待对象提供的类，可被用于 await 表达式中

class Awaitable(metaclass=ABCMeta):
	__slots__ = ()

	@abstractmethod
	def __await__(self):	# __await__方法必须返回一个 iterator
		yield

	@classmethod
	def __subclasshook__(cls, C):
		if cls is Awaitable:
			return _check_methods(C, "__await__")
		return NotImplemented
1
2
3
4
5
6
7
8
9
10
11
12

用async def复合语句创建的函数，它返回的是一个Coroutine对象，而Coroutine继承Awaitable。
并发
使用协程进行并发操作
方法一
使用asyncio.create_task(coro)方法，返回一个Task对象，Task类继承Future，在python3.7以下版本中使用asyncio.ensure_future(coro_or_future)。

import asyncio


async def foo(char:str, count: int):
    for i in range(count):
        print(f"{char}-{i}")
        await asyncio.sleep(.5)


async def main():
    task1 = asyncio.create_task(foo("A", 2))
    task2 = asyncio.create_task(foo("B", 3))
    task3 = asyncio.create_task(foo("C", 2))

    await task1
    await task2
    await task3


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

执行结果

A-0
B-0
C-0
A-1
B-1
C-1
B-2
方法二
使用asyncio.gather()方法，其内部调用的是asyncio.ensure_future()方法

import asyncio


async def foo(char:str, count: int):
    for i in range(count):
        print(f"{char}-{i}")
        await asyncio.sleep(.5)


async def main():

    await asyncio.gather(foo("A", 2), foo("B", 3), foo("C", 2))

if __name__ == '__main__':
    asyncio.run(main())
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

回调、返回值等操作
要完成这些功能需要Task对象，即asyncio.create_task()的返回值。由于Task继承Future，实现了除Future.set_result() 和 Future.set_exception()外的全部API，而asyncio.Future模仿的是 concurrent.futures.Future类，所以Task很多方法和 在使用线/进程池时用到的方法类似（有细微差别）。

Task的方法，见官方文档
使用回调函数和取得返回值的例子：

import asyncio


def callback(future):
    # 唯一参数是一个Task对象
    # print(type(future))     # <class '_asyncio.Task'>

    print(future)
    # <Task finished name='Task-2' coro=<foo() done, defined at E: ... xxx.py:11> result=123>

    print(future.result())      # 123   # 接收返回值
    print(future.get_name())    # foo


async def foo():
    print("running")
    return 123


async def main():

task = asyncio.create_task(foo(), name="foo")   # name形参3.8及以上版本可用
task.add_done_callback(callback)                # 添加回调函数
await task
1
2
3

if name == ‘main’:
asyncio.run(main())
与线程结合
我们知道，协程本身就只有一个线程，假如这协程阻塞了，那么整个程序也就阻塞了。为此我们在执行一些必然会产生阻塞的代码时，可以把代码放入到其它线程/进程中，这样可以继续执行协程的其它代码了。

方法一
coroutine asyncio.to_thread(func, /, args, **kwargs)
这是python3.9的新方法，3.9以下版本看方法二
在不同的线程中异步地运行函数 func。向此函数提供的任何args 和 **kwargs 会被直接传给 func。其返回值是一个协程，所以假如有回调等操作，使用asyncio.create_task(coro)方法，再调用Task对象的方法。

import asyncio
import time


def block_func(name: str):
    time.sleep(2)       # 模拟阻塞时间
    print(f"name = {name}")


async def foo():
    # 一个协程
    print("async foo")
    await asyncio.sleep(1)


async def main():
    await asyncio.gather(
        asyncio.to_thread(block_func, name="lczmx"),
        foo()
    )

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

方法二
awaitable loop.run_in_executor(executor, func, *args)
安排在指定的执行器(线/进程池)中调用 func。该方法的返回值awaitable对象，其实就是一个asyncio.Future对象。这个方法使用起来也比较简单，不过要注意传参方式：位置参数可以直接传入，而关键字参数需要使用functools.partial()。

from concurrent.futures import ThreadPoolExecutor
from functools import partial
import asyncio
import time


def block_func(name: str, age: int):
    time.sleep(2)  # 模拟阻塞时间
    print(f"name = {name}, age = {age}")


async def foo():
    # 一个协程
    print("async foo")
    await asyncio.sleep(1)


async def main():
    loop = asyncio.get_running_loop()

    with ThreadPoolExecutor(5) as pool:
        task = loop.run_in_executor(pool, partial(block_func, "lczmx", age=18))
        # task可以添加回调等操作

    await asyncio.gather(foo(), task)


if __name__ == '__main__':
    asyncio.run(main())
21
22
23
24
25
26
27
28
29

这个两个方法的关系
asyncio.to_thread()方法实质上就是对loop.run_in_executor()方法进行了封装：

async def to_thread(func, /, *args, **kwargs):
    loop = events.get_running_loop()
    ctx = contextvars.copy_context()
    func_call = functools.partial(ctx.run, func, *args, **kwargs)
    return await loop.run_in_executor(None, func_call)
1
2
3
4
5

假如B线程想要把协程c放入到A线程的事件循环中执行，使用syncio.run_coroutine_threadsafe(coro_func(), loop)方法，具体怎么使用见范例

异步上下文管理器
异步上下文管理器使用的是async with语法， 是 上下文管理器 的一种，它能够在其__aenter__ 和 aexit 方法中暂停执行。使用异步上下文管理器之前，可以先看一看一般的上下文管理器，类比过来。

aenter(self)
在语义上类似于 enter()，仅有的区别是它必须返回一个 可等待对象。
aexit(self, exc_type, exc_value, traceback)
在语义上类似于 exit()，仅有的区别是它必须返回一个 可等待对象。
一个简单的例子：

import asyncio


class AContext:
    def __init__(self):
        print("init running")

    async def __aenter__(self):
        print("aenter running")

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print("aexit running")


async def main():
    async with AContext() as ac:
        print("AContext", ac)   # AContext None	# __aenter__ 没有return东西


if __name__ == '__main__':
    print("start")
    asyncio.run(main())
21
22

使用标准库
除了上述方法可以实现异步上下文管理器外，还可python基础教程以使用contextlib.asynccontextmanager装饰器+yield实现，yield前面的代码对应__aenter__，其后的代码对应__aexit__。

import contextlib
import asyncio


# 加上装饰器
@contextlib.asynccontextmanager
async def foo():
    try:
        # 进行初始化
        yield "返回你要操作的对象"
    finally:
        # 处理释放资源等操作
        pass


async def main():
    async with foo() as f:
        print(f)


if __name__ == "__main__":
    asyncio.run(main())
21
22

异步迭代器
异步迭代器使用的时async for语法，和常规的for表达式一样， async for也有一个可选的else 分句。异步迭代器有以下几个要求：

必须实现__aiter__方法，该方法返回一个异步迭代器对象。
异步迭代器对象必须实现__anext__方法，该方法返回一个awaitable类型的值。
为了停止迭代，__anext__必须抛出一个StopAsyncIteration异常。
跟迭代器协议非常类似。
例子：

class AsyncIteratorWrapper:
    def __init__(self, obj):
        self._it = iter(obj)

    def __aiter__(self):
        return self

    async def __anext__(self):
        try:
            value = next(self._it)
        except StopIteration:
            raise StopAsyncIteration
        return value

async for letter in AsyncIteratorWrapper("abc"):
    print(letter)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

综合例子
非阻塞爬虫

from contextlib import asynccontextmanager
from concurrent.futures import ThreadPoolExecutor
from functools import partial
import requests
import asyncio
import time


class ASyncRequests:
    """非阻塞的requests"""

    def __init__(self, count):
        """
        :param count: 线程池大小
        """
        assert isinstance(count, int)
        self.pool = ThreadPoolExecutor(count)  # 创建线程池
        self.loop = asyncio.get_running_loop()  # 获得事件循环
        if not self.loop:
            raise RuntimeError("event loop为None")

        # self.all_future 用于装run_in_executor返回的future对象
        #                  python3.9版本用to_thread

        self.all_future = []

    def get(self, url, callback, **kwargs):
        self.requests(url, "get", callback, **kwargs)

    def post(self, url, callback, **kwargs):
        self.requests(url, "post", callback, **kwargs)

    def requests(self, url, method, callback, **kwargs):
        """
        写了两种方法，第一种适合版本python，另一种适合最新版的python
        """

        # python3.5及以上版本可用
        # run_in_executor:
        #   把阻塞的函数放入线程池中执行
        #   返回一个asyncio.Future对象
        #   传值时，利用 偏函数 传关键字参数

        result = self.loop.run_in_executor(self.pool, partial(requests.request,
                                                              url=url, method=method, **kwargs))
        result.add_done_callback(callback)  # 回调
        self.all_future.append(result)

        # python3.9及以上版本可用
        # to_thread:
        #   使用其它的线程，执行阻塞函数
        #   返回一个协程
        #   除第一个参数外的所有参数都会给func

        # coro = asyncio.to_thread(requests.request, url=url, method=method, **kwargs)
        # task = asyncio.create_task(coro)
        # task.add_done_callback(callback)            # 回调
        # self.all_future.append(task)


def process_get(future):
    """回调函数，处理返回值"""

    print(type(future))  # <class '_asyncio.Future'>

    # future.result()的返回值就是Response对象
    print(future.result().status_code)  # 200


@asynccontextmanager
async def request_session(count=5):
    try:
        async_req = ASyncRequests(count)
        yield async_req
    finally:
        await asyncio.gather(*async_req.all_future)


async def main():
    async with request_session(10) as session:
        session.get("https://www.baidu.com/", callback=process_get)
        session.get("https://www.sogo.com/", callback=process_get)
        session.get("https://cn.bing.com/", callback=process_get)
        session.get("https://www.cnblogs.com/lczmx", callback=process_get)


def block_req():
    requests.get("https://www.baidu.com/")
    requests.get("https://www.sogo.com/")
    requests.get("https://cn.bing.com/")
    requests.get("https://www.cnblogs.com/lczmx")


if __name__ == '__main__':
    start = time.time()
    asyncio.run(main())
    end = time.time()
    print("使用协程加线程：总用时: %f sec" % (end - start))

    block_req()
    print("直接访问：总用时: %f sec" % (time.time() - end))
    # 访问数越多，协程的优势就越大
    # 访问同一个网站过多可能会报错 Connection aborted.', ConnectionResetError(10054
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103