Python协程的用法和例子详解

· Python · 204889

这篇文章主要为大家详细介绍了Python协程的用法和例子，具有一定的参考价值，感兴趣的小伙伴们可以参考一下

从句法上看，协程与生成器类似，都是定义体中包含 yield 关键字的函数。可是，在协程中， yield 通常出现在表达式的右边（例如， datum = yield），可以产出值，也可以不产出 —— 如果 yield 关键字后面没有表达式，那么生成器产出 None。

协程可能会从调用方接收数据，不过调用方把数据提供给协程使用的是 .send(datum) 方法，而不是next(…) 函数。

==yield 关键字甚至还可以不接收或传出数据。不管数据如何流动， yield 都是一种流程控制工具，使用它可以实现协作式多任务：协程可以把控制器让步给中心调度程序，从而激活其他的协程==。

协程的生成器的基本行为

这里有一个最简单的协程代码：

def simple_coroutine():
 print('-> start')
 x = yield
 print('-> recived', x)

sc = simple_coroutine()

next(sc)
sc.send('zhexiao')

解释：

1. 协程使用生成器函数定义：定义体中有 yield 关键字。
2. yield 在表达式中使用；如果协程只需从客户那里接收数据，那么产出的值是 None —— 这个值是隐式指定的，因为 yield 关键字右边没有表达式。
3. 首先要调用 next(…) 函数，因为生成器还没启动，没在 yield 语句处暂停，所以一开始无法发送数据。
4. 调用send方法，把值传给 yield 的变量，然后协程恢复，继续执行下面的代码，直到运行到下一个 yield 表达式，或者终止。

==注意：send方法只有当协程处于 GEN_SUSPENDED 状态下时才会运作，所以我们使用 next() 方法激活协程到 yield 表达式处停止，或者我们也可以使用 sc.send(None)，效果与 next(sc) 一样==。

协程的四个状态：

协程可以身处四个状态中的一个。当前状态可以使用inspect.getgeneratorstate(…) 函数确定，该函数会返回下述字符串中的一个：
1. GEN_CREATED：等待开始执行
2. GEN_RUNNING：解释器正在执行
3. GEN_SUSPENED：在yield表达式处暂停
4. GEN_CLOSED：执行结束

==最先调用 next(sc) 函数这一步通常称为“预激”（prime）协程==（即，让协程向前执行到第一个 yield 表达式，准备好作为活跃的协程使用）。

import inspect

def simple_coroutine(a):
 print('-> start')

 b = yield a
 print('-> recived', a, b)

 c = yield a + b
 print('-> recived', a, b, c)

# run 
sc = simple_coroutine(5)

next(sc)
sc.send(6) # 5, 6
sc.send(7) # 5, 6, 7

示例：使用协程计算移动平均值

def averager():
 total = 0.0
 count = 0
 avg = None

 while True:
  num = yield avg
  total += num
  count += 1
  avg = total/count

# run
ag = averager()
# 预激协程
print(next(ag))  # None

print(ag.send(10)) # 10
print(ag.send(20)) # 15

解释：

1. 调用 next(ag) 函数后，协程会向前执行到 yield 表达式，产出 average 变量的初始值——None。
2. 此时，协程在 yield 表达式处暂停。
3. 使用 send() 激活协程，把发送的值赋给 num，并计算出 avg 的值。
4. 使用 print 打印出 yield 返回的数据。

终止协程和异常处理

协程中未处理的异常会向上冒泡，传给 next 函数或 send 方法的调用方（即触发协程的对象）。

==终止协程的一种方式：发送某个哨符值，让协程退出。内置的 None 和Ellipsis 等常量经常用作哨符值==。

显式地把异常发给协程

从 Python 2.5 开始，客户代码可以在生成器对象上调用两个方法，显式地把异常发给协程。

generator.throw(exc_type[, exc_value[, traceback]])

致使生成器在暂停的 yield 表达式处抛出指定的异常。如果生成器处理了抛出的异常，代码会向前执行到下一个 yield 表达式，而产出的值会成为调用 generator.throw方法得到的返回值。如果生成器没有处理抛出的异常，异常会向上冒泡，传到调用方的上下文中。

generator.close()

致使生成器在暂停的 yield 表达式处抛出 GeneratorExit 异常。如果生成器没有处理这个异常，或者抛出了 StopIteration 异常（通常是指运行到结尾），调用方不会报错。如果收到 GeneratorExit 异常，生成器一定不能产出值，否则解释器会抛出RuntimeError 异常。生成器抛出的其他异常会向上冒泡，传给调用方。

异常处理示例：

class DemoException(Exception):
 """
 custom exception
 """

