本网站相关文章:

变化历史

版本

首先, 关于协程的优势已经在同步编程和异步编程中介绍过, 这里不再阐述. 那么, Python 中协程的演变是怎样的呢?

版本 详细说明
2.2 引入生成器作为一种语言特性，通过yield语句实现基础协程功能。
2.5 引入greenlet库，通过手动控制协程切换实现更灵活的协程编程。
3.2 引入yield from语法，简化协程编程的嵌套和异常处理。
3.3 引入asyncio库，提供异步编程的基础设施，包括协程、事件循环和异步 IO 操作。
3.4 引入async和await关键字，作为定义协程函数和在协程中进行异步操作的语法糖。
3.5 引入asyncio的async和await的原生语法支持。
3.6 引入asyncio的asyncio.run()函数，简化异步代码的入口点。
3.7 引入asyncio的asyncio.create_task()函数，简化任务的创建和管理。
3.8 引入asyncio的asyncio.get_running_loop()函数，获取当前运行的事件循环。
3.9 引入asyncio的asyncio.to_thread()函数，将阻塞操作移动到线程池中执行。
3.10 引入asyncio的asyncio.all_tasks()函数，获取所有当前运行的任务。

请注意，这些版本是根据 Python 的发布历史作出的概括，某些功能的添加和改进可能在不同的小版本中发生。其中 yield from 和 asyncio 库实际上存在时间极其短暂, 因为python3.0~python3.4提出阶段, 使用 python3 的开发者实际上不是非常多.

yield 和 await

yield from和await是用于在协程中等待另一个协程或可迭代对象的关键字，它们有一些区别和不同的语法使用：

  1. 语法差异：yield from是 Python 3.3 引入的语法，而await是 Python 3.5 引入的语法。yield from使用在生成器中，而await使用在异步函数（包含async def声明的函数）中。

  2. 表达方式：yield from用于委派生成器，它将控制权委托给另一个生成器，并能从中获取生成的值。await用于等待可等待对象，它能挂起当前协程的执行，并等待另一个协程或异步操作的结果。

  3. 功能扩展：await语法更加灵活，它可以等待各种可等待对象，包括协程、异步函数、异步生成器和其他实现了__await__特殊方法的对象。而yield from则主要用于在生成器中委派子生成器。

  4. 异常处理：yield from能够自动处理委派生成器和子生成器之间的异常，异常会被透明地传递给委派生成器的调用方。而await使用try/except块来捕获和处理异常，通过try/except语句块包裹await表达式来处理可能发生的异常。

总体来说，yield from和await都用于在协程中等待其他的可迭代对象或协程，但在语法和使用方式上有一些差异。await更加通用且灵活，是 Python 3.5 及以上版本中推荐使用的协程等待语法。

生成器和协程

  • a. 用途不同：生成器主要用于生成一系列的值，可以通过yield语句逐个产生值。它们通常用于迭代器的实现和惰性计算。协程则更加强调并发和异步编程，允许在协程之间进行切换和暂停，以便处理异步任务。
  • b. 调用方式不同：生成器通过next()函数或迭代器协议进行迭代，每次调用next()会执行生成器中的代码，直到遇到yield语句停止。协程通过事件循环或异步框架调度运行，可以被暂停、恢复和取消。
  • c. 状态维护不同：生成器维护自己的状态，每次从yield语句继续执行时会从上次停止的位置继续。协程可以通过await语句暂停执行，并保留当前的上下文状态，下次恢复时会继续执行暂停的位置。
  • d. 通信方式不同：生成器主要通过产生和消费值的方式进行通信，通过yield语句从生成器中产生值，并可以通过函数参数或外部迭代器向生成器传递值。协程则更加灵活，可以在协程之间进行双向通信，通过await语句等待其他协程的结果，并通过send()方法向协程发送值。

需要注意的是，Python 中的协程通常是基于生成器实现的，通过使用yield语句来实现协程的暂停和恢复。因此，协程可以看作是生成器的一种特殊形式，但协程拥有更多的特性和能力，用于处理并发和异步编程。

  1. 那么, 为何协程可以看成是生成器的一种特殊形式呢? 为何不直接使用 async/await 来实现协程呢?

实际上，Python 3.5 及以上版本引入了asyncio模块，引入了新的async和await关键字，用于定义和管理协程。async和await提供了更简洁和直观的语法来编写和管理协程，使得异步编程更加易于理解和维护。

