浅谈Python响应式类库RxPy

第浅谈Python响应式类库RxPy目录一、基本概念1.1、Observable和Observer（可观察对象和观察者）1.2、Operator（操作符）1.3、Single（单例）1.4、Subject（主体）1.5、Scheduler（调度器）1.6、Observer和Observable1.7、操作符1.8、创建型操作符1.9、过滤型操作符1.10、转换型操作符1.11、算术操作符1.12、Subject1.13、ReplaySubject1.14、BehaviorSubject1.15、AsyncSubject1.16、Scheduler二、应用场景2.1、防止重复发送2.2、操作数据流

一、基本概念

ReactiveX中有几个核心的概念，先来简单介绍一下。

1.1、Observable和Observer（可观察对象和观察者）

首先是Observable和Observer，它们分别是可观察对象和观察者。Observable可以理解为一个异步的数据源，会发送一系列的值。Observer则类似于消费者，需要先订阅Observable，然后才可以接收到其发射的值。可以说这组概念是设计模式中的观察者模式和生产者-消费者模式的综合体。

1.2、Operator（操作符）

另外一个非常重要的概念就是操作符了。操作符作用于Observable的数据流上，可以对其施加各种各样的操作。更重要的是，操作符还可以链式组合起来。这样的链式函数调用不仅将数据和操作分隔开来，而且代码更加清晰可读。一旦熟练掌握之后，你就会爱上这种感觉的。

1.3、Single（单例）

在RxJava和其变体中，还有一个比较特殊的概念叫做Single，它是一种只会发射同一个值的Observable，说白了就是单例。当然如果你对Java等语言比较熟悉，那么单例想必也很熟悉。

1.4、Subject（主体）

主体这个概念非常特殊，它既是Observable又是Observer。正是因为这个特点，所以Subject可以订阅其他Observable，也可以将发射对象给其他Observer。在某些场景中，Subject会有很大的作用。

1.5、Scheduler（调度器）

默认情况下ReactiveX只运行在当前线程下，但是如果有需要的话，也可以用调度器来让ReactiveX运行在多线程环境下。有很多调度器和对应的操作符，可以处理多线程场景下的各种要求。

1.6、Observer和Observable

先来看看一个最简单的例子，运行的结果会依次打印这些数字。这里的of是一个操作符，可以根据给定的参数创建一个新的Observable。创建之后，就可以订阅Observable，三个回调方法在对应的时机执行。一旦Observer订阅了Observable，就会接收到后续Observable发射的各项值。

fromrximportof

ob=of(1,2,34,5,6,7,7)

ob.subscribe(

on_next=lambdai:print(f'Received:{i}'),

on_error=lambdae:print(f'Error:{e}'),

on_completed=lambda:print('Completed')

)

这个例子看起来好像很简单，并且看起来没什么用。但是当你了解了Rx的一些核心概念，就会理解到这是一个多么强大的工具。更重要的是，Observable生成数据和订阅的过程是异步的，如果你熟悉的话，就可以利用这个特性做很多事情。

1.7、操作符

在RxPy中另一个非常重要的概念就是操作符了，甚至可以说操作符就是最重要的一个概念了。几乎所有的功能都可以通过组合各个操作符来实现。熟练掌握操作符就是学好RxPy的关键了。操作符之间也可以用pipe函数连接起来，构成复杂的操作链。

fromrximportof,operatorsasop

importrx

ob=of(1,2,34,5,6,7,7)

ob.pipe(

op.map(lambdai:i**2),

op.filter(lambdai:i=10)

).subscribe(lambdai:print(f'Received:{i}'))

在RxPy中有大量操作符，可以完成各种各样的功能。我们来简单看看其中一些常用的操作符。如果你熟悉Java8的流类库或者其他函数式编程类库的话，应该对这些操作符感到非常亲切。

1.8、创建型操作符

首先是创建Observable的操作符，列举了一些比较常用的创建型操作符。

1.9、过滤型操作符

过滤型操作符的主要作用是对Observable进行筛选和过滤。

1.10、转换型操作符

1.11、算术操作符

1.12、Subject

Subject是一种特殊的对象，它既是Observer又是Observable。不过这个对象一般不太常用，但是假如某些用途还是很有用的。所以还是要介绍一下。下面的代码，因为订阅的时候第一个值已经发射出去了，所以只会打印订阅之后才发射的值。

fromrx.subjectimportSubject,AsyncSubject,BehaviorSubject,ReplaySubject

#Subject同时是Observer和Observable

print('--------Subject---------')

subject=Subject()

subject.on_next(1)

subject.subscribe(lambdai:print(i))

subject.on_next(2)

subject.on_next(3)

subject.on_next(4)

subject.on_completed()

#234

另外还有几个特殊的Subject，下面来介绍一下。

1.13、ReplaySubject

ReplaySubject是一个特殊的Subject，它会记录所有发射过的值，不论什么时候订阅的。所以它可以用来当做缓存来使用。ReplaySubject还可以接受一个bufferSize参数，指定可以缓存的最近数据数，默认情况下是全部。

下面的代码和上面的代码几乎完全一样，但是因为使用了ReplaySubject，所以所有的值都会被打印。当然大家也可以试试把订阅语句放到其他位置，看看输出是否会产生变化。

#ReplaySubject会缓存所有值，如果指定参数的话只会缓存最近的几个值

print('--------ReplaySubject---------')

subject=ReplaySubject()

subject.on_next(1)

subject.subscribe(lambdai:print(i))

subject.on_next(2)

subject.on_next(3)

subject.on_next(4)

subject.on_completed()

#1234

1.14、BehaviorSubject

BehaviorSubject是一个特殊的Subject，它只会记录最近一次发射的值。而且在创建它的时候，必须指定一个初始值，所有订阅它的对象都可以接收到这个初始值。当然如果订阅的晚了，这个初始值同样会被后面发射的值覆盖，这一点要注意。

#BehaviorSubject会缓存上次发射的值，除非Observable已经关闭

print('--------BehaviorSubject---------')

subject=BehaviorSubject(0)

subject.on_next(1)

subject.on_next(2)

subject.subscribe(lambdai:print(i))

subject.on_next(3)

subject.on_next(4)

subject.on_completed()

#234

1.15、AsyncSubject

AsyncSubject是一个特殊的Subject，顾名思义它是一个异步的Subject，它只会在Observer完成的时候发射数据，而且只会发射最后一个数据。因此下面的代码仅仅会输出4.假如注释掉最后一行co_completed调用，那么什么也不会输出。

#AsyncSubject会缓存上次发射的值，而且仅会在Observable关闭后开始发射

print('--------AsyncSubject---------')

subject=AsyncSubject()

subject.on_next(1)

subject.on_next(2)

subject.subscribe(lambdai:print(i))

subject.on_next(3)

subject.on_next(4)

subject.on_completed()

#4

1.16、Scheduler

虽然RxPy算是异步的框架，但是其实它默认还是运行在单个线程之上的，因此如果使用了某些会阻碍线程运行的操作，那么程序就会卡死。当然针对这些情况，我们就可以使用其他的Scheduler来调度任务，保证程序能够高效运行。

下面的例子创建了一个ThreadPoolScheduler，它是