上一章，我们介绍了Python的协程，并讲到用yield达到协程的效果，这一章，我们来介绍yield from的结构和作用
我们先来对比下yield和yield from的用法
def first_gen():
for c in "AB":
yield c
for i in range(0, 3):
yield i
print(list(first_gen()))
def second_gen():
yield from "AB"
yield from range(0, 3)
print(list(second_gen()))
运行结果：
['A', 'B', 0, 1, 2] ['A', 'B', 0, 1, 2]
我们可以看到，两个方法都可以达到一样的效果，但是second_gen()方法比first_gen()方法来的简练，在second_gen()中使用yield from subgen()时，second_gen()是陷入阻塞的，真正在交互的是调用second_gen()的那一方，也就是调用方和subgen()在交互，second_gen()的阻塞状态会一直持续到subgen()调用完毕，才会接着执行yield from后续的代码
再来看下面两个例子：
第一个例子：
def chain(*iterables):
for it in iterables:
yield from it
s = "ABC"
t = tuple(range(0, 3))
print(list(chain(s, t)))
运行结果：
['A', 'B', 'C', 0, 1, 2]
第二个例子：
from collections import Iterable
def flatten(items, ignore_types=(str, bytes)):
for x in items:
if isinstance(x, Iterable) and not isinstance(x, ignore_types):
yield from flatten(x) # 这里递归调用，如果x是可迭代对象，继续分解
else:
yield x
items = [1, 2, [3, 4, [5, 6], 7], 8]
# Produces 1 2 3 4 5 6 7 8
for x in flatten(items):
print(x)
items = ['Dave', 'Paula', ['Thomas', 'Lewis']]
for x in flatten(items):
print(x)
运行结果：
1 2 3 4 5 6 7 8 Dave Paula Thomas Lewis
yield from item 表达式对item对象所做的第一件事，就是调用iter(item)，从中获取迭代器，因此，item可以是任何一个可迭代的对象，在某些时候，yield from可以代替for循环，使得我们的代码更加的精炼，yield from是委派生成器，主要功能还是和yield一样，打开双向通道，把最外层的调用方和实际在传输值的生成器连接起来，这样，二者就可以发送和产出值
下面一个例子，让我们用yield from计算平均值并返回统计报告，假设我们有若干男孩和女孩，我们分别拥有这些男孩和女孩的体重和身高，现在我们要分别计算男孩女孩的体重、身高的平均值
from collections import namedtuple
Result = namedtuple('Result', 'count average')
def averager(): # <1>
total = 0.0
count = 0
average = None
while True:
term = yield # <2>
if term is None: # <3>
break
total += term
count += 1
average = total / count
return Result(count=count, average=average) # <4>
def grouper(results, key): # <5>
while True: # <6>
results[key] = yield from averager() # <7>
def main(data):
results = {}
for key, values in data.items():
group = grouper(results, key)
next(group) # <8>
for value in values:
group.send(value)
group.send(None) # <9>
report(results)
def report(results):
for key, result in sorted(results.items()):
group, unit = key.split(";")
print('{:2} {:5} averaging {:.2f}{}'.format(result.count, group, result.average, unit))
data = {
'girls;kg':
[40.9, 38.5, 44.3],
'girls;m':
[1.6, 1.51, 1.4],
'boys;kg':
[39.0, 40.8],
'boys;m':
[1.38, 1.5],
}
if __name__ == '__main__':
main(data)
运行结果：
2 boys averaging 39.90kg 2 boys averaging 1.44m 3 girls averaging 41.23kg 3 girls averaging 1.50m
<1>中的yield标明averager是一个生成器，这里我们在grouper()方法中调用yield from averager()，所以averager是一个子生成器
<2>main()函数中的代码发送值绑定到term上
<3>当子生成器发现客户端代码发送一个None时，跳出循环，如果不那么做，averager()方法永远不会返回我们要的统计量，也就是Result对象，我们可以通过可以关闭委托生成器，但Result对象依旧无法返回
<4>当调用方传入一个None值，子生成器跳出循环，并返回Result对象
<5>yield from averager()表达式拿到上一个步骤返回的Result对象，赋值给左边的变量
<6>这个循环每次迭代时会新建一个averager生成器的实例，每个实例都是作为协程使用的生成器对象
<7>grouper会在yield from处暂停，等待调用方和子生成器交互完毕后，子生成器返回Result对象，grouper把返回的结果绑定到result[key]上
<8>我们依旧用next()方法激活一个委派生成器，然后我们开始不断往委派生成器中输送值
<9>当我们将值输送完毕后，传入None方法，使得自生成器内部跳出循环，并计算Result对象返回给grouper委托生成器存储起来
让我们用协程模拟一个出租车运营场景，系统会创建几辆出租车，每辆车都会拉客，每辆车拉客的次数不同，而每次拉客都有乘客上车和乘客下车这一动作，出租车继续四处拉客，当到达各自拉客的次数后，出租车回家。
出租车运营程序：
import random
import collections
import queue
DEFAULT_NUMBER_OF_TAXIS = 3
DEFAULT_END_TIME = 180
SEARCH_DURATION = 5
TRIP_DURATION = 20
DEPARTURE_INTERVAL = 5
Event = collections.namedtuple('Event', ["time", "proc", "action"])
def taxi_process(ident, trips, start_time=0):
time = yield Event(start_time, ident, '出库')
for i in range(trips):
time = yield Event(time, ident, '开始拉客')
time = yield Event(time, ident, '乘客下车')
yield Event(time, ident, '车入库')
class Simulator:
def __init__(self, procs_map):
self.events = queue.PriorityQueue()
self.procs = dict(procs_map)
def run(self, end_time):
for _, proc in sorted(self.procs.items()):
first_event = next(proc)
self.events.put(first_event)
sim_time = 0
while sim_time < end_time:
if self.events.empty():
print('*** 事件结束 ***')
break
current_event = self.events.get()
sim_time, proc_id, previous_action = current_event
print('taxi:', proc_id, proc_id * ' ', current_event)
active_proc = self.procs[proc_id]
next_time = sim_time + compute_duration(previous_action)
try:
next_event = active_proc.send(next_time)
except StopIteration:
del self.procs[proc_id]
else:
self.events.put(next_event)
else:
msg = '*** end of simulation time: {} events pending ***'
print(msg.format(self.events.qsize()))
def compute_duration(previous_action):
if previous_action in ['出库', '乘客下车']:
interval = SEARCH_DURATION
elif previous_action == '开始拉客':
interval = TRIP_DURATION
elif previous_action == '车入库':
interval = 1
else:
raise ValueError('Unknown previous_action: %s' % previous_action)
return int(random.expovariate(1 / interval)) + 1
def main(end_time=DEFAULT_END_TIME, num_taxis=DEFAULT_NUMBER_OF_TAXIS):
taxis = {i: taxi_process(i, (i + 1) * 2, i * DEPARTURE_INTERVAL)
for i in range(num_taxis)}
print("出租车数量:", len(taxis))
sim = Simulator(taxis)
sim.run(end_time)
if __name__ == '__main__':
main()
执行结果：
出租车数量: 3 taxi: 0 Event(time=0, proc=0, action='出库') taxi: 1 Event(time=5, proc=1, action='出库') taxi: 0 Event(time=6, proc=0, action='开始拉客') taxi: 0 Event(time=7, proc=0, action='乘客下车') taxi: 0 Event(time=9, proc=0, action='开始拉客') taxi: 2 Event(time=10, proc=2, action='出库') taxi: 2 Event(time=12, proc=2, action='开始拉客') taxi: 1 Event(time=24, proc=1, action='开始拉客') taxi: 1 Event(time=26, proc=1, action='乘客下车') taxi: 1 Event(time=28, proc=1, action='开始拉客') taxi: 0 Event(time=33, proc=0, action='乘客下车') taxi: 0 Event(time=36, proc=0, action='车入库') taxi: 2 Event(time=39, proc=2, action='乘客下车') taxi: 1 Event(time=41, proc=1, action='乘客下车') taxi: 2 Event(time=41, proc=2, action='开始拉客') taxi: 1 Event(time=45, proc=1, action='开始拉客') taxi: 1 Event(time=51, proc=1, action='乘客下车') taxi: 1 Event(time=52, proc=1, action='开始拉客') taxi: 1 Event(time=55, proc=1, action='乘客下车') taxi: 1 Event(time=61, proc=1, action='车入库') taxi: 2 Event(time=67, proc=2, action='乘客下车') taxi: 2 Event(time=70, proc=2, action='开始拉客') taxi: 2 Event(time=75, proc=2, action='乘客下车') taxi: 2 Event(time=80, proc=2, action='开始拉客') taxi: 2 Event(time=87, proc=2, action='乘客下车') taxi: 2 Event(time=90, proc=2, action='开始拉客') taxi: 2 Event(time=96, proc=2, action='乘客下车') taxi: 2 Event(time=97, proc=2, action='开始拉客') taxi: 2 Event(time=98, proc=2, action='乘客下车') taxi: 2 Event(time=106, proc=2, action='车入库') *** 事件结束 ***
利用协程，我们可以看到每辆出租车各自的出库、拉客、乘客下车、再拉客，直到车入库虽然是顺序执行，但车与车之间的行为却是并行，这里我们分析一下主要的代码块
首先是我们的taxi_process()方法：
def taxi_process(ident, trips, start_time=0):
time = yield Event(start_time, ident, '出库')
for i in range(trips): # <1>
time = yield Event(time, ident, '开始拉客')
time = yield Event(time, ident, '乘客下车')
yield Event(time, ident, '车入库')
这个方法会返回一个协程，这个协程控制着车的出库时间，循环拉客的次数以及入库时间，<1>处的trips就代表这辆车可以拉几次乘客
再来是Simulator模块
class Simulator:
def __init__(self, procs_map):
self.events = queue.PriorityQueue() # <1>
self.procs = dict(procs_map)
def run(self, end_time):
for _, proc in sorted(self.procs.items()):
first_event = next(proc) # <2>
self.events.put(first_event)
sim_time = 0
while sim_time < end_time:
if self.events.empty():
print('*** 事件结束 ***')
break
current_event = self.events.get()
sim_time, proc_id, previous_action = current_event # <3>
print('taxi:', proc_id, proc_id * ' ', current_event)
active_proc = self.procs[proc_id]
next_time = sim_time + compute_duration(previous_action)
try:
next_event = active_proc.send(next_time)
except StopIteration:
del self.procs[proc_id]
else:
self.events.put(next_event) # <4>
else:
msg = '*** end of simulation time: {} events pending ***'
print(msg.format(self.events.qsize())) # <5>
我们在<1>处生成一个队列，这个队列是后续协程在模拟出租车运营时需要用到的，之后我们copy一个字典，这个字典key是每辆车的id，value是每辆车的生成器
在<2>处，我们激活每辆车的生成器，把每辆车生成器返回的初始状态存入事先准备好的队列中
在<3>处，我们会从队列循环取出每辆车当前的状态，并从先前copy好的字典里取出每辆车的生成器，计算该车下一次行动的时间，并传入给生成器，如果生成器抛出StopIteration异常，则代表该生成器已结束，则在异常捕捉处从字典里删去该车的记录
在我们往生成器传入值的时候，如果生成器没有报错，则会返回车的下一个状态，而我们把最新的状态存入队列中，如果循环，一直到队列为空为止，则跳出循环，又或者我们设定的初始时间大于终止时间，则跳出循环，进入到<5>打印队列中还有多少个事件