def handle_exception():
 print('-> start')

 while True:
  try:
   x = yield
  except DemoException:
   print('-> run demo exception')
  else:
   print('-> recived x:', x)

 raise RuntimeError('this line should never run')

he = handle_exception()
next(he)
he.send(10) # recived x: 10
he.send(20) # recived x: 20

he.throw(DemoException) # run demo exception

he.send(40) # recived x: 40
he.close()

如果传入无法处理的异常，则协程会终止：

he.throw(Exception) # run demo exception

yield from获取协程的返回值

为了得到返回值，协程必须正常终止；然后生成器对象会抛出StopIteration 异常，异常对象的 value 属性保存着返回的值。

==yield from 结构会在内部自动捕获 StopIteration 异常==。对 yield from 结构来说，解释器不仅会捕获 StopIteration 异常，还会把value 属性的值变成 yield from 表达式的值。

yield from基本用法

==在生成器 gen 中使用 yield from subgen() 时， subgen 会获得控制权，把产出的值传给 gen 的调用方，即调用方可以直接控制 subgen。与此同时， gen 会阻塞，等待 subgen 终止==。

下面2个函数的作用一样，只是使用了 yield from 的更加简洁：

def gen():
 for c in 'AB':
  yield c

print(list(gen()))

def gen_new():
 yield from 'AB'

print(list(gen_new()))

==yield from x 表达式对 x 对象所做的第一件事是，调用 iter(x)，从中获取迭代器，因此， x 可以是任何可迭代的对象，这只是 yield from 最基础的用法==。

yield from高级用法

==yield from 的主要功能是打开双向通道，把最外层的调用方与最内层的子生成器连接起来，这样二者可以直接发送和产出值，还可以直接传入异常，而不用在位于中间的协程中添加大量处理异常的样板代码==。

yield from 专门的术语

委派生成器：包含 yield from 表达式的生成器函数。
子生成器：从 yield from 中 部分获取的生成器。

图示

解释：

1. 委派生成器在 yield from 表达式处暂停时，调用方可以直接把数据发给子生成器。
2. 子生成器再把产出的值发给调用方。
3. 子生成器返回之后，解释器会抛出 StopIteration 异常，并把返回值附加到异常对象上，此时委派生成器会恢复。

高级示例

from collections import namedtuple

ResClass = namedtuple('Res', 'count average')


# 子生成器
def averager():
 total = 0.0
 count = 0
 average = None

 while True:
  term = yield
  if term is None:
   break
  total += term
  count += 1
  average = total / count

 return ResClass(count, average)


# 委派生成器
def grouper(storages, key):
 while True:
  # 获取averager()返回的值
  storages[key] = yield from averager()


# 客户端代码
def client():
 process_data = {
  'boys_2': [39.0, 40.8, 43.2, 40.8, 43.1, 38.6, 41.4, 40.6, 36.3],
  'boys_1': [1.38, 1.5, 1.32, 1.25, 1.37, 1.48, 1.25, 1.49, 1.46]
 }

 storages = {}
 for k, v in process_data.items():
  # 获得协程
  coroutine = grouper(storages, k)

  # 预激协程
  next(coroutine)

  # 发送数据到协程
  for dt in v:
   coroutine.send(dt)

  # 终止协程
  coroutine.send(None)
 print(storages)

# run
client()

解释：

1. 外层 for 循环每次迭代会新建一个 grouper 实例，赋值给 coroutine 变量； grouper 是委派生成器。

2. 调用 next(coroutine)，预激委派生成器 grouper，此时进入 while True 循环，调用子生成器 averager 后，在 yield from 表达式处暂停。
3. 内层 for 循环调用 coroutine.send(value)，直接把值传给子生成器 averager。同时，当前的 grouper 实例（coroutine）在 yield from 表达式处暂停。
4. 内层循环结束后， grouper 实例依旧在 yield from 表达式处暂停，因此， grouper函数定义体中为 results[key] 赋值的语句还没有执行。
5. coroutine.send(None) 终止 averager 子生成器，子生成器抛出 StopIteration 异常并将返回的数据包含在异常对象的value中，yield from 可以直接抓取 StopItration 异常并将异常对象的 value 赋值给 results[key]

yield from的意义

  • 子生成器产出的值都直接传给委派生成器的调用方（即客户端代码）。
  • 使用 send() 方法发给委派生成器的值都直接传给子生成器。如果发送的值是None，那么会调用子生成器的 next() 方法。如果发送的值不是 None，那么会调用子生成器的 send() 方法。如果调用的方法抛出 StopIteration 异常，那么委派生成器恢复运行。任何其他异常都会向上冒泡，传给委派生成器。
  • 生成器退出时，生成器（或子生成器）中的 return expr 表达式会触发 StopIteration(expr) 异常抛出。
  • yield from 表达式的值是子生成器终止时传给 StopIteration 异常的第一个参数。
  • 传入委派生成器的异常，除了 GeneratorExit 之外都传给子生成器的 throw() 方法。如果调用 throw() 方法时抛出 StopIteration 异常，委派生成器恢复运行。 StopIteration 之外的异常会向上冒泡，传给委派生成器。
  • 如果把 GeneratorExit 异常传入委派生成器，或者在委派生成器上调用 close() 方法，那么在子生成器上调用 close() 方法，如果它有的话。如果调用close()方法导致异常抛出，那么异常会向上冒泡，传给委派生成器；否则，委派生成器抛出GeneratorExit 异常。

