当前位置：极客教程 > Python > RxPY教程 > RxPY – 与主题一起工作

RxPY – 与主题一起工作

RxPY – 与主题一起工作

主体是一个可观察的序列，也是一个可以组播的观察者，即与许多已订阅的观察者交谈。

我们将讨论以下关于主题的话题 —

  • 创建一个主题
  • 订阅一个主题
  • 将数据传递给主体
  • 行为主体(BehaviorSubject)
  • 回放主体
  • 异步主体

创建一个主题

为了处理一个主题，我们需要导入主题，如下图所示

from rx.subject import Subject
Python

你可以按以下方式创建一个主客体-

subject_test = Subject()
Python

该对象是一个观察者，它有三个方法 –

  • on_next(value)
  • on_error(error)和
  • on_completed()。

订阅一个主题

你可以在主题上创建多个订阅，如下图所示

subject_test.subscribe(
   lambda x: print("The value is {0}".format(x))
)
subject_test.subscribe(
   lambda x: print("The value is {0}".format(x))
)
Python

传递数据给主题

你可以使用on_next(value)方法向创建的主题传递数据，如下所示

subject_test.on_next("A")
subject_test.on_next("B")
Python

这些数据将被传递给所有的订阅，添加到主题上。

这里，是一个关于主题的工作实例。

例子

from rx.subject import Subject
subject_test = Subject()
subject_test.subscribe(
   lambda x: print("The value is {0}".format(x))
)
subject_test.subscribe(
   lambda x: print("The value is {0}".format(x))
)
subject_test.on_next("A")
subject_test.on_next("B")
Python

subject_test 对象是通过调用 Subject() 创建的。subject_test对象有对on_next(value)、on_error(error)和on_completed()方法的引用。上述例子的输出如下所示

输出

E:\pyrx>python testrx.py
The value is A
The value is A
The value is B
The value is B
Python

我们可以使用on_completed()方法，来停止主题的执行，如下图所示。

例子

from rx.subject import Subject
subject_test = Subject()
subject_test.subscribe(
   lambda x: print("The value is {0}".format(x))
)
subject_test.subscribe(
   lambda x: print("The value is {0}".format(x))
)
subject_test.on_next("A")
subject_test.on_completed()
subject_test.on_next("B")
Python

一旦我们调用complete，后面调用的下一个方法就不会被调用。

输出

E:\pyrx>python testrx.py
The value is A
The value is A
Python

现在让我们看看，如何调用on_error(error)方法。

例子

from rx.subject import Subject
subject_test = Subject()
subject_test.subscribe(
   on_error = lambda e: print("Error : {0}".format(e))
)
subject_test.subscribe(
   on_error = lambda e: print("Error : {0}".format(e))
)
subject_test.on_error(Exception('There is an Error!'))
Python

输出

E:\pyrx>python testrx.py
Error: There is an Error!
Error: There is an Error!
Python

BehaviorSubject

BehaviorSubject在被调用时将给你最新的值。你可以创建行为主体，如下图所示

from rx.subject import BehaviorSubject
behavior_subject = BehaviorSubject("Testing Behaviour Subject"); // initialized the behaviour subject with value:Testing Behaviour Subject
Python

这里，是一个使用行为主体的工作实例

例子

from rx.subject import BehaviorSubject
behavior_subject = BehaviorSubject("Testing Behaviour Subject");
behavior_subject.subscribe(
   lambda x: print("Observer A : {0}".format(x))
)
behavior_subject.on_next("Hello")
behavior_subject.subscribe(
   lambda x: print("Observer B : {0}".format(x))
)
behavior_subject.on_next("Last call to Behaviour Subject")
Python

输出

E:\pyrx>python testrx.py
Observer A : Testing Behaviour Subject
Observer A : Hello
Observer B : Hello
Observer A : Last call to Behaviour Subject
Observer B : Last call to Behaviour Subject
Python

Repaysubject

Repaysubject类似于行为主体，它可以缓冲数值并将其重放给新的订阅者。这里，是一个重放主体的工作实例。

