RxPY-クイックガイド

この章では、リアクティブプログラミングとは何か、RxPYとは何か、その演算子、機能、長所と短所について説明します。

リアクティブプログラミングとは何ですか？

リアクティブプログラミングはプログラミングパラダイムであり、データフローと変更の伝播を扱います。これは、データフローが1つのコンポーネントによって発行されると、変更がリアクティブプログラミングライブラリによって他のコンポーネントに伝播されることを意味します。変更の伝播は、最終的な受信者に到達するまで続きます。

RxPYを使用すると、非同期データストリームを適切に制御できます。たとえば、URLに対して行われたリクエストは、observableを使用して追跡でき、オブザーバーを使用して、リクエストが完了したときに応答またはエラーをリッスンできます。

RxPYは、を使用して非同期データストリームを処理することを提供します Observables、を使用してデータストリームをクエリします Operators つまり、フィルター、合計、連結、マップを行い、データストリームの同時実行性を利用します。 Schedulers。Observableを作成すると、on_next（v）、on_error（e）、およびon_completed（）メソッドを持つオブザーバーオブジェクトが提供されます。subscribed イベントが発生したときに通知を受け取るようにします。

Observableは、パイプ演算子を使用して、チェーン形式の複数の演算子を使用してクエリできます。

RxPYは、次のようなさまざまなカテゴリの演算子を提供します。

• 数学演算子

• 変換演算子

• フィルタリング演算子

• エラー処理演算子

• ユーティリティオペレーター

• 条件付き演算子

• 生成演算子

• 接続可能なオペレーター

これらの演算子については、このチュートリアルで詳しく説明します。

RxPyとは何ですか？

RxPYは次のように定義されます a library for composing asynchronous and event-based programs using observable collections and pipable query operators in Python RxPyの公式ウェブサイトによると、 https://rxpy.readthedocs.io/en/latest/.

RxPYは、リアクティブプログラミングをサポートするPythonライブラリです。RxPyはの略ですReactive Extensions for Python。これは、オブザーバブルを使用して、非同期データ呼び出し、コールバック、およびイベントベースのプログラムを処理するリアクティブプログラミングを処理するライブラリです。

RxPyの機能

RxPyでは、次の概念が非同期タスクの処理を処理します-

観察可能

オブザーバブルは、オブザーバーを作成し、それを、たとえばツイート、コンピューター関連のイベントなどから期待されるデータストリームを持つソースにアタッチする関数です。

観察者

これは、on_next（）、on_error（）、およびon_completed（）メソッドを持つオブジェクトであり、オブザーバブルとの相互作用がある場合、つまりソースが受信ツイートの例などに対して相互作用する場合に呼び出されます。

サブスクリプション

オブザーバブルが作成されたら、オブザーバブルを実行するためにサブスクライブする必要があります。

演算子

演算子は、オブザーバブルを入力として取り込む純粋関数であり、出力もオブザーバブルです。パイプ演算子を使用すると、監視可能なデータに対して複数の演算子を使用できます。

件名

サブジェクトは、監視可能なシーケンスであり、マルチキャストできるオブザーバーです。つまり、サブスクライブしている多くのオブザーバーと話します。サブジェクトはコールドオブザーバブルです。つまり、値はサブスクライブされたオブザーバー間で共有されます。

スケジューラー

RxPyの重要な機能の1つは同時実行性です。つまり、タスクを並行して実行できるようにします。これを実現するために、RxPyには、スケジューラーと連携してサブスクライブされたタスクの実行を決定する2つの演算子subscribe_on（）とobserve_on（）があります。

RxPYを使用する利点

RxPy −の利点は次のとおりです。

• RxPYは、非同期データストリームとイベントの処理に関しては素晴らしいライブラリです。RxPYは、オブザーバブルを使用して、非同期データ呼び出し、コールバック、およびイベントベースのプログラムを処理するリアクティブプログラミングを処理します。

• RxPYは、数学、変換、フィルタリング、ユーティリティ、条件付き、エラー処理、結合カテゴリの演算子の膨大なコレクションを提供し、リアクティブプログラミングで使用すると作業が楽になります。

• 並行性、つまり複数のタスクを一緒に処理することは、RxPYのスケジューラーを使用して実現されます。

