• 流畅的Python(十六)-协程

    一、核心要义

    1. 生成器作为协程使用时的行为状态

    2.使用装饰器自动预激协程

    3.调用方使用close和throw方法控制协程

    4.协程终止时如何返回值

    5.yield from新句法的用途和语义

    6.案例：出租车队运营仿真

    二、代码示例

    1、协程入门案例

    1. #!/usr/bin/env python
    2. # -*- coding: utf-8 -*-
    3. # @Time    : 2024/3/2 9:07
    4. # @Author  : Maple
    5. # @File    : 01-协程入门案例.py
    6. # @Software: PyCharm
    7. import inspect
    8.  
    9. """
    10. 协程有4种状态
    11. 1.'GEN_CREATED': 等待开始执行
    12. 2.'GEN_RUNNINF': 解释器正在执行
    13. 3.'GEN_SUSPENDED': 在yield表达式处暂停
    14. 4.'GEN_CLOSED': 执行结束
    16. 注意：只有当协程处于状态3的时候,才可以send值.而刚创建的协程,没有预激活(即没有执行next方法)
    17.    ,此时处于状态1,如果给协程send值,会报错
    18. """
    19.  
    20.  
    21. def simple_coro(a):
    22.     print('-->start: a = ',a)
    23.     b = yield a
    24.     print('-->Received：b =  ',b)
    25.     c = yield  a + b
    26.     print('-->Received：c = ', c)
    27.  
    28. if __name__ == '__main__':
    29.  
    30.     # 1.实例化生成器函数
    31.     s = simple_coro(2)
    32.     ## 查看此时协程的状态: 创建状态
    33.     print(inspect.getgeneratorstate(s)) # GEN_CREATED
    34.  
    35.     # 2.调用next,预激活协程,此时协程会停止于第一个yield a处(等待外部调用函数发送send,往下推进执行)，并且将a值返回给外部
    36.     a = next(s)
    37.     print(a)
    38.     ## 查看此时协程的状态: 阻塞状态(等待外部通过send发送值)
    39.     print(inspect.getgeneratorstate(s))  # GEN_SUSPENDED
    40.  
    41.     # 3.通过send给s发送值,重新启动协程,并且外部的值会赋给b,最终再次阻塞在下一个yield处(yield a + b)
    42.        #并将外部 a + b的值返回给调用方【此时协程又会处于阻塞状态,等待外部发送send值】
    43.     c =  s.send(10)
    44.     print(c) # 12(备注:2 + 10)
    45.     ##  ## 查看此时协程的状态: 阻塞状态(等待外部通过send发送值)
    46.     print(inspect.getgeneratorstate(s)) # GEN_SUSPENDED
    47.  
    48.     # 4.再次通过send给s发送值,重新启动协程,并且外部的值会赋值给c,输出print(-->Received：c =   xx)
    49.     # 由于协程后面没有yield了,所以最后协程终止,导致[生成器]抛出StopIteration异常
    50.     """
    51.     Traceback (most recent call last):
    52.     File "D:/01-study/python/fluent_python/16-协程/01-协程入门案例.py", line 39, in 
    53.     s.send(100)
    54.     StopIteration
    55.     """
    56.     s.send(100)

    2、使用协程计算平均值

    1. #!/usr/bin/env python
    2. # -*- coding: utf-8 -*-
    3. # @Time    : 2024/3/2 9:28
    4. # @Author  : Maple
    5. # @File    : 02-使用协程计算平均值.py
    6. # @Software: PyCharm
    7.  
    8.  
    9. def avg():
    10.  
    11.     sum = 0
    12.     count = 0
    13.     average = None
    14.     while True:
    15.         a = yield average
    16.         sum += a
    17.         count +=1
    18.         average = sum/count
    19.  
    20. if __name__ == '__main__':
    21.  
    22.     a = avg()
    23.     # 预激活协程
    24.     next(a)
    25.     # 计算初始平均值
    26.     avg1 = a.send(10)
    27.     print(avg1) # 10.0
    28.  
    29.     # 发送新数据,计算前两个数据的平均值
    30.     avg2 = a.send(20)
    31.     print(avg2) # 15.0
    32.  
    33.     # 再次发送新数据,计算前三个数据的平均值
    34.     avg3 = a.send(30)
    35.     print(avg3) # 20.0

    3、预激协程的装饰器

    1. #!/usr/bin/env python
    2. # -*- coding: utf-8 -*-
    3. # @Time    : 2024/3/2 9:36
    4. # @Author  : Maple
    5. # @File    : 03-预激协程的装饰器.py
    6. # @Software: PyCharm
    7.  
    8.  
    9. from inspect import getgeneratorstate
    10.  
    11. def coroutine(func):
    12.     """装饰器：让被装饰的协程向前执行一步,阻带到第一个yield处"""
    13.     def primer(*args,**kwargs):
    14.         gen = func(*args,**kwargs)
    15.         next(gen)
    16.         return gen
    17.     return primer
    18.  
    19. # 使用装饰器,装饰协程
    20. @coroutine
    21. def avg():
    22.  
    23.     sum = 0
    24.     count = 0
    25.     average = None
    26.     while True:
    27.         a = yield average
    28.         sum += a
    29.         count +=1
    30.         average = sum/count
    31.  
    32.  
    33. if __name__ == '__main__':
    34.  
    35.     a = avg()
    36.     # 1.初始状态的协程就已经处于阻塞状态了(而不是刚创建)
    37.     print(getgeneratorstate(a)) # GEN_SUSPENDED,即无需再通过next去预激活协程
    38.  
    39.     # 计算初始平均值
    40.     avg1 = a.send(10)
    41.     print(avg1)  # 10.0
    42.  
    43.     # 发送新数据,计算前两个数据的平均值
    44.     avg2 = a.send(20)
    45.     print(avg2)  # 15.0
    46.  
    47.     # 再次发送新数据,计算前三个数据的平均值
    48.     avg3 = a.send(30)
    49.     print(avg3)  # 20.0

    4、终止协程和异常处理

    1. #!/usr/bin/env python
    2. # -*- coding: utf-8 -*-
    3. # @Time    : 2024/3/2 9:54
    4. # @Author  : Maple
    5. # @File    : 04-终止协程和异常处理.py
    6. # @Software: PyCharm
    7.  
    8.  
    9. # 自定义异常
    10. class DemoException(Exception):
    11.     """演示用异常"""
    12.  
    13.  
    14. def demo_exc_handing():
    15.     print('-->coroutine started')
    16.     try:
    17.         while True:
    18.             try:
    19.                 x = yield
    20.             except DemoException:
    21.                 print('**DemoException handling,Continuning****')
    22.             else:
    23.                 print('-->coroutine received: {!r}'.format(x))
    24.         raise RuntimeError('This line should never run')
    25.  
    26.     # 该条语句永远也不会执行:如果while循环中遇到DemoException,协程处理后,会继续运行
    27.     # 如果遇到其它异常,生成器无法运行,则协程立刻终止(此时RuntimeError仍然没有机会执行)
    28.     finally:
    29.         """如果不管协程是否结束,都有一些清理工作需要协程处理,可以使用finally"""
    30.         print('-->coroutine ending')
    31.  
    32.  
    33. if __name__ == '__main__':
    34.  
    35.     # 1.协程测试1
    36.     print('*****1.协程测试1***********')
    37.     d = demo_exc_handing()
    38.     # 1-1.预激活协程
    39.     next(d) # ->coroutine started
    40.  
    41.     # 1-2.给协程发送数据
    42.     d.send(10) # -->coroutine received: 10
    43.     d.send(20) # -->coroutine received: 20
    44.  
    45.     # 1-3.通过throw给协程发送异常
    46.     # 异常被处理,协程可以继续运行
    47.     d.throw(DemoException) # **DemoException handling,Continuning****
    48.     d.send(30) # -->coroutine received: 30
    49.  
    50.     # 1-4.通过close方法，关闭协程
    51.     d.close()
    52.     from inspect import getgeneratorstate
    53.     print(getgeneratorstate(d)) # GEN_CLOSED
    54.  
    55.     # 2.协程测试2
    56.     print('*****2.协程测试2***********')
    57.     d2 = demo_exc_handing()
    58.  
    59.     next(d2) #-->coroutine started
    60.     d2.send(10) # -->coroutine received: 10
    61.     # 通过throw给协程发送一个其无法处理的异常,协程会停止，状态变成'GEN_CLOSED'
    62.     d2.throw(ZeroDivisionError)
    63.     print(getgeneratorstate(d2))  # GEN_CLOSED

    5、协程返回值

    1. #!/usr/bin/env python
    2. # -*- coding: utf-8 -*-
    3. # @Time    : 2024/3/2 10:22
    4. # @Author  : Maple
    5. # @File    : 05-协程返回值.py
    6. # @Software: PyCharm
    7. from collections import namedtuple
    8.  
    9. Result = namedtuple('Result','count average')
    10.  
    11. """
    12. 接收协程返回值
    13. """
    14.  
    15. def avg():
    16.     sum = 0
    17.     count = 0
    18.     average = None
    19.     while True:
    20.         a = yield
    21.         # 由于协程必须正常终止,外部才能接收到返回值,所以要提供一种机制,保证协程可以正常终止
    22.         if a is None:
    23.             break
    24.         sum += a
    25.         count +=1
    26.         average = sum/count
    27.  
    28.     # 协程正常终止,抛出StopIteration异常,此返回值会绑定在异常的一个属性中
    29.     return Result(count,average)
    30.  
    31.  
    32. if __name__ == '__main__':
    33.  
    34.     a = avg()
    35.     next(a)
    36.     a.send(10)
    37.     a.send(20)
    38.     a.send(30)
    39.  
    40.     try:
    41.         a.send(None)
    42.     except StopIteration as exc:
    43.         # 协程返回值绑定在异常中的属性中
    44.         result = exc.value
    45.  
    46.     print(result) # Result(count=3, average=20.0)

    6、委派生成器

    1. #!/usr/bin/env python
    2. # -*- coding: utf-8 -*-
    3. # @Time    : 2024/3/2 10:31
    4. # @Author  : Maple
    5. # @File    : 06-委派生成器.py
    6. # @Software: PyCharm
    7. from collections import namedtuple
    8.  
    9. Result = namedtuple('Result','count average')
    10.  
    11.  
    12. # 子生成器
    13. def avg():
    14.     sum = 0
    15.     count = 0
    16.     average = None
    17.     while True:
    18.         a = yield
    19.         # 由于协程必须正常终止,外部才能接收到返回值,所以要提供一种机制,保证协程可以正常终止
    20.         if a is None:
    21.             break
    22.         sum += a
    23.         count +=1
    24.         average = sum/count
    25.  
    26.     # 协程正常终止,抛出StopIteration异常,此返回值会绑定在异常的一个属性中
    27.     return Result(count,average)
    28.  
    29.  
    30. # 委派生成器
    31. def grouper(results,key):
    32.     while True:
    33.         # results：接收子生成器的返回值
    34.         results[key] = yield from avg()
    35.  
    36. # 调用方
    37. def main(data):
    38.     result = {}
    39.     for key,value in data.items():
    40.         # 每次遍历都生成一个新的委派生成器
    41.         group = grouper(result,key)
    42.         # 预激活委派生成器
    43.         next(group)
    44.  
    45.         # 利用委派生成器,给子生成器avg发送数据,
    46.         for data in value:
    47.             group.send(data)
    48.         # 给子生成器发送一个None,终止子生成器(抛出StopIteration异常,yield from能够处理该异常)
    49.         # 最终result会接收到 子生成器返回的值(注意:子生成器产出的值 是直接反馈给调用方,而不是委派生成器)
    50.         group.send(None)
    51.     report(result)
    52.  
    53. def report(results):
    54.  
    55.     for key,result in sorted(results.items()):
    56.         group,unit = key.split(';')
    57.         print('{:2} {:5} averaging {:.2f}{}'.format(result.count,group,result.average,unit))
    58.  
    59. if __name__ == '__main__':
    60.  
    61.     data = {
    62.         'girls;kg' : [40.5,50.2,51.2,40.8],
    63.         'girls;m': [1.53, 1.61, 1.46, 1.8],
    64.         'boys;kg': [50.5, 62.4, 71.4, 60.4],
    65.         'boys;m': [1.63, 1.71, 1.76, 1.8]
    66.     }
    67.  
    68.     """
    69.      4 boys  averaging 61.18kg
    70.      4 boys  averaging 1.72m
    71.      4 girls averaging 45.67kg
    72.      4 girls averaging 1.60m
    73.     """
    74.     main(data)

    7、出租车运营仿真案例

    1. #!/usr/bin/env python
    2. # -*- coding: utf-8 -*-
    3. # @Time    : 2024/3/2 12:58
    4. # @Author  : Maple
    5. # @File    : 07-出租车运营仿真案例.py
    6. # @Software: PyCharm
    7. import queue
    8. import random
    9. from collections import namedtuple
    10.  
    11. Event = namedtuple('Event','time ident action')
    12.  
    13. DEFAULT_NUMBER_OF_TAXIS = 3
    14. DEFAULT_END_TIME = 180
    15. SEARCH_DURATION = 5
    16. TRIP_DURATION = 20
    17. DEPARTURE_INTERVAL = 5
    18.  
    19. def taxi_process(ident,trips,start_time =0):
    20.  
    21.     time = yield Event(start_time,ident,'leave garage')
    22.     for i in range(trips):
    23.         time = yield Event(time,ident,'pick up passenger')
    24.         time = yield Event(time,ident,'drop off passenger')
    25.     yield Event(time,ident,'go home')
    26.  
    27.  
    28.  
    29. class Simulator:
    30.  
    31.     def __init__(self,procs_map):
    32.         self.events = queue.PriorityQueue()
    33.         self.procs = dict(procs_map)
    34.  
    35.     def run(self,end_time):
    36.  
    37.         for _,proc in self.procs.items():
    38.             # 放入每位司机的初始事件(从停车场出发)
    39.             self.events.put(next(proc))
    40.  
    41.         # 仿真的主循环
    42.         sim_time = 0
    43.  
    44.         while sim_time < end_time:
    45.             if self.events.empty():
    46.                 print('end simulate')
    47.                 break
    48.  
    49.             # 取出start_time最早的事件
    50.             current_event = self.events.get()
    51.             sim_time, ident, previous_action = current_event
    52.             # 输出当前事件
    53.             print('taxi:',ident,ident *'  ',current_event)
    54.  
    55.  
    56.             # 当前事件，是归属于哪一位司机
    57.             active_proc = self.procs[ident]
    58.             # 获取该司机下一个事件,并传递下一个事件开始时间
    59.             next_time = sim_time + Caculate_nexttime(previous_action)
    60.             try:
    61.                 next_event = active_proc.send(next_time)
    62.             except StopIteration:
    63.                 # 如果某位司机的事件已经遍历完毕,就删掉其对应记录
    64.                 del self.procs[ident]
    65.             else:
    66.                 # 将下一次事件放入事件库
    67.                 self.events.put(next_event)
    68.         else:
    69.             msg = '** end stimulation time: {} events pending ***'
    70.             msg.format(self.events.qsize())
    71.  
    72.  
    73.  
    74. def Caculate_nexttime(previous_action):
    75.     if previous_action in ['leave garage','drop off passenger']:
    76.         # 新状态是四处徘徊
    77.         interval = SEARCH_DURATION
    78.  
    79.     elif previous_action == 'pick up passenger':
    80.         # 新状态是行程开始
    81.         interval = TRIP_DURATION
    82.     elif previous_action == 'go home':
    83.         interval = 1
    84.     else:
    85.         raise ValueError('Unknown previous_action:%s' % previous_action)
    86.     return  int(random.expovariate(1/interval)) + 1
    87.  
    88.  
    89.  
    90.  
    91.  
    92. def main(end_time = DEFAULT_END_TIME,num_taxis = DEFAULT_NUMBER_OF_TAXIS,seed = None):
    93.     if seed is not None:
    94.         random.seed(seed)
    95.  
    96.     # 默认创建的司机事件集合
    97.     """
    98.     proc_map = {
    99.     '0':taxi_process(0,2,0),
    100.     '1':taxi_process(1,4,5),
    101.     '2':taxi_process(2,6,10)
    102. }
    104.     
    105.     """
    106.     proc_map = {i:taxi_process(i,(i+1) *2 ,i * DEPARTURE_INTERVAL ) for i in range(num_taxis)}
    107.     sim = Simulator(proc_map)
    108.     sim.run(end_time)
    109.  
    110. if __name__ == '__main__':
    111.  
    112.     main(seed=3)

  • 相关阅读:
    认识时间复杂度和简单排序算法
    华玉通软发布“云雀”执行管理中间件，为智能驾驶关键应用提供确定性执行保障
    Java EE——线程(3)
    编写自己的 WordPress 模板
    word2vec的算法原理（不用开源包，python实现）
    基于 JSch 实现服务的自定义监控解决方案
    【vue2】前端如何播放rtsp 视频流，拿到rtsp视频流地址如何处理，海康视频rtsp h264 如何播放
    PostgreSQL 16 关于vacuum freezing 性能提升的来由 与 之前PostgreSQL 的缺陷讨论
    Chapter 12 贝叶斯网络
    vue目录树的封装
  • 原文地址：https://blog.csdn.net/weixin_37901366/article/details/136415050