在早期版本的 Python 中，协程功能并没有原生支持。因此，通过使用生成器和yield语句来模拟协程是一种常见的做法。生成器提供了协程所需的暂停和恢复执行的能力，但语法相对较为复杂。

随着 Python 的发展，引入了async/await关键字，使得协程的编写和使用更加直观和易于理解。使用async/await，我们可以更清晰地定义异步任务和操作，而不需要手动管理生成器和yield语句。

因此，现代的 Python 代码通常会使用async/await来定义和管理协程，而不是直接使用生成器和yield语句。这样可以使代码更加简洁、易读和易于维护。但对于早期版本的 Python，使用生成器和yield语句来模拟协程仍然是一种有效的方式。

  1. 一个早期的通过 yield 实现的协程案例如下:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
import time

# 定义一个协程函数
def countdown(name, n):
    print(f'{name} started')
    while n > 0:
        print(f'{name}: T-minus {n}')
        yield
        time.sleep(1)
        n -= 1
    print(f'{name} finished')

# 创建两个协程实例
coroutine1 = countdown('Coro1', 5)
coroutine2 = countdown('Coro2', 3)

# 模拟协程的调度器
def scheduler(*coroutines):
    while True:
        for coroutine in coroutines:
            try:
                next(coroutine)  # 恢复协程的执行
            except StopIteration:
                coroutines.remove(coroutine)
        if not coroutines:
            break

# 调度器调度协程的执行
scheduler(coroutine1, coroutine2)

yield

yield 用法

  1. 利用 yield 实现一个递减技术迭代器
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
def iterator_yield(num):
    """ yield 组成的迭代器, 利用yield 不但返回, 一般来说这里value是一个IO操作 """
    value = num
    while value > 0:
        value = value - 1
        yield value
    # 返回空的时候for会自动处理异常
    return


if __name__ == '__main__':
    # 1. 测试yield组成的迭代器
    for i in iterator_yield(5):
        print(i)
    # 2. 利用next
    it = iterator_yield(3)
    print('测试next:{}'.format(next(it)))

此时协程有四种状态位:

  • GEN_CREATED: 等待执行, 此时还未进入协程
  • GEN_RUNNING: 解释器执行
  • GEN_SUSPENDED: yield 处暂停等待
  • GEN_CLOSED: 执行结束
  1. 利用 yield 抛出迭代值, 利用 send 传入 jump 值, 注意, 抛出值和传入值是不同的变量.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
def iterator_yield(maxnum):
    """ 传入一个最大值, 一旦迭代超过该值, 结束 """
    step = 0
    while step < maxnum:
        # ***注意, 这里yield将结果step抛出, 之后会将外面通过send传入的结果传入jump中***
        jump = yield step
        if jump is None:
            step += 1
        else:
            print('传入一个新的step值:{}'.format(jump))
            step = step + jump


if __name__ == '__main__':
    # 1. 迭代器测试, 其中next(it)就相当于it.send(None)数据, 这点很重要
    it = iterator_yield(6)
    print(next(it))
    # 2. 将3传入迭代器中
    it.send(3)
    print(next(it))

另外, 注意, 如果多次 send, 则会抛出异常. 在开始讲解 yield from 之前我们先简单的介绍下 yield 的缺点以对后面要讲解的 yield from 有一个清晰的认识:

  • a. 协程函数返回值获取不方便, 只能通过触发StopIteration来最终返回异常的 value, 例如 for 循环中被语法糖自动处理
  • b. 无法进行 yield 嵌套或者实现复杂, 每次只能向外层 yield 一个值
  • c. 需要手动处理异常
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
def bamboo_generator():
    for _i in range(10):
        if _i == 5:
            return '中断循环, 返回'
        yield _i


def check_iterator_return():
    """ 验证yield的缺点, 在for语法糖中无法获取函数返回值 """
    print('------------test iterator return---------------')
    try:
        for item in bamboo_generator():
            print(item)
    except StopIteration as msg:
        print(msg)

注意, 上面的代码在 python2.7 下运行有误, 从执行结果可知, bamboo_generator中的 return 返回结果不知道怎么获取.

yield from