• RxPYを使用すると、非同期タスクの処理と並列処理が容易になるため、パフォーマンスが向上します。

RxPYを使用するデメリット

• オブザーバブルを使用してコードをデバッグするのは少し難しいです。

この章では、RxPyのインストールについて説明します。RxPYの使用を開始するには、最初にPythonをインストールする必要があります。それで、私たちは以下に取り組むつもりです-

• Pythonをインストールする
• RxPyをインストールします

Pythonのインストール

Pythonの公式サイトにアクセスします。 https://www.python.org/downloads/.以下に示すように、Windows、Linux / Unix、およびmacosで利用可能な最新バージョンをクリックします。利用可能な64ビットまたは32ビットOSに従ってPythonをダウンロードします。

ダウンロードしたら、をクリックします .exe file 手順に従って、システムにpythonをインストールします。

Pythonパッケージマネージャー、つまりpipも、上記のインストールでデフォルトでインストールされます。システム上でグローバルに機能させるには、Pythonの場所をPATH変数に直接追加します。これは、インストールの開始時に表示されます。チェックボックスをオンにすることを忘れないでください。チェックを忘れた場合は、以下の手順でPATHに追加してください。

PATHに追加するには、以下の手順に従います-

コンピュータアイコンを右クリックし、プロパティ→システムの詳細設定をクリックします。

以下のような画面が表示されます-

上記のように環境変数をクリックします。以下のような画面が表示されます-

[パス]を選択して[編集]ボタンをクリックし、最後にPythonのロケーションパスを追加します。それでは、Pythonのバージョンを確認しましょう。

Pythonバージョンの確認

E:\pyrx>python --version
Python 3.7.3

RxPYをインストールします

Pythonがインストールされたので、RxPyをインストールします。

Pythonがインストールされると、Pythonパッケージマネージャー、つまりpipもインストールされます。以下は、pipのバージョンを確認するコマンドです-

E:\pyrx>pip --version
pip 19.1.1 from c:\users\xxxx\appdata\local\programs\python\python37\lib\site-
packages\pip (python 3.7)

pipがインストールされており、バージョンは 19.1.1. 次に、pipを使用してRxPyをインストールします

コマンドは次のとおりです-

pip install rx

このチュートリアルでは、RxPYバージョン3とPythonバージョン3.7.3を使用しています。RxPYバージョン3の動作は、以前のバージョン、つまりRxPYバージョン1とは少し異なります。

この章では、2つのバージョンの違いと、PythonバージョンとRxPYバージョンを更新する場合に行う必要のある変更について説明します。

RxPYで観測可能

RxPyバージョン1では、Observableは別のクラスでした-

from rx import Observable

Observableを使用するには、次のように使用する必要があります-

Observable.of(1,2,3,4,5,6,7,8,9,10)

RxPyバージョン3では、Observableは直接rxパッケージの一部です。

Example

import rx
rx.of(1,2,3,4,5,6,7,8,9,10)

RxPyの演算子

バージョン1では、演算子はObservableクラスのメソッドでした。たとえば、演算子を使用するには、以下に示すようにObservableをインポートする必要があります-

from rx import Observable

演算子は、たとえば、以下に示すようにObservable.operatorとして使用されます。

Observable.of(1,2,3,4,5,6,7,8,9,10)\
   .filter(lambda i: i %2 == 0) \
   .sum() \
   .subscribe(lambda x: print("Value is {0}".format(x)))

RxPYバージョン3の場合、演算子は関数であり、次のようにインポートおよび使用されます。

import rx
from rx import operators as ops
rx.of(1,2,3,4,5,6,7,8,9,10).pipe(
   ops.filter(lambda i: i %2 == 0),
   ops.sum()
).subscribe(lambda x: print("Value is {0}".format(x)))

Pipe（）メソッドを使用した演算子の連鎖

RxPyバージョン1では、オブザーバブルで複数の演算子を使用する必要がある場合、次のように実行する必要がありました。

Example

from rx import Observable
Observable.of(1,2,3,4,5,6,7,8,9,10)\
   .filter(lambda i: i %2 == 0) \
   .sum() \
   .subscribe(lambda x: print("Value is {0}".format(x)))

ただし、RxPYバージョン3の場合は、以下に示すように、pipe（）メソッドと複数の演算子を使用できます。

Example

