当前位置：首页 > news >正文

75、Python之函数式编程：生成器的核心方法及更多使用场景

news 2024/11/14 20:53:34

引言

Python中的函数式编程，依托生成器，可以实现惰性求值的特性。但是，生成器其实还可以有更多的使用场景。本文就聚焦生成器，再次聊聊生成器中的主要方法以及更多的使用场景。

本文的主要内容有：

1、生成器的核心方法

2、生成器的使用场景

生成器的核心方法

首先还是来看下生成器的定义：

生成器类中，主要用到的有这几个方法：

1、__next__()方法，用于获取生成器的下一个元素，如果没有更多元素了，则会抛出StopIteration异常。

2、send()方法，用于像生成器对象发送一个值，从而实现双向通信的场景。

3、throw()方法，用于向生成器内部引发一个异常，生成器内部可以捕获并处理该异常。

4、close()方法，用于关闭一个生成器对象。

接下来，我们通过代码实例演示一下这几个方法的使用。

def generator():while True:try:msg = yield '等待接收新的消息...'if msg:print(f'收到新消息: {msg}')if msg.lower() == 'exit':breakexcept ValueError:yield '发生异常，通信结束'yield '通信结束'gen = generator()
# __next__()
print(gen.__next__())
# send()发送一个值
print(gen.send('hello python'))
# 发送一个异常
print(gen.throw(ValueError))
# __next__()
print(gen.__next__())
print(gen.close())
# 发送exit消息可以终止
print(gen.send('exit'))
# 也可以手动调用close()
# 再次调用__next__()就会抛StopIteration异常了
gen.__next__()

执行结果：

简单总结一下这几个方法的使用：

1、__next__()方法，前面已经介绍过，会获取到yield表达式右侧的值，如果生成器已经被耗尽（没有更多的值了），则抛出StopIteration异常，等价于内置函数next()。

2、send()方法，用于向生成器发送一个值，并使生成器继续执行下一个yield表达式。发送的值会被赋值给一个yield表达式的左侧。与__next__()方法不同，send()方法可以在生成器与调用者之间实现双向通信。

3、throw()方法，用于向生成器内部引发一个异常，生成器可以在内部捕获这个异常，并处理它。如果生成器内部没有捕获这个异常，它将传播到调用者，并终止生成器。

4、close()方法，用于关闭一个生成器对象，之后，将不能对该生成器执行任何方法，否则均会抛出StopIteration异常。

生成器的使用场景

了解了生成器中的核心方法，接下来看一下更多的应用生成器的场景。

1、基本的协程的场景

协程是一种可以在执行过程中暂停和恢复的子程序，基于生成器自动阻塞的特性，以及send()方法可以实现的双向通信功能，可以利用生成器实现基本的协程的功能，适用于异步编程和多任务的写作。

这里暂时不进行代码的演示了，后面会在并发编程中详细展开。

2、状态机

基于生成器可以实现一个简单的状态机的功能，每次可以使用send()方法来改变生成器的状态。可以通过代码来看一个简单的状态机的模拟。

直接看代码：

def state_machine():state = 'PLAN'while True:if state == 'PLAN':state = yield 'PLAN PHASE'elif state == 'DO':state = yield 'DO PHASE'elif state == 'CHECK':state = yield 'CHECK PHASE'elif state == 'ACT':state = yield 'ACT PHASE'# 模拟PDCA/戴明环
sm = state_machine()
print(next(sm))
print(sm.send('DO'))
print(sm.send('CHECK'))
print(sm.send('ACT'))
print(sm.send('PLAN'))

执行结果：

3、实时数据处理

生成器可以用于处理实时输入的数据，例如传感器监测数据、网络日志等，通过send()方法实现实时数据的传入及处理。

以一个实时计算平均值的示例作为演示：

def calculate_avg():total = 0count = 0avg = Nonewhile True:num = yield avgtotal += numcount += 1avg = total / countprint(f'已接收到{count}个数字，总和：{total}，平均值：{avg}')averager = calculate_avg()
# 启动生成器，首次返回None
print(averager.__next__())
# 模拟实时传值并计算
print(averager.send(10))
print(averager.send(15))
print(averager.send(25))

执行结果：

其实代码逻辑很简单，只要能够理解生成器的基本执行原理即可，这里重点回顾一下：