利用 yield from 来使生成器能够很容易分为多个拥有 send 和 throw 方法的子生成器, 类似一个大函数可以分为多个子函数一样简单. 该语法的使用格式: yield from {普通的可迭代对象, 迭代器, 生成器}, 其实际返回两个值:

  • 从子生成器返回的一个迭代对象, 被直接抛出给上层, 例如下面例子中直接抛给 main 层
  • yield from 表达式本身返回子生成器的 return 值, 例如下面例子中show_return的 value 值, 其中 yield from 会自动捕获子生成器的 StopIteration 异常并返回

在整个交互管道上有几个相关的专业术语需要简单介绍下:

  • 委派生成器: 包含yield from <iterable>的生成器函数, 其相当于一个 wrap channel
  • 子生成器: yield from后面的生成器
  • 调用方: 调用委派生成器的客户端
  • 生成器链: 若子生成器本身也是一个 yield from 封装的生成器, 则可以将任意数量的委托派生器关联在一个链上

下面就是一个简单的例子:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
def return_ex():
    """ 测试yield from会返回其后生成器的 return 值 """
    yield 'return_yield_1'
    yield 'return_yield_2'
    yield 'return_yield_3'
    return 'return_return_4'


def show_return():
  	# yield from等价于: for item in iterable: yield item
    value = yield from return_ex()
    print('生成器 1值:{}'.format(value))
    yield 'End'


def show_return_eq():
  	# 注意, 默认请下yield item是无法获取generator的return, 因为一旦发生StopIteration则会"自动"被for处理掉,
    #	这也是yeild from提出的目的之一
  	try:
      for item in return_ex():
          yield item
    except StopIteration as exc:
      value = exec.value  # 终止迭代后的返回值


if __name__ == '__main__':
    show = show_return()
    for v in show:
        print('生成器 2 返回:{}'.format(v))

上面的代码输出结果如下:

1
2
3
4
5
生成器 2 返回:return_yield_1
生成器 2 返回:return_yield_2
生成器 2 返回:return_yield_3
生成器 1值:return_return_4
生成器 2 返回:End

再回顾 2.1 节 yield 缺点的例子, 如果使用 yield from 实现则效果如何呢?

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
def bamboo_generator():
    for _i in range(10):
        if _i == 5:
            return '中断循环, 返回'
        yield _i


def check_iterator_return():
    """ 验证yield的缺点, 在for语法糖中无法获取函数返回值 """
    print('------------test iterator return---------------')
    try:
        for item in bamboo_generator():
            print(item)
    except StopIteration as msg:
        print(msg)


def wrap_bamboo_generator(generator):
    """ 包装已有的生成器, 返回一个新的生成器 """
    result = yield from generator
    print(f'被包装生成器结果:{result}')


def check_yield_from_return():
    """ 验证yield from修复yield 无法返回return的问题 """
    print('------------test yield from return---------------')
    try:
        for item in wrap_bamboo_generator(bamboo_generator()):
            print(item)
    except StopIteration as msg:
        print(msg)

从结果可知, 通过封装已有的 generator 并返回该 generator 返回值最后打印出了 return 的返回值. 实际上yield from iterable等价于for item in iterable: yield item, 另外 yield from 还能作为一个中介机构, 作为双通道来串联多个生成器.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
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
import json
from collections import namedtuple

Result = namedtuple('Result', 'count average')

def averager():
    """ 子生成器: 求取平均值, 根据send过来的值不断进行累加并返回平均值 """
    print('Go into average generator...')
    count = total = 0.0
    while True:
        term = yield 'test'  # 这里'test'仅仅作为测试使用
        if term is None:
            break
        total += term
        count += 1
    return Result(count, total / count)

def grouper(results, key):
    """ 双管道, 会自动将main中send的值传递给averager,
    在averager最终return之后才会返回表达式本身的值, 其他使用作为一个数据传输管道
    """
    results[key] = yield from averager()
    print('---Grouper结束运行---')
    return '委托end'

def main(_data):
    """ 统计平均值入口函数 """
    results = {}
    for key, values in _data.items():
        try:
            group = grouper(results, key)
            next(group)  # a. 初始化生成器(此时就开始进入子生成器), 注意, 这里相当于send(None)
            for value in values:  # b. 进行数据交互
                group.send(value)  # 'test'
            group.send(None)  # c. 将子生成器退出
        except Exception as msg:
            print(F'委托生成器最终的返回值:{msg}')
    # 虽然程序最后结束了, 但是grouper以及averager生成器并没有结束.
    print(json.dumps(results, indent=2))

