第淺談Python響應(yīng)式類庫(kù)RxPy目錄一、基本概念1.1、Observable和Observer（可觀察對(duì)象和觀察者）1.2、Operator（操作符）1.3、Single（單例）1.4、Subject（主體）1.5、Scheduler（調(diào)度器）1.6、Observer和Observable1.7、操作符1.8、創(chuàng)建型操作符1.9、過(guò)濾型操作符1.10、轉(zhuǎn)換型操作符1.11、算術(shù)操作符1.12、Subject1.13、ReplaySubject1.14、BehaviorSubject1.15、AsyncSubject1.16、Scheduler二、應(yīng)用場(chǎng)景2.1、防止重復(fù)發(fā)送2.2、操作數(shù)據(jù)流

一、基本概念

ReactiveX中有幾個(gè)核心的概念，先來(lái)簡(jiǎn)單介紹一下。

1.1、Observable和Observer（可觀察對(duì)象和觀察者）

首先是Observable和Observer，它們分別是可觀察對(duì)象和觀察者。Observable可以理解為一個(gè)異步的數(shù)據(jù)源，會(huì)發(fā)送一系列的值。Observer則類似于消費(fèi)者，需要先訂閱Observable，然后才可以接收到其發(fā)射的值。可以說(shuō)這組概念是設(shè)計(jì)模式中的觀察者模式和生產(chǎn)者-消費(fèi)者模式的綜合體。

1.2、Operator（操作符）

另外一個(gè)非常重要的概念就是操作符了。操作符作用于Observable的數(shù)據(jù)流上，可以對(duì)其施加各種各樣的操作。更重要的是，操作符還可以鏈?zhǔn)浇M合起來(lái)。這樣的鏈?zhǔn)胶瘮?shù)調(diào)用不僅將數(shù)據(jù)和操作分隔開(kāi)來(lái)，而且代碼更加清晰可讀。一旦熟練掌握之后，你就會(huì)愛(ài)上這種感覺(jué)的。

1.3、Single（單例）

在RxJava和其變體中，還有一個(gè)比較特殊的概念叫做Single，它是一種只會(huì)發(fā)射同一個(gè)值的Observable，說(shuō)白了就是單例。當(dāng)然如果你對(duì)Java等語(yǔ)言比較熟悉，那么單例想必也很熟悉。

1.4、Subject（主體）

主體這個(gè)概念非常特殊，它既是Observable又是Observer。正是因?yàn)檫@個(gè)特點(diǎn)，所以Subject可以訂閱其他Observable，也可以將發(fā)射對(duì)象給其他Observer。在某些場(chǎng)景中，Subject會(huì)有很大的作用。

1.5、Scheduler（調(diào)度器）

默認(rèn)情況下ReactiveX只運(yùn)行在當(dāng)前線程下，但是如果有需要的話，也可以用調(diào)度器來(lái)讓ReactiveX運(yùn)行在多線程環(huán)境下。有很多調(diào)度器和對(duì)應(yīng)的操作符，可以處理多線程場(chǎng)景下的各種要求。

1.6、Observer和Observable

先來(lái)看看一個(gè)最簡(jiǎn)單的例子，運(yùn)行的結(jié)果會(huì)依次打印這些數(shù)字。這里的of是一個(gè)操作符，可以根據(jù)給定的參數(shù)創(chuàng)建一個(gè)新的Observable。創(chuàng)建之后，就可以訂閱Observable，三個(gè)回調(diào)方法在對(duì)應(yīng)的時(shí)機(jī)執(zhí)行。一旦Observer訂閱了Observable，就會(huì)接收到后續(xù)Observable發(fā)射的各項(xiàng)值。

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')

)

這個(gè)例子看起來(lái)好像很簡(jiǎn)單，并且看起來(lái)沒(méi)什么用。但是當(dāng)你了解了Rx的一些核心概念，就會(huì)理解到這是一個(gè)多么強(qiáng)大的工具。更重要的是，Observable生成數(shù)據(jù)和訂閱的過(guò)程是異步的，如果你熟悉的話，就可以利用這個(gè)特性做很多事情。

1.7、操作符