使用案例

协程能自然地表述很多算法，例如仿真、游戏、异步 I/O，以及其他事件驱动型编程形式或协作式多任务。协程是 asyncio 包的基础构建。通过仿真系统能说明如何使用协程代替线程实现并发的活动。

在仿真领域，进程这个术语指代模型中某个实体的活动，与操作系统中的进程无关。仿真系统中的一个进程可以使用操作系统中的一个进程实现，但是通常会使用一个线程或一个协程实现。

出租车示例

import collections

# time 字段是事件发生时的仿真时间，
# proc 字段是出租车进程实例的编号，
# action 字段是描述活动的字符串。
Event = collections.namedtuple('Event', 'time proc action')


def taxi_process(proc_num, trips_num, start_time=0):
 """
 每次改变状态时创建事件，把控制权让给仿真器
 :param proc_num:
 :param trips_num:
 :param start_time:
 :return:
 """
 time = yield Event(start_time, proc_num, 'leave garage')

 for i in range(trips_num):
  time = yield Event(time, proc_num, 'pick up people')
  time = yield Event(time, proc_num, 'drop off people')

 yield Event(time, proc_num, 'go home')

# run
t1 = taxi_process(1, 1)
a = next(t1) 
print(a) # Event(time=0, proc=1, action='leave garage')
b = t1.send(a.time + 6)
print(b) # Event(time=6, proc=1, action='pick up people')
c = t1.send(b.time + 12)
print(c) # Event(time=18, proc=1, action='drop off people')
d = t1.send(c.time + 1)
print(d) # Event(time=19, proc=1, action='go home')

模拟控制台控制3个出租车异步

import collections
import queue
import random

# time 字段是事件发生时的仿真时间，
# proc 字段是出租车进程实例的编号，
# action 字段是描述活动的字符串。
Event = collections.namedtuple('Event', 'time proc action')


def taxi_process(proc_num, trips_num, start_time=0):
 """
 每次改变状态时创建事件，把控制权让给仿真器
 :param proc_num:
 :param trips_num:
 :param start_time:
 :return:
 """
 time = yield Event(start_time, proc_num, 'leave garage')

 for i in range(trips_num):
  time = yield Event(time, proc_num, 'pick up people')
  time = yield Event(time, proc_num, 'drop off people')

 yield Event(time, proc_num, 'go home')


class SimulateTaxi(object):
 """
 模拟出租车控制台
 """

 def __init__(self, proc_map):
  # 保存排定事件的 PriorityQueue 对象，
  # 如果进来的是tuple类型，则默认使用tuple[0]做排序
  self.events = queue.PriorityQueue()
  # procs_map 参数是一个字典,使用dict构建本地副本
  self.procs = dict(proc_map)

 def run(self, end_time):
  """
  排定并显示事件，直到时间结束
  :param end_time:
  :return:
  """
  for _, taxi_gen in self.procs.items():
   leave_evt = next(taxi_gen)
   self.events.put(leave_evt)

  # 仿真系统的主循环
  simulate_time = 0
  while simulate_time < end_time:
   if self.events.empty():
    print('*** end of events ***')
    break

   # 第一个事件的发生
   current_evt = self.events.get()
   simulate_time, proc_num, action = current_evt
   print('taxi:', proc_num, ', at time:', simulate_time, ', ', action)

   # 准备下个事件的发生
   proc_gen = self.procs[proc_num]
   next_simulate_time = simulate_time + self.compute_duration()

   try:
    next_evt = proc_gen.send(next_simulate_time)
   except StopIteration:
    del self.procs[proc_num]
   else:
    self.events.put(next_evt)
  else:
   msg = '*** end of simulation time: {} events pending ***'
   print(msg.format(self.events.qsize()))

 @staticmethod
 def compute_duration():
  """
  随机产生下个事件发生的时间
  :return:
  """
  duration_time = random.randint(1, 20)
  return duration_time


# 生成3个出租车，现在全部都没有离开garage
taxis = {i: taxi_process(i, (i + 1) * 2, i * 5)
   for i in range(3)}

# 模拟运行
st = SimulateTaxi(taxis)
st.run(100)

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持昆居客【www.kunjuke.com】。