import rx
from rx import operators as ops
rx.of(1,2,3,4,5,6,7,8,9,10).pipe(
   ops.filter(lambda i: i %2 == 0),
   ops.sum()
).subscribe(lambda x: print("Value is {0}".format(x)))

オブザーバブルは、オブザーバーを作成し、値が期待されるソース（クリック、dom要素からのマウスイベントなど）にアタッチする関数です。

この章では、以下のトピックについて詳しく説明します。

• オブザーバブルを作成する

• Observableをサブスクライブして実行する

オブザーバブルを作成する

オブザーバブルを作成するには、 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).

Observableをサブスクライブして実行する

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（）メソッドは、observableの実行を処理します。コールバック関数on_next、 on_error そして on_completedsubscribeメソッドに渡す必要があります。次に、サブスクライブメソッドを呼び出すと、test_observable（）関数が実行されます。

3つのコールバック関数すべてをsubscribe（）メソッドに渡す必要はありません。要件に応じて、on_next（）、on_error（）、およびon_completed（）を渡すことができます。

ラムダ関数は、on_next、on_error、およびon_completedに使用されます。引数を取り込んで、指定された式を実行します。

これが、作成されたオブザーバブルの出力です。

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

この章では、RxPYの演算子について詳しく説明します。これらの演算子には次のものが含まれます-

• オペレーターとの協力
• 数学演算子
• 変換演算子
• フィルタリング演算子
• エラー処理演算子
• ユーティリティオペレーター
• 条件付き演算子
• 生成演算子
• 接続可能なオペレーター
• 演算子の組み合わせ

リアクティブ（Rx）pythonにはほとんど多くの演算子があり、Pythonコーディングで簡単に作業できます。これらの複数の演算子を一緒に使用できます。たとえば、文字列を操作するときに、マップ、フィルター、マージ演算子を使用できます。

オペレーターとの協力

pipe（）メソッドを使用して、複数の演算子を一緒に操作できます。この方法では、複数の演算子をチェーン化できます。

これは、演算子を使用する実際の例です-

test = of(1,2,3) // an observable
subscriber = test.pipe(
   op1(),
   op2(),
   op3()
)

上記の例では、値1、2、および3を受け取るobservable using of（）メソッドを作成しました。これで、このobservableで、示されているように、pipe（）メソッドを使用して任意の数の演算子を使用して別の操作を実行できます。上記。演算子の実行は、指定されたオブザーバブルで順番に実行されます。

演算子を操作するには、最初に以下に示すようにインポートします-

from rx import of, operators as op

ここに、実際の例があります-

testrx.py

from rx import of, operators as op
test = of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
sub1 = test.pipe(
   op.filter(lambda s: s%2==0),
   op.reduce(lambda acc, x: acc + x)
)
sub1.subscribe(lambda x: print("Sum of Even numbers is {0}".format(x)))

上記の例では、数値のリストがあり、そこからフィルター演算子を使用して偶数をフィルタリングし、後でリデュース演算子を使用してそれを追加しています。

Output

E:\pyrx>python testrx.py
Sum of Even numbers is 30

これが私たちが議論しようとしている演算子のリストです-

• オブザーバブルの作成
• 数学演算子
• 変換演算子
• フィルタリング演算子
• エラー処理演算子
• ユーティリティオペレーター
• Conditional
• Connectable
• 演算子の組み合わせ

オブザーバブルの作成

以下はオブザーバブルです。作成カテゴリで説明します。

例を表示

観察可能	説明
作成する	このメソッドは、オブザーバブルを作成するために使用されます。
空の	このオブザーバブルは何も出力せず、完全な状態を直接放出します。
決して	このメソッドは、完全な状態に達することのないオブザーバブルを作成します。
スロー	このメソッドは、エラーをスローするオブザーバブルを作成します。
から_	このメソッドは、指定された配列またはオブジェクトをオブザーバブルに変換します。
間隔	このメソッドは、タイムアウト後に生成された一連の値を提供します。
ただ	このメソッドは、指定された値をオブザーバブルに変換します。
範囲	このメソッドは、与えられた入力に基づいて整数の範囲を与えます。
repeat_value	このメソッドは、カウントが指定されたとおりに指定された値を繰り返すオブザーバブルを作成します。
開始	このメソッドは、関数を入力として受け取り、入力関数から値を返すオブザーバブルを返します。
タイマー	このメソッドは、タイムアウトが完了した後、値を順番に出力します。