在RxPy中另一個(gè)非常重要的概念就是操作符了，甚至可以說(shuō)操作符就是最重要的一個(gè)概念了。幾乎所有的功能都可以通過(guò)組合各個(gè)操作符來(lái)實(shí)現(xiàn)。熟練掌握操作符就是學(xué)好RxPy的關(guān)鍵了。操作符之間也可以用pipe函數(shù)連接起來(lái)，構(gòu)成復(fù)雜的操作鏈。

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中有大量操作符，可以完成各種各樣的功能。我們來(lái)簡(jiǎn)單看看其中一些常用的操作符。如果你熟悉Java8的流類庫(kù)或者其他函數(shù)式編程類庫(kù)的話，應(yīng)該對(duì)這些操作符感到非常親切。

1.8、創(chuàng)建型操作符

首先是創(chuàng)建Observable的操作符，列舉了一些比較常用的創(chuàng)建型操作符。

1.9、過(guò)濾型操作符

過(guò)濾型操作符的主要作用是對(duì)Observable進(jìn)行篩選和過(guò)濾。

1.10、轉(zhuǎn)換型操作符

1.11、算術(shù)操作符

1.12、Subject

Subject是一種特殊的對(duì)象，它既是Observer又是Observable。不過(guò)這個(gè)對(duì)象一般不太常用，但是假如某些用途還是很有用的。所以還是要介紹一下。下面的代碼，因?yàn)橛嗛喌臅r(shí)候第一個(gè)值已經(jīng)發(fā)射出去了，所以只會(huì)打印訂閱之后才發(fā)射的值。

fromrx.subjectimportSubject,AsyncSubject,BehaviorSubject,ReplaySubject

#Subject同時(shí)是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

另外還有幾個(gè)特殊的Subject，下面來(lái)介紹一下。

1.13、ReplaySubject

ReplaySubject是一個(gè)特殊的Subject，它會(huì)記錄所有發(fā)射過(guò)的值，不論什么時(shí)候訂閱的。所以它可以用來(lái)當(dāng)做緩存來(lái)使用。ReplaySubject還可以接受一個(gè)bufferSize參數(shù)，指定可以緩存的最近數(shù)據(jù)數(shù)，默認(rèn)情況下是全部。

下面的代碼和上面的代碼幾乎完全一樣，但是因?yàn)槭褂昧薘eplaySubject，所以所有的值都會(huì)被打印。當(dāng)然大家也可以試試把訂閱語(yǔ)句放到其他位置，看看輸出是否會(huì)產(chǎn)生變化。

#ReplaySubject會(huì)緩存所有值，如果指定參數(shù)的話只會(huì)緩存最近的幾個(gè)值

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是一個(gè)特殊的Subject，它只會(huì)記錄最近一次發(fā)射的值。而且在創(chuàng)建它的時(shí)候，必須指定一個(gè)初始值，所有訂閱它的對(duì)象都可以接收到這個(gè)初始值。當(dāng)然如果訂閱的晚了，這個(gè)初始值同樣會(huì)被后面發(fā)射的值覆蓋，這一點(diǎn)要注意。

#BehaviorSubject會(huì)緩存上次發(fā)射的值，除非Observable已經(jīng)關(guān)閉

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是一個(gè)特殊的Subject，顧名思義它是一個(gè)異步的Subject，它只會(huì)在Observer完成的時(shí)候發(fā)射數(shù)據(jù)，而且只會(huì)發(fā)射最后一個(gè)數(shù)據(jù)。因此下面的代碼僅僅會(huì)輸出4.假如注釋掉最后一行co_completed調(diào)用，那么什么也不會(huì)輸出。

#AsyncSubject會(huì)緩存上次發(fā)射的值，而且僅會(huì)在Observable關(guān)閉后開(kāi)始發(fā)射

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算是異步的框架，但是其實(shí)它默認(rèn)還是運(yùn)行在單個(gè)線程之上的，因此如果使用了某些會(huì)阻礙線程運(yùn)行的操作，那么程序就會(huì)卡死。當(dāng)然針對(duì)這些情況，我們就可以使用其他的Scheduler來(lái)調(diào)度任務(wù)，保證程序能夠高效運(yùn)行。

下面的例子創(chuàng)建了一個(gè)ThreadPoolScheduler，它是