相关推荐：

《太吾绘卷：天幕心帷》创建人物详解攻略分享

《太吾绘卷：天幕心帷》中的创建人物选择是游戏里非常重要的事情，而创建人物需要注意的事情有很多，首先是捏脸选择性别，男女都可以，男性的优势就是可以多生很多孩子，女性的优势就是可以进行幼教胎教以及与他人初识的好感度会更高。 太吾绘卷天幕心帷创建人物要注意什么 本文内容来源于互联网，如有侵权请联系删除。

三国志：王道天下周瑜武将详解

三国志王道天下周瑜怎么样，周瑜+孙策+张昭/大乔是一个利用激流之书+流言四起/龙战于野等战法与孙策配合打出前三回合高概率控制，多种持续性状态配合周瑜打出高额输出，多种控制配合张昭/大乔使敌方受到伤害提升的爆发法师队。 三国志：王道天下周瑜武将详解： 一、周瑜武将解析 周瑜 强度评级：T0 7将御输出将 50级面板： 武力102.4，智力228.8，统率188.1，速度66.5。 满级智力全游戏第3...

《挖掘者米娜》沼泽梯子用法介绍

《挖掘者米娜》中的平基沙龙商店里面的梯子是游戏里非常重要的一种工具，用得好能够得到很多好东西，首先梯子要偷偷摸摸的用，不能跑到平基的面前，要不然你就会被扔到外面，首先是蕾丝手套，就在沙龙商店的高处平台上。 挖掘者米娜沼泽梯子怎么用 店里梯子拿到的是魔能匣我记错了，漏了一个拿到梯子往沼泽入口走有俩灯地方放上去有个一火花能进的洞，靠提灯的光走到最后有一瓶血浆上限。 本文内容来源于互联网，如有侵权请联系...

夜幕之下界面及战斗体系详解

夜幕之下怎么操作角色战斗，本篇介绍游戏内的基础的战斗规则，包括战斗的基本规则、选卡界面信息、局内战斗信息等，boss/精英怪有白条(下面统称白条怪)，只有对应的弱点属性(欲望)可以降低、打掉白条。白条存在时有高额减伤，白条打掉瞬间可以打断怪的招式并眩晕约5s。 夜幕之下界面及战斗体系详解： 一、基本规则 1、通关条件 战斗有时间限制。材料副本没有星级，只要倒计时结束前完成即可，也没有死亡人数限制。...

多开分身怎么恢复默认设置？重置方法详解

华为/荣耀可直接在设置→应用和服务→应用分身→右上角三点→重置应用分身；小米/Redmi需在手机分身界面长按空白处选删除分身；一加/realme/OPPO须进入应用分身→高级设置→三点→重置所有应用分身。 想把手机里所有应用分身全部清空、回到最初没开过分身的状态，而不是一个个手动关掉，就得用系统级的重置功能——它会直接删除所有分身数据和配置，不保留任何残留。 华为/荣耀（HarmonyOS）一键重...

华为 Nova 6 参数配置详解：核心硬件与影像实力深度剖析

华为Nova 6在2026年仍可满足日常使用与多人自拍，但需验证麒麟990真伪、升级HarmonyOS 3、优化后台管理，并注意前置畸变校准、长焦固件更新及LCD屏幕可视角度限制。 想确认华为 Nova 6 是否仍能满足当前日常使用、视频拍摄或多人自拍需求，必须看清它的真实硬件底子——不是参数表里的堆砌文字，而是麒麟990在2026年实际调度表现、4200mAh电池在EMUI/HarmonyOS ...

百度浏览器怎样开启无痕模式_百度APP无痕浏览设置【详解】

可通过五种方式启用百度浏览器或APP的无痕模式：一、右上角菜单新建无痕窗口；二、快捷键Ctrl+Shift+N（Win/Linux）或Command+Shift+N（macOS）；三、地址栏输入chrome://incognito；四、手机APP搜索框旁盾牌图标一键启用；五、APP“我的”→设置→无痕模式开关。 如果您在使用百度浏览器或百度APP时希望避免浏览历史、搜索记录、Cookie及表单数据...

OPPO Reno2连接显示器与桌面模式使用详解【教程】

OPPO Reno2需通过USB-C转HDMI有线连接显示器并启用抽屉模式以适配大屏操作，再配置应用快捷方式和快应用提升外接显示效率。 OPPO Reno2连接显示器后想直接用桌面模式操作文件、投屏办公或看高清视频，但默认桌面布局在大屏下图标太小、操作不便，必须同步调整手机端桌面模式才能获得合理的大屏交互体验。 将OPPO Reno2连接到外接显示器 确认显示器具备HDMI接口，或已配备兼容OPP...