if __name__ == '__main__':
    data = {
        'girls;kg': [40.9, 38.5, 44.3, 42.2, 45.2, 41.7, 44.5, 38.0, 40.6, 44.5],
        'girls;m': [1.6, 1.51, 1.4, 1.3, 1.41, 1.39, 1.33, 1.46, 1.45, 1.43],
    }
    main(data)

上面的代码说明以及yield from中三个角色关系图如下:yield_chain

async

asyncio

asyncio是用来编写并发代码的库, 为构建 IO 密集型和网络连接代码提供了最佳的解决方案. 首先,让我们简单了解下 asyncio 中各个简单的概念:

  1. event_loop, 事件循环, 了解过 libev 等事件库的都了解, 这是一个无限事件循环, 不断监听事件的监听器
  2. coroutine, 协程对象, 可以通过内置的方法创建协程对象, 之后将协程对象注册到事件 event_loop 中.注意, 在 python3.4 中协程使用@asyncio.coroutine, 使用yield from来驱动, 但是在 python3.5 中发生了一些变化: @asyncio.coroutine -> async, yield from -> await, 当然新版本仍然向后兼容.
  3. Future, 表示尚未完成的计算或者结果, 或者表示为将来执行或还没有执行的任务
  4. Task, Future 的子类, 运行某个任务的同时可以并发的运行多个任务, 其通过asyncio.async(), loop.create_task(), asyncio.ensure_future()来创建.

下面是一个仅仅在python3.4环境下的异步编写例子:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
@asyncio.coroutine
def countdown(number, n):
    while n > 0:
        print('T-minus', n, '({})'.format(number))
        # 在python3.5之前, 使用yield from代替await
        # 注意asyncio.sleep是专门用于异步编程的, 如果是time.sleep, 则会直接阻塞
        _start = time.time()
        yield from asyncio.sleep(1)
        _end = time.time()
        print(f'{number} --> {round(_end - _start, 3)}')
        n -= 1


loop = asyncio.get_event_loop()  # 获取事件循环句柄
start = time.time()
tasks = [  # 创建多个coroutine对象
    asyncio.ensure_future(countdown("A", 2)),
    asyncio.ensure_future(countdown("B", 3)),
    asyncio.ensure_future(countdown("c", 2)),
    asyncio.ensure_future(countdown("d", 0)),
    asyncio.ensure_future(countdown("e", 5)),
    asyncio.ensure_future(countdown("f", 2)),
]
loop.run_until_complete(asyncio.wait(tasks))  # 循环运行
loop.close()
end = time.time()
""" 输出说明:
1. 串行执行预期耗时:
    2 + 3 + 2 + 0 + 5 + 2 == 14
2. 实际耗时:
    max(2, 3, 2, 0, 5, 2) == 5
所以协程并发缺失达到预期的目标, 将IO和CPU执行区分开来.
"""
print(f'整个过程总耗时:{int(end - start)}')

上面的代码输出如下:

1
2
3
4
...
T-minus 1 (e)
e --> 1.002
整个过程总耗时:5

注意上面的asyncio.sleep函数, 如果在异步协程中出现同步模块相关的代码, 实际上无法实现异步编程. 例如在进行网络爬取的时候, 可以使用功能基于 asyncio 的库aiohttp来进行网络请求.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# python版本为3.5+
import aiohttp
import asyncio
import async_timeout

async def fetch(session, url):
    with async_timeout.timeout(10):
        async with session.get(url) as response:
            return await response.text()

async def main():
    async with aiohttp.ClientSession() as session:
        html = await fetch(session, 'https://python.org')
        print(html)

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

又比如在scrapy中使用request来请求网络资源就会浪费整个框架的异步设计.

async/await

在 Python3.5 之后, python 内置支持异步编程操作, 上面的测试例子可以改为:

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

async def countdown(number, n):
    while n > 0:
        print('T-minus', n, '({})'.format(number))
        await asyncio.sleep(1)
        n -= 1

loop = asyncio.get_event_loop()  # 获取事件循环句柄
tasks = [  # 创建多个coroutine对象
    asyncio.ensure_future(countdown("A", 2)),
    asyncio.ensure_future(countdown("B", 3)),
    asyncio.ensure_future(countdown("c", 2)),
]
loop.run_until_complete(asyncio.wait(tasks))  # 循环运行
loop.close()

注意, 更多关于 async, await 的例子说明见项目asyncio-ftwpd, 该项目中有非常详细的 asyncio 的使用实例.

参考