数学演算子

数学演算子のカテゴリで説明する演算子は次のとおりです。

例を表示

オペレーター	説明
平均	この演算子は、指定されたソースオブザーバブルから平均を計算し、平均値を持つオブザーバブルを出力します。
concat	この演算子は、2つ以上のオブザーバブルを取り込み、シーケンス内のすべての値を含む1つのオブザーバブルを指定します。
カウント	この演算子は、値を持つObservableを取り込み、それを単一の値を持つObservableに変換します。count関数は、オプションの引数として述語関数を取ります。 この関数はブール型であり、条件を満たす場合にのみ出力に値を追加します。
最大	この演算子は、ソースオブザーバブルからの最大値を持つオブザーバブルを提供します。
分	この演算子は、ソースオブザーバブルからの最小値でオブザーバブルを提供します。
減らす	この演算子は、ソースobservableからの値に使用されるアキュムレータ関数と呼ばれる関数を取り込んで、アキュムレータ関数に渡されるオプションのシード値とともに、累積値をobservableの形式で返します。
和	この演算子は、ソースオブザーバブルからのすべての値の合計を含むオブザーバブルを返します。

変換演算子

変換演算子のカテゴリで説明する演算子は、以下のとおりです。

例を表示

オペレーター	カテゴリー
バッファ	この演算子は、監視可能なソースからすべての値を収集し、指定された境界条件が満たされると、定期的にそれらを出力します。
Ground_by	この演算子は、指定されたkey_mapper関数に基づいて、監視可能なソースからの値をグループ化します。
地図	この演算子は、指定されたmapper_funcの出力に基づいて、監視可能なソースからの各値を新しい値に変更します。
スキャン	この演算子は、ソースobservableからの値にアキュムレータ関数を適用し、新しい値でobservableを返します。

フィルタリング演算子

フィルタリング演算子カテゴリで説明する演算子を以下に示します-

例を表示

オペレーター	カテゴリー
デバウンス	この演算子は、指定されたタイムスパンまで、ソースからの値を監視可能に指定し、残りの時間は無視します。
明確な	この演算子は、観測可能なソースとは異なるすべての値を提供します。
element_at	この演算子は、指定されたインデックスに対して監視可能なソースからの要素を提供します。
フィルタ	この演算子は、指定された述語関数に基づいて、監視可能なソースから値をフィルタリングします。
最初	この演算子は、ソースからの最初の要素を監視可能にします。
ignore_elements	この演算子は、ソースobservableからのすべての値を無視し、コールバック関数を完了するかエラーにするための呼び出しのみを実行します。
最終	この演算子は、ソースからの最後の要素を監視可能にします。
スキップ	この演算子は、入力として取得されたカウント項目の最初の出現をスキップするオブザーバブルを返します。
skip_last	この演算子は、入力として取得されたカウント項目の最後の出現をスキップするオブザーバブルを返します。
取る	この演算子は、指定されたカウントに基づいて、連続した順序でソース値のリストを提供します。
take_last	この演算子は、指定されたカウントに基づいて、最後から連続した順序でソース値のリストを提供します。

エラー処理演算子

エラー処理演算子のカテゴリで説明する演算子は次のとおりです。

例を表示

オペレーター	説明
キャッチ	この演算子は、例外が発生したときに監視可能なソースを終了します。
リトライ	このオペレーターは、エラーが発生したときに監視可能なソースで再試行し、再試行カウントが完了すると終了します。

ユーティリティオペレーター

以下は、ユーティリティ演算子のカテゴリで説明する演算子です。

例を表示

オペレーター	説明
ディレイ	このオペレーターは、指定された時間または日付に従って、発生源の観測可能な放出を遅らせます。
マテリアライズ	この演算子は、明示的な通知値の形式で発行された値を使用して、監視可能なソースからの値を変換します。
時間間隔	この演算子は、観測可能なソースからの値の間の経過時間を示します。
タイムアウト	この演算子は、経過時間後に監視可能なソースからのすべての値を提供します。そうでない場合、エラーがトリガーされます。
タイムスタンプ	この演算子は、ソースの監視可能なすべての値にタイムスタンプを添付します。

