摘要：本文介绍了Python中的响应式编程及其实现方式。响应式编程以数据流为核心，通过ReactiveX框架的Observable和Observer类实现数据传播。重点讲解了RxPY库的使用方法，包括创建数据源和观察者，并展示了基本示例。同时对比了PyFunctional库，指出其侧重函数式数据转换的特点。两个库都支持响应式编程，但RxPY更关注系统事件处理，而PyFunctional专注于数据流转换。文中提供了两个库的安装方法和代码示例，帮助读者理解响应式编程在Python中的实际应用。

目录

Python 中的并发 —— 响应式编程

响应式编程的实现：ReactiveX（RX）

可观察对象类（Observable）

观察者类（Observer）

用于响应式编程的 Python 模块：RxPY

示例：RxPY 库的使用

用于响应式编程的 PyFunctional 库

RxPY 与 PyFunctional 的区别

PyFunctional 模块的安装

示例：PyFunctional 库的使用

---

Python 中的并发 —— 响应式编程

响应式编程是一种处理数据流和变更传播的编程范式。这意味着当某个组件发出数据流时，响应式编程库会将该变更传播至其他组件，且这种变更传播会持续进行，直到抵达最终的接收方。事件驱动编程与响应式编程的区别在于：事件驱动编程围绕事件展开，而响应式编程围绕数据展开。

响应式编程的实现：ReactiveX（RX）

ReactiveX（也译作响应式扩展）是响应式编程最知名的实现框架，其运行机制依赖于以下两个核心类：

可观察对象类（Observable）

该类是数据流或事件的数据源，它会对传入的数据进行封装，使数据能在不同线程间传递。只有当有观察者对其进行订阅时，它才会向外提供数据。

观察者类（Observer）

该类用于消费可观察对象发出的数据流。一个可观察对象可以对应多个观察者，且每个观察者都会接收到可观察对象发出的每一个数据项。观察者通过订阅可观察对象，可接收三种类型的事件：

• on_next () 事件：表示数据流中存在一个数据元素。
• on_completed () 事件：表示数据流的发射结束，后续将不再有新的数据项发出。
• on_error () 事件：同样表示数据流发射结束，但触发原因是可观察对象抛出了异常。

用于响应式编程的 Python 模块：RxPY

RxPY 是一款可用于实现响应式编程的 Python 模块，使用前需先完成安装，可通过以下命令进行安装：

plaintext

pip install RxPY

示例：RxPY 库的使用

以下是一个使用 RxPY 模块及其 Observable、Observer 类实现响应式编程的 Python 脚本，核心包含两个自定义类 / 函数：

• get_strings ()：用于向观察者发送字符串数据。
• PrintObserver ()：用于打印从可观察对象接收到的字符串，该类实现了观察者的全部三种事件方法，同时会调用订阅（subscribe）方法。

main.py 代码

from rx import Observable, Observer

def get_strings(observer):
    observer.on_next("Ram")
    observer.on_next("Mohan")
    observer.on_next("Shyam")
    observer.on_completed()

class PrintObserver(Observer):
    def on_next(self, value):
        print("Received {0}".format(value))
    def on_completed(self):
        print("Finished")
    def on_error(self, error):
        print("Error: {0}".format(error))

source = Observable.create(get_strings)
source.subscribe(PrintObserver())

运行结果执行上述代码，输出结果如下：

plaintext

Received Ram
Received Mohan
Received Shyam
Finished

用于响应式编程的 PyFunctional 库

PyFunctional 是另一款可用于响应式编程的 Python 库，它支持开发者使用 Python 编写函数式程序。该库的实用价值在于，可通过链式函数操作符创建数据处理管道。

RxPY 与 PyFunctional 的区别

两款库均用于响应式编程，且对数据流的处理方式相似，核心差异体现在数据处理的侧重点上：RxPY 主要处理系统中的数据和事件；而 PyFunctional 则聚焦于通过函数式编程范式实现数据转换。

PyFunctional 模块的安装

使用前需先安装该模块，通过以下 pip 命令即可完成安装：

plaintext

pip install pyfunctional

示例：PyFunctional 库的使用

以下示例使用 PyFunctional 模块的 seq 类（该类作为流对象，支持迭代和数据操作）实现数据处理：程序中先通过匿名函数（lambda）对序列进行映射处理，将每个数值翻倍；再过滤出大于 4 的数值；最后通过归约操作，计算剩余所有数值的总和。

main.py 代码

from functional import seq

result = seq(1,2,3).map(lambda x: x*2).filter(lambda x: x > 4).reduce(lambda x, y: x + y)
print("Result: {}".format(result))

运行结果执行上述代码，输出结果如下：

plaintext

Result: 6