例子

from rx.subject import ReplaySubject
replay_subject = ReplaySubject(2)
replay_subject.subscribe(lambda x: print("Testing Replay Subject A: {0}".format(x)))
replay_subject.on_next(1)
replay_subject.on_next(2)
replay_subject.on_next(3)
replay_subject.subscribe(lambda x: print("Testing Replay Subject B: {0}".format(x)));
replay_subject.on_next(5)
Python

在重放主题上使用的缓冲值是2。因此，最后两个值将被缓冲并用于调用新的用户。

输出

E:\pyrx>python testrx.py
Testing Replay Subject A: 1
Testing Replay Subject A: 2
Testing Replay Subject A: 3
Testing Replay Subject B: 2
Testing Replay Subject B: 3
Testing Replay Subject A: 5
Testing Replay Subject B: 5
Python

AsyncSubject

在AsyncSubject的情况下，最后调用的值会传递给订阅者，而且只有在调用complete()方法后才会完成。

例子

from rx.subject import AsyncSubject
async_subject = AsyncSubject()
async_subject.subscribe(lambda x: print("Testing Async Subject A: {0}".format(x)))
async_subject.on_next(1)
async_subject.on_next(2)
async_subject.on_completed()
async_subject.subscribe(lambda x: print("Testing Async Subject B: {0}".format(x)))
Here, before complete is called, the last value passed to the subject is 2, and the same is given to the subscribers.
Python

输出

E:\pyrx>python testrx.py
Testing Async Subject A: 2
Testing Async Subject B: 2
Python

Python教程

Python 教程

Python 教程
Tkinter 教程

Tkinter 教程
Pandas 教程

Pandas 教程
NumPy 教程

NumPy 教程
Flask 教程

Flask 教程
Django 教程

Django 教程
PySpark 教程

PySpark 教程
wxPython 教程

wxPython 教程
SymPy 教程

SymPy 教程
Seaborn 教程

Seaborn 教程
SciPy 教程

SciPy 教程
RxPY 教程

RxPY 教程
Pycharm 教程

Pycharm 教程
Pygame 教程

Pygame 教程
PyGTK 教程

PyGTK 教程
PyQt 教程

PyQt 教程
PyQt5 教程

PyQt5 教程
PyTorch 教程

PyTorch 教程
Matplotlib 教程

Matplotlib 教程
Web2py 教程

Web2py 教程
BeautifulSoup 教程

BeautifulSoup 教程

Java教程

Java 教程

Java 教程

Web教程

HTML 教程

HTML 教程
CSS 教程

CSS 教程
CSS3 教程

CSS3 教程
jQuery 教程

jQuery 教程
Ajax 教程

Ajax 教程
AngularJS 教程

AngularJS 教程
TypeScript 教程

TypeScript 教程
WordPress 教程

WordPress 教程
Laravel 教程

Laravel 教程
Next.js 教程

Next.js 教程
PhantomJS 教程

PhantomJS 教程
Three.js 教程

Three.js 教程
Underscore.JS 教程

Underscore.JS 教程
WebGL 教程

WebGL 教程
WebRTC 教程

WebRTC 教程
VueJS 教程

VueJS 教程

数据库教程

SQL 教程

SQL 教程
MySQL 教程

MySQL 教程
MongoDB 教程

MongoDB 教程
PostgreSQL 教程

PostgreSQL 教程
SQLite 教程

SQLite 教程
Redis 教程

Redis 教程
MariaDB 教程

MariaDB 教程

图形图像教程

Vulkan 教程

Vulkan 教程
OpenCV 教程

OpenCV 教程

大数据教程

R语言 教程

R语言 教程

开发工具教程

Git 教程

Git 教程
VSCode 教程

VSCode 教程
Docker 教程

Docker 教程
Gerrit 教程

Gerrit 教程
Excel 教程

Excel 教程

计算机教程

Go语言 教程

Go语言 教程
C++ 教程

C++ 教程

RxPY教程

登录

找回密码？

注册