条件演算子とブール演算子

条件演算子とブール演算子のカテゴリで説明する演算子は、次のとおりです。

例を表示

オペレーター	説明
すべて	この演算子は、ソースオブザーバブルからのすべての値が指定された条件を満たすかどうかを確認します。
含まれています	この演算子は、指定された値が存在し、それがソースのオブザーバブルの値である場合、値がtrueまたはfalseのオブザーバブルを返します。
default_if_empty	ソースオブザーバブルが空の場合、この演算子はデフォルト値を返します。
sequence_equal	この演算子は、オブザーバブルの2つのシーケンスまたは値の配列を比較し、値がtrueまたはfalseのオブザーバブルを返します。
skip_until	この演算子は、2番目のオブザーバブルが値を発行するまで、ソースオブザーバブルから値を破棄します。
skip_while	この演算子は、渡された条件を満たすソースオブザーバブルからの値を持つオブザーバブルを返します。
take_until	この演算子は、2番目のオブザーバブルが値を発行するか終了した後、ソースオブザーバブルから値を破棄します。
take_while	この演算子は、条件が失敗したときに監視可能なソースから値を破棄します。

接続可能な演算子

接続可能な演算子のカテゴリで説明する演算子は次のとおりです。

例を表示

オペレーター	説明
公開する	このメソッドは、オブザーバブルを接続可能なオブザーバブルに変換します。
ref_count	この演算子は、オブザーバブルを通常のオブザーバブルにします。
リプレイ	このメソッドは、replaySubjectと同様に機能します。このメソッドは、オブザーバブルがすでに発行されていて、一部のサブスクライバーがサブスクライブに遅れている場合でも、同じ値を返します。

演算子の組み合わせ

以下は、結合演算子カテゴリで説明する演算子です。

例を表示

オペレーター	説明
Combine_latest	この演算子は、入力として指定されたオブザーバブルのタプルを作成します。
マージ	この演算子は、指定されたオブザーバブルをマージします。
皮切りに	この演算子は、指定された値を取り込んで、ソースの開始時に監視可能なリターンを追加して完全なシーケンスを返します。
zip	この演算子は、指定されたオブザーバブルの最初の値を取得することによって形成されるタプル形式の値を持つオブザーバブルを返します。

サブジェクトは、監視可能なシーケンスであり、マルチキャストできるオブザーバーです。つまり、サブスクライブしている多くのオブザーバーと話します。

主題について以下のトピックについて議論します-

• 件名を作成する
• 件名を購読する
• 被験者にデータを渡す
• BehaviorSubject
• ReplaySubject
• AsyncSubject

件名を作成する

サブジェクトを操作するには、以下に示すようにサブジェクトをインポートする必要があります-

from rx.subject import Subject

次のようにサブジェクトオブジェクトを作成できます-

subject_test = Subject()

オブジェクトは、3つのメソッドを持つオブザーバーです-

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

サブジェクトへのデータの受け渡し

以下に示すように、on_next（value）メソッドを使用して作成されたサブジェクトにデータを渡すことができます-

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

データはすべてのサブスクリプションに渡され、件名に追加されます。

これは、主題の実際の例です。

例

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

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

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

completeを呼び出すと、後で呼び出される次のメソッドは呼び出されません。

出力

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

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

出力

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

BehaviorSubject

BehaviorSubjectは、呼び出されたときに最新の値を提供します。以下に示すように行動サブジェクトを作成できます-

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

これは、BehaviorSubjectを使用するための実用的な例です

例

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

出力

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

件名を再生

replaysubjectは動作subjectに似ており、値をバッファリングして、新しいサブスクライバーに対して同じものを再生できます。これは、リプレイサブジェクトの実例です。

例

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)

使用されるバッファ値は、再生対象で2です。したがって、最後の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

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.

出力

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

RxPyの重要な機能の1つは同時実行性です。つまり、タスクを並行して実行できるようにします。これを実現するために、サブスクライブされたタスクの実行を決定するスケジューラーと連携する2つの演算子subscribe_on（）とobserve_on（）があります。

これは、subscibe_on（）、observe_on（）、およびスケジューラーの必要性を示す実用的な例です。

例

import random
import time
import rx
from rx import operators as ops
def adding_delay(value):
   time.sleep(random.randint(5, 20) * 0.1)
   return value
