RxPY – 使用可观察变量

观察者，是一个创建观察者的函数，并将其附加到期望值的源头，例如，点击，来自dom元素的鼠标事件等。

下面提到的主题将在本章中详细研究。

• 创建观察者

• 订阅和执行一个可观察变量

创建可观察变量

为了创建一个可观察变量，我们将使用 create() 方法，并向其传递具有以下项目的函数。

• on_next() – 当Observable发出一个项目时，这个函数被调用。

• on_completed() – 这个函数在Observable完成时被调用。

• on_error() – 当Observable上发生错误时，这个函数被调用。

要使用create()方法，首先要导入该方法，如下图所示。

from rx import create

下面是一个工作实例，创建一个可观察的：

testrx.py

from rx import create
deftest_observable(observer, scheduler):
   observer.on_next("Hello")
   observer.on_error("Error")
   observer.on_completed()
source = create(test_observable).

订阅和执行一个可观察变量

要订阅一个可观察变量，我们需要使用subscribe()函数并传递回调函数on_next、on_error和on_completed。

下面是一个工作实例 –

testrx.py

from rx import create
deftest_observable(observer, scheduler):
   observer.on_next("Hello")
   observer.on_completed()
source = create(test_observable)
source.subscribe(
   on_next = lambda i: print("Got - {0}".format(i)),
   on_error = lambda e: print("Error : {0}".format(e)),
   on_completed = lambda: print("Job Done!"),
)

subscribe()方法负责执行可观察的内容。回调函数 on_next 、 on_error 和 on_completed 必须被传递给subscribe方法。对subscribe方法的调用反过来会执行test_observable()函数。

并非一定要向subscribe()方法传递所有三个回调函数。你可以根据你的要求传递on_next()、on_error()和on_completed()。

lambda函数被用于on_next、on_error和on_completed。它将接收参数并执行给定的表达式。

下面是创建的可观察变量的输出结果-

E:\pyrx>python testrx.py
Got - Hello
Job Done!