反應式編程

反應式編程是處理數據流和變化傳播的編程範例。這意味着，當數據流由一個組件發出時，更改將通過反應式編程庫傳播到其他組件。變化的傳播將持續到最終的接收者。事件驅動和反應式編程的區別在於，事件驅動式編程圍繞事件展開，反應式編程圍繞着數據展開。

ReactiveX或RX用於反應式編程
ReactiveX或者Raective Extension是反應式編程最着名的實現。 ReactiveX的工作取決於以下兩個類 -

可觀察的類
這個類是數據流或事件的來源，它打包傳入的數據，以便數據可以從一個線程傳遞到另一個線程。在某些觀察者訂閱它之前，它不會提供數據。

觀察員類
該類使用observable發出的數據流。可以有多個可觀察的觀察者，每個觀察者將接收每個發射的數據項。觀察者可以通過訂閱可觀察到的三種類型的事件 -

on_next()事件 - 它意味着數據流中有一個元素。
on_completed()事件 - 它意味着排放已經結束，沒有更多數據項到來。
on_error()事件 - 它也意味着排放的結束，但在可觀察到拋出錯誤的情況下。

RxPY - 用於反應式編程的Python模塊

RxPY是一個Python模塊，可用於反應式編程。我們需要確保模塊已安裝。以下命令可用於安裝RxPY模塊 -

pip install RxPY

例子
以下是一個Python腳本，它使用RxPY模塊及Observable類和Observe類來進行反應式編程。基本上有兩類 -

get_strings() - 用於從觀察者獲取字符串。
PrintObserver() - 用於從觀察者打印字符串。它使用觀察員班的所有三個事件。它也使用subscribe()類。

參考以下實現代碼 -

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

執行上面示例代碼，得到以下結果 -

Received Ram
Received Mohan
Received Shyam
Finished

用於反應式編程的PyFunctional庫

PyFunctionalis是另一個可用於響應式編程的Python庫。它使我們能夠使用Python編程語言創建功能程序。這很有用，因爲它允許我們通過使用鏈式函數操作符來創建數據管道。

RxPY和PyFunctional之間的區別

這兩個庫都用於響應式編程，並以類似的方式處理流，但兩者的主要區別取決於數據的處理。 RxPY處理系統中的數據和事件，而PyFunctional專注於使用函數式編程範例轉換數據。

安裝PyFunctional模塊

需要在使用之前安裝這個模塊。可以通過以下pip命令來安裝 -

pip install pyfunctional

例子
以下示例使用PyFunctional模塊及其seq類，它們充當可以迭代和操作的流對象。在這個程序中，它使用將每個值加倍的lamda函數映射序列，然後過濾x大於4的值，最後將序列減少爲所有剩餘值的和。

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

執行上面示例代碼，得到以下結果 -

Result: 6

Python併發編程教程

RxPY - 用於反應式編程的Python模塊

用於反應式編程的PyFunctional庫