# Task 1
rx.of(1,2,3,4,5).pipe(
   ops.map(lambda a: adding_delay(a))
).subscribe(
   lambda s: print("From Task 1: {0}".format(s)),
   lambda e: print(e),
   lambda: print("Task 1 complete")
)
# Task 2
rx.range(1, 5).pipe(
   ops.map(lambda a: adding_delay(a))
).subscribe(
   lambda s: print("From Task 2: {0}".format(s)),
   lambda e: print(e),
   lambda: print("Task 2 complete")
) 
input("Press any key to exit\n")

上記の例では、タスク1とタスク2の2つのタスクがあります。タスクの実行は順番に行われます。2番目のタスクは、最初のタスクが完了したときにのみ開始されます。

出力

E:\pyrx>python testrx.py
From Task 1: 1
From Task 1: 2
From Task 1: 3
From Task 1: 4
From Task 1: 5
Task 1 complete
From Task 2: 1
From Task 2: 2
From Task 2: 3
From Task 2: 4
Task 2 complete

RxPyは多くのスケジューラーをサポートしています。ここでは、ThreadPoolSchedulerを利用します。ThreadPoolSchedulerは、主に使用可能なCPUスレッドで管理しようとします。

前に見た例では、cpu_countを提供するマルチプロセッシングモジュールを利用します。カウントは、使用可能なスレッドに基づいてタスクを並行して動作させるために管理するThreadPoolSchedulerに与えられます。

ここに、実際の例があります-

import multiprocessing
import random
import time
from threading import current_thread
import rx
from rx.scheduler import ThreadPoolScheduler
from rx import operators as ops
# calculate cpu count, using which will create a ThreadPoolScheduler
thread_count = multiprocessing.cpu_count()
thread_pool_scheduler = ThreadPoolScheduler(thread_count)
print("Cpu count is : {0}".format(thread_count))
def adding_delay(value):
   time.sleep(random.randint(5, 20) * 0.1)
   return value
# Task 1
rx.of(1,2,3,4,5).pipe(
   ops.map(lambda a: adding_delay(a)),
   ops.subscribe_on(thread_pool_scheduler)
).subscribe(
   lambda s: print("From Task 1: {0}".format(s)),
   lambda e: print(e),
   lambda: print("Task 1 complete")
)
# Task 2
rx.range(1, 5).pipe(
   ops.map(lambda a: adding_delay(a)),
   ops.subscribe_on(thread_pool_scheduler)
).subscribe(
   lambda s: print("From Task 2: {0}".format(s)),
   lambda e: print(e),
   lambda: print("Task 2 complete")
)
input("Press any key to exit\n")

上記の例では、2つのタスクがあり、cpu_countは4です。タスクが2で、使用可能なスレッドが4であるため、両方のタスクを並行して開始できます。

出力

E:\pyrx>python testrx.py
Cpu count is : 4
Press any key to exit
From Task 1: 1
From Task 2: 1
From Task 1: 2
From Task 2: 2
From Task 2: 3
From Task 1: 3
From Task 2: 4
Task 2 complete
From Task 1: 4
From Task 1: 5
Task 1 complete

出力が表示されている場合は、両方のタスクが並行して開始されています。

ここで、タスクがCPU数より多い、つまりCPU数が4で、タスクが5であるシナリオを考えてみましょう。この場合、タスクの完了後にスレッドが解放されたかどうかを確認する必要があります。キューで使用可能な新しいタスクに割り当てられます。

この目的のために、任意のスレッドが空いている場合にスケジューラーを監視するobserve_on（）演算子を使用できます。これは、observe_on（）を使用した実際の例です。

例

import multiprocessing
import random
import time
from threading import current_thread
import rx
from rx.scheduler import ThreadPoolScheduler
from rx import operators as ops
# calculate cpu count, using which will create a ThreadPoolScheduler
thread_count = multiprocessing.cpu_count()
thread_pool_scheduler = ThreadPoolScheduler(thread_count)
print("Cpu count is : {0}".format(thread_count))
def adding_delay(value):
   time.sleep(random.randint(5, 20) * 0.1)
   return value
