Observableの解説
非同期処理やリアクティブプログラミングの文脈で登場する "Observable" は、Rx（Reactive Extensions）ライブラリの中核的な概念です。この記事では、PythonのRxPYをベースに、Observableの基本からHot/Coldの違い、イテレータとの比較、処理方法までを徹底的に解説します。
Observableとは？
Observable（オブザーバブル）は、時間とともに発生する値やイベントのストリームを表現するオブジェクトです。データが「発行」され、それを購読（subscribe）することで、非同期に受け取ることができます。
特徴：
- データの流れ（ストリーム）を表現
-
subscribe()によってデータを受信 - 時間軸に沿って複数の値を送る
-
map,filter,combine_latestなどのオペレータで処理を組み立てられる
サンプルコード（基本）
import rx
from rx import operators as ops
# 1から5のストリーム
source = rx.from_iterable(range(1, 6))
# 2倍してフィルター
source.pipe(
ops.map(lambda x: x * 2),
ops.filter(lambda x: x > 5)
).subscribe(lambda x: print(f"Received: {x}"))
出力:
Received: 6
Received: 8
Received: 10
イテレータ、Hot Observable、Cold Observable の比較
イテレータ（同期・Pull型）
def number_generator():
for i in range(3):
yield i
for val in number_generator():
print(f"A: {val}")
出力:
A: 0
A: 1
A: 2
Cold Observable（非同期・Push型、購読ごとに再実行）
import rx
from rx import operators as ops
import time
source = rx.from_iterable([0, 1, 2])
source.pipe(
ops.do_action(lambda _: time.sleep(1))
).subscribe(lambda x: print(f"A: {x}"))
time.sleep(2)
source.pipe(
ops.do_action(lambda _: time.sleep(1))
).subscribe(lambda x: print(f"B: {x}"))
出力:
A: 0
A: 1
A: 2
B: 0
B: 1
B: 2
Hot Observable（非同期・Push型、共有ストリーム）
import rx
from rx.subject import Subject
from threading import Thread
import time
subject = Subject()
def emit_values():
for i in range(3):
time.sleep(1)
subject.on_next(i)
Thread(target=emit_values).start()
subject.subscribe(lambda x: print(f"A: {x}"))
time.sleep(2)
subject.subscribe(lambda x: print(f"B: {x}"))
time.sleep(2)
出力例:
A: 0
A: 1
A: 2
B: 2
ユースケースの違い：Cold vs Hot Observable
Cold Observable（遅延評価、再実行されるストリーム）
Cold Observable は以下のような「購読者ごとに毎回新しい処理をしたい」場面で有効です。
ユースケース例：
- HTTPリクエスト（購読するたびにリクエストが実行される）
- ファイルの読み込み（購読するごとに別々の読み取り）
- データベースクエリの発行
特徴：
- 各購読者がデータ生成の開始点から観測できる
- 再現性のある処理に向く
Hot Observable（イベント共有、現在値だけを配信）
Hot Observable は、継続的に発生しているリアルタイムデータやイベントを「共有」したい場合に有効です。
ユースケース例：
- センサーの値（温度、加速度など）
- WebSocket や Kafka のようなストリーミングAPI
- UIイベント（マウスの動き、クリックなど）
- ログ監視ストリーム（複数モジュールでリアルタイム解析）
特徴：
- すでに開始しているデータフローに「途中から参加」する
- 複数の購読者でストリームを共有
Observableの処理方法
Observableの値は、基本的に subscribe() を使って受け取ります。
方法1: subscribe()
rx.from_iterable([1, 2, 3]).subscribe(lambda x: print(x))
方法2: .pipe() と演算子で加工
rx.from_iterable([1, 2, 3]).pipe(
ops.map(lambda x: x * 10),
ops.filter(lambda x: x > 10)
).subscribe(lambda x: print(x))
方法3: .to_list() + .subscribe() または .run()（RxPY v3）
rx.from_iterable([1, 2, 3]).pipe(
ops.to_list()
).subscribe(lambda x: print(f"List: {x}"))
または：
result = rx.from_iterable([1, 2, 3]).pipe(ops.to_list()).run()
print(result) # [1, 2, 3]
知っておくと良いこと
-
遅延評価（Lazy Evaluation）：Observableは
subscribe()されるまで実行されません - Hot Observable の扱いには注意：途中から購読すると過去の値は見えません
-
Schedulers を使えば非同期・スレッド処理も制御できます（例：
ThreadPoolScheduler） - Rxは
async/awaitよりも宣言的で、複雑なストリーム処理やUIイベント連携に向いています
まとめ
| 種類 | タイプ | データ発行のタイミング | 複数購読者対応 |
|---|---|---|---|
| イテレータ | Pull | 要求時 | × |
| Cold Observable | Push | 購読時に開始 | △（再実行） |
| Hot Observable | Push | ストリーム開始時 | ○（途中参加） |
Observableは非同期処理やリアクティブアーキテクチャにおいて非常に強力なツールです。特に複数の非同期イベントをシンプルに結合・加工したい場面では、イテレータやコールバックよりもはるかに表現力があります。
より高度な例（combine_latest, debounce, switch_map）なども活用することで、RxPYの真価が見えてきます。