#0128（2025/05/10）RectiveXとは (RxPYの用例あり)

## はじめに

近年、複雑な非同期・イベント駆動システムの要件が増大する中で、\*\*ReactiveX（Reactive Extensions）\*\*は関数型リアクティブプログラミング（FRP）を実現する強力なライブラリ群として注目を集めています。本稿では上級エンジニアの視点で、ReactiveXのコア概念から実践的ユースケース、マルチランゲージ対応状況、そしてPythonにおけるシンプル／複雑サンプルを通じて、async/awaitとの違いや使い分け指針までを解説します。

---

### 1. ReactiveXの概念

* **Observable**: イベントやデータのストリームを表すプッシュ型シーケンス。Cold／Hot／Connectableなど多彩なタイプが存在。
* **Observer**: Observableから発行されるデータ（Next／Error／Completed）を受信し、処理を行うエンドポイント。
* **Operator**: Streamに対する変換・結合・フィルタ・バッファ・バックプレッシャー制御などの演算を提供。
* **Subscription**: ObservableとObserverをつなぐ購読契約。キャンセルや自動解除（AutoDispose）も重要。

> **ポイント**: 宣言的にストリーム処理を定義し、エラーハンドリングや並行性制御、背圧（backpressure）を統一的に扱えるのがReactiveXの強みです。

---

### 2. ReactiveXのユースケース

1. **UI イベント処理**

   * ボタン連打のデバウンス／スロットリング、複数ウィジェットのイベント合成
2. **リアルタイムデータ分析**

   * センサーデータ／IoTメトリクスのウィンドウ集計や異常検知パイプライン
3. **分散システム間メッセージング**

   * Kafka/RabbitMQと連携したストリーム処理、リアクティブマイクロサービス設計（Reactive Streams Spec準拠）
4. **ネットワーク通信**

   * WebSocket／gRPCストリーミング、バックプレッシャー対応の双方向通信

---

### 3. ReactiveXの対応言語

* **RxJava** / RxKotlin
* **RxJS** （JavaScript／TypeScript）
* **RxPY**（Python）
* **Rx.NET**（C#）
* **RxSwift** / RxCocoa（Swift）
* **RxGo**（Go）
* **RxRuby**, **RxPHP**

> **補足**: 多くの実装が[Reactive Streams仕様](https://github.com/reactive-streams/reactive-streams-jvm)に準拠し、言語間で似たAPIと動作保証を提供します。

---

### 4. Pythonサンプルコード（単純）

```python
from rx import of
from rx.operators import map, filter

# 単純な数列発行 → フィルタ → 変換 → 購読
of(1, 2, 3, 4, 5).pipe(
    filter(lambda x: x % 2 == 1),
    map(lambda x: x * 100)
).subscribe(
    on_next=lambda v: print(f"Result: {v}"),
    on_error=lambda e: print(f"Error: {e}"),
    on_completed=lambda: print("Completed")
)
```

#### シーケンス解説

下記のフローでデータが処理され、出力されます。

```
入力ストリーム: 1 → 2 → 3 → 4 → 5
               │    │    │    │    │
               ▼    ▼    ▼    ▼    ▼
filter (奇数のみ): 1    ✕    3    ✕    5
               │         │         │
               ▼         ▼         ▼
map (×100):   100       300       500
               │         │         │
               ▼         ▼         ▼
subscribe → print → Result: 100, Result: 300, Result: 500 → Completed
```

---

### 5. Pythonサンプルコード（複雑）

```python
import rx
from rx import operators as ops
import time

# Sensor A: 1秒ごとに 1, 2, 3... を発行
sensor_a = rx.interval(1.0).pipe(
    ops.map(lambda i: i + 1)
)

# Sensor B: 1.5秒ごとに 10, 20, 30... を発行
sensor_b = rx.interval(1.5).pipe(
    ops.map(lambda i: (i + 1) * 10)
)

# A発行時にBの最新値と合成し、5件取得
combined = sensor_a.pipe(
    ops.with_latest_from(sensor_b),
    ops.map(lambda ab: ab[0] + ab[1]),
    ops.take(5)
)

combined.subscribe(
    on_next=lambda v: print(f"Combined: {v}"),
    on_completed=lambda: print("Done")
)

# 約10秒待機してメインスレッドを維持
time.sleep(10)
```

上記で扱う `with_latest_from` や `take` は、**ストリーム合成と完結制御**を宣言的に実装できる好例です。

```text
時間:    | 0s |0.5s| 1s |1.5s| 2s |2.5s| 3s |3.5s| 4s |4.5s| 5s |5.5s| 6s |
---------------------------------------------------------------------------
A出力:   |     |     |  1 |     |  2 |     |  3 |     |  4 |     |  5 |     |    |
B出力:   |     |     |     | 10 |     |     | 20 |     | 30 |     |     |     |    |
---------------------------------------------------------------------------
合成出力:|     |     |11=A1+B10|     |12=A2+B10|     |23=A3+B20|     |34=A4+B30|     |35=A5+B30|     |    |


[出力] A:1 + B:10 = 11
[出力] A:2 + B:10 = 12
[出力] A:3 + B:20 = 23
[出力] A:4 + B:30 = 34
[出力] A:5 + B:30 = 35
完了
```

---

### 6. async/awaitとReactiveXの違い

| 比較軸       | async/await                  | ReactiveX                  |
| --------- | ---------------------------- | -------------------------- |
| 記述スタイル    | 命令的（逐次・明示的await）             | 宣言的（ストリーム操作をチェイン）          |
| 対象        | **単発**の非同期I/O                | **継続的**なイベント／データストリーム      |
| 並行制御      | `gather`, `create_task`で並列実行 | 演算子内でバックプレッシャー含む詳細制御が可能    |
| エラーハンドリング | try/exceptブロック               | ストリームのon\_errorハンドラ        |
| 学習コスト     | 低\~中                         | 中\~高（多彩なOperatorを理解する必要あり） |

---

### 7. ユースケース別の選び方

* **リアルタイム／継続データ** → **ReactiveX**

  * 高頻度に流れるイベントを合成・フィルタ・バッファ
  * UIフレームワークやデータパイプラインに最適
  * バックプレッシャーやキャンセル制御が重要な場面

* **単発I/O／シンプルな並行実行** → **async/await**

  * APIリクエストやファイルアクセスなどの一回きり処理
  * コード可読性重視で簡潔に書きたいとき

* **混合シナリオ** → 両者の併用

  * RxPYストリーム内で `async def` 処理をawaitしたい場合や、
  * `from_future` / `start` 演算子を使ってCoroutineをObservableに取り込む。

---

## おわりに

本稿では、ReactiveXのコア概念と高度ユースケース、マルチランゲージ対応状況、Pythonにおける単純＆複雑サンプルを示し、async/awaitとの違いと選び方をまとめました。上級エンジニアの皆様には、**システム要件に応じて両者を適材適所に使い分ける設計力**を磨く一助となれば幸いです。