# Task 1
rx.of(1,2,3,4,5).pipe(
   ops.map(lambda a: adding_delay(a)),
   ops.subscribe_on(thread_pool_scheduler)
).subscribe(
   lambda s: print("From Task 1: {0}".format(s)),
   lambda e: print(e),
   lambda: print("Task 1 complete")
)
# Task 2
rx.range(1, 5).pipe(
   ops.map(lambda a: adding_delay(a)),
   ops.subscribe_on(thread_pool_scheduler)
).subscribe(
   lambda s: print("From Task 2: {0}".format(s)),
   lambda e: print(e),
   lambda: print("Task 2 complete")
)
#Task 3
rx.range(1, 5).pipe(
   ops.map(lambda a: adding_delay(a)),
   ops.subscribe_on(thread_pool_scheduler)
).subscribe(
   lambda s: print("From Task 3: {0}".format(s)),
   lambda e: print(e),
   lambda: print("Task 3 complete")
)
#Task 4
rx.range(1, 5).pipe(
   ops.map(lambda a: adding_delay(a)),
   ops.subscribe_on(thread_pool_scheduler)
).subscribe(
   lambda s: print("From Task 4: {0}".format(s)),
   lambda e: print(e),
   lambda: print("Task 4 complete")
)
#Task 5
rx.range(1, 5).pipe(
   ops.map(lambda a: adding_delay(a)),
   ops.observe_on(thread_pool_scheduler)
).subscribe(
   lambda s: print("From Task 5: {0}".format(s)),
   lambda e: print(e),
   lambda: print("Task 5 complete")
)
input("Press any key to exit\n")

出力

E:\pyrx>python testrx.py
Cpu count is : 4
From Task 4: 1
From Task 4: 2
From Task 1: 1
From Task 2: 1
From Task 3: 1
From Task 1: 2
From Task 3: 2
From Task 4: 3
From Task 3: 3
From Task 2: 2
From Task 1: 3
From Task 4: 4
Task 4 complete
From Task 5: 1
From Task 5: 2
From Task 5: 3
From Task 3: 4
Task 3 complete
From Task 2: 3
Press any key to exit
From Task 5: 4
Task 5 complete
From Task 1: 4
From Task 2: 4
Task 2 complete
From Task 1: 5
Task 1 complete

出力が表示された場合、タスク4が完了した瞬間に、スレッドは次のタスク、つまりタスク5に渡され、同じタスクが実行を開始します。

この章では、次のトピックについて詳しく説明します。

• オブザーバブル、オペレーターの動作、およびオブザーバーへのサブスクライブを示す基本的な例。
• 観測量と対象の違い。
• 冷たい観測量と熱い観測量を理解する。

以下に示すのは、observable、operatorsの動作、およびobserverへのサブスクライブを示す基本的な例です。

例

test.py

import requests
import rx
import json
from rx import operators as ops
def filternames(x):
   if (x["name"].startswith("C")):
      return x["name"]
   else :
      return ""
content = requests.get('https://jsonplaceholder.typicode.com/users')
y = json.loads(content.text)
source = rx.from_(y)
case1 = source.pipe(
   ops.filter(lambda c: filternames(c)),
   ops.map(lambda a:a["name"])
)
case1.subscribe(
   on_next = lambda i: print("Got - {0}".format(i)), 8. RxPy — Examples
   on_error = lambda e: print("Error : {0}".format(e)),
   on_completed = lambda: print("Job Done!"),
)

これは非常に単純な例です。ここでは、このURLからユーザーデータを取得しています-

https://jsonplaceholder.typicode.com/users.

データをフィルタリングして「C」で始まる名前を付け、後でマップを使用して名前のみを返します。これが同じの出力です-

E:\pyrx\examples>python test.py
Got - Clementine Bauch
Got - Chelsey Dietrich
Got - Clementina DuBuque
Job Done!

観測量と対象の違い

この例では、オブザーバブルとサブジェクトの違いを確認します。

from rx import of, operators as op
import random
test1 = of(1,2,3,4,5)
sub1 = test1.pipe(
   op.map(lambda a : a+random.random())
)
print("From first subscriber")
subscriber1 = sub1.subscribe(lambda i: print("From sub1 {0}".format(i)))
print("From second subscriber")
subscriber2 = sub1.subscribe(lambda i: print("From sub2 {0}".format(i)))

出力

E:\pyrx>python testrx.py
From first subscriber
From sub1 1.610450821095726
From sub1 2.9567564032037335
From sub1 3.933217537811936
From sub1 4.82444905626622
From sub1 5.929414892567188
From second subscriber
From sub2 1.8573813517529874
From sub2 2.902433239469483
From sub2 3.2289868093016825
From sub2 4.050413890694411
From sub2 5.226515068012821

上記の例では、observableをサブスクライブするたびに、新しい値が提供されます。

件名の例

from rx import of, operators as op
import random
from rx.subject import Subject
subject_test = Subject()
subject_test.subscribe(
   lambda x: print("From sub1 {0}".format(x))
)
subject_test.subscribe(
   lambda x: print("From sub2 {0}".format(x))
)
test1 = of(1,2,3,4,5)
sub1 = test1.pipe(
   op.map(lambda a : a+random.random())
)
subscriber = sub1.subscribe(subject_test)

出力

E:\pyrx>python testrx.py
From sub1 1.1789422863284509
From sub2 1.1789422863284509
From sub1 2.5525627903260153
From sub2 2.5525627903260153
From sub1 3.4191549324778325
From sub2 3.4191549324778325
From sub1 4.644042420199624
From sub2 4.644042420199624
From sub1 5.079896897489065
From sub2 5.079896897489065

サブジェクトを使用している両方のサブスクライバー間で値が共有されていることがわかります。

コールドオブザーバブルとホットオブザーバブルを理解する

オブザーバブルは次のように分類されます

• 冷たい観測量
• ホットオブザーバブル

複数のサブスクライバーがサブスクライブしている場合、オブザーバブルの違いに気付くでしょう。

冷たい観測量

コールドオブザーバブルは、実行されるオブザーバブルであり、サブスクライブされるたびにデータをレンダリングします。サブスクライブされると、observableが実行され、新しい値が与えられます。

次の例は、寒冷観測量の理解を示しています。

from rx import of, operators as op
import random
test1 = of(1,2,3,4,5)
sub1 = test1.pipe(
   op.map(lambda a : a+random.random())
)
print("From first subscriber")
subscriber1 = sub1.subscribe(lambda i: print("From sub1 {0}".format(i)))
print("From second subscriber")
subscriber2 = sub1.subscribe(lambda i: print("From sub2 {0}".format(i)))

出力

E:\pyrx>python testrx.py
From first subscriber
From sub1 1.610450821095726
From sub1 2.9567564032037335
From sub1 3.933217537811936
From sub1 4.82444905626622
From sub1 5.929414892567188
From second subscriber
From sub2 1.8573813517529874
From sub2 2.902433239469483
From sub2 3.2289868093016825
From sub2 4.050413890694411
From sub2 5.226515068012821

上記の例では、observableをサブスクライブするたびに、observableが実行され、値が出力されます。上記の例に示すように、値はサブスクライバーごとに異なる場合もあります。

ホットオブザーバブル

ホットオブザーバブルの場合、準備ができたときに値を出力し、サブスクリプションを常に待機するとは限りません。値が発行されると、すべてのサブスクライバーが同じ値を取得します。

オブザーバブルの準備ができたときに値を発行する場合、またはすべてのサブスクライバーに同じ値を共有する場合は、ホットオブザーバブルを利用できます。

ホットオブザーバブルの例は、サブジェクトおよび接続可能なオペレーターです。

from rx import of, operators as op
import random
from rx.subject import Subject
subject_test = Subject()
subject_test.subscribe(
   lambda x: print("From sub1 {0}".format(x))
)
subject_test.subscribe(
   lambda x: print("From sub2 {0}".format(x))
)
test1 = of(1,2,3,4,5)
sub1 = test1.pipe(
   op.map(lambda a : a+random.random())
)
subscriber = sub1.subscribe(subject_test)

出力

E:\pyrx>python testrx.py
From sub1 1.1789422863284509
From sub2 1.1789422863284509
From sub1 2.5525627903260153
From sub2 2.5525627903260153
From sub1 3.4191549324778325
From sub2 3.4191549324778325
From sub1 4.644042420199624
From sub2 4.644042420199624
From sub1 5.079896897489065
From sub2 5.079896897489065

表示されている場合、同じ値がサブスクライバー間で共有されています。公開（）接続可能な監視可能な演算子を使用して同じことを実現できます。