Skip to content

MS_ReactiveExtensions

Jump to bottom
nishi_74322014 edited this page Aug 21, 2026 · 1 revision

Reactive Extensions（Rx）

概要

  • Observer パターンを実装するフレームワークだが、

  • 特に、LINQ to Events もしくは LINQ to Asynchronous と表現できる。

  • これは、要するに、

    • イベントや非同期処理を LINQ っぽく扱える。
    • 非同期 / イベント / 時間に関する処理を LINQ 的に簡潔かつ宣言的に記述できる。

と言うトコロを意味する。

移行メモ（表記）: 原文の「LINQ to Asynchronus」は
Asynchronous の綴りの誤りと判断し修正した。

補足（IEnumerable と IObservable は双対）: Rx を理解する鍵は、
IObservable<T> が IEnumerable<T> の裏返しである、という点にある。

IEnumerable<T>（Pull） IObservable<T>（Push）
主導権 受け手が MoveNext() で取りに行く 送り手が OnNext() で押し込む
終了 MoveNext() が false OnCompleted()
異常 例外が throw される OnError()
演算子 Where / Select / … 同じ名前の演算子が使える

「値の並び」であることは同じなので、
LINQ の演算子がそのまま通用する。
これが「イベントを LINQ で扱える」という発想の正体である。

// 「クリックが 300ms 以内に 2 回」＝ ダブルクリック、を宣言的に書く
clicks.Buffer(clicks.Throttle(TimeSpan.FromMilliseconds(300)))
      .Where(x => x.Count == 2)
      .Subscribe(_ => Console.WriteLine("double click"));

時間を第一級に扱える点が、Rx が今も価値を持つ理由である。

経緯

  • Silverlight Toolkit に "System.Reactive.dll" が同梱される。

  • Reactive Framework ---> Reactive Extensions と名称変更。

  • DevLabs（昔の MS コミュニティサイト）でプロジェクト公開
    ソースコードは（恐らく）CodePlex で公開されていた。

  • 何度も API の消滅・追加を繰り返す。

  • JavaScript 版が登場（RxJS）

  • .NET Framework 4 SP1 に標準搭載

  • さまざまな開発言語に移植され、幅広く使われている。

    • JavaScript 用の RxJS
    • Java / Android 用の RxJava
    • Swift 用の RxSwift
    • Unity 用の UniRx

補足（その後／最新化）: 原文以降の動きを補う。

時期 内容
2016 頃 ReactiveX として言語横断の仕様・サイトに集約
Rx.NET が .NET Foundation 配下でメンテナンスされる
2019 IAsyncEnumerable<T> が C# 8.0 / .NET Core 3.0 に標準搭載
現在 System.Reactive パッケージとして提供（Rx.NET リポジトリ）
Unity UniRx → R3 / UniTask に世代交代が進む

重要な変化は IAsyncEnumerable<T> の登場である。
「非同期に届く値の並びを扱う」という Rx の用途の一部が、
言語機能（await foreach）で素直に書けるようになった。

// 昔は Rx が必要だった処理が、標準構文で書ける
await foreach (var item in GetItemsAsync())
    Console.WriteLine(item);

現在の使い分けの目安は次の通り。

やりたいこと 適した手段
1 回の非同期処理 async/await
非同期に届く有限の並び（Pull） IAsyncEnumerable<T>
時間・イベントの合成（Push） Rx
UI のイベント処理・デバウンス Rx

Rx は「万能の非同期基盤」から
**「イベントと時間の合成に特化した道具」**へと位置づけが定まった、
と理解するのが実態に近い。

ユースケース

Observer パターンの実装

LINQ（フィルタ、変換、集計、合成）

通知に対して、LINQ（フィルタ、変換、集計、合成）を適用できる。

イベントの LINQ 化

イベントに Observer パターンを適用して、
LINQ（フィルタ、変換、集計、合成）を適用できる。

非同期の LINQ 化

  • イベントに Observer パターンを適用して、
    LINQ（フィルタ、変換、集計、合成）を適用できる。

  • 具体的には、Observer を合成してネストを処理が可能。

詳細

Observer パターンで構成される。

  • 何らかのProvider（IObservable<T> など）で監視を行い、
  • Providerから状態の Push（＝通知、発行、イベント）を、
  • Subscriber（IObserver<T> など）が受け取る。

Subscriber

IObserver<T> を実装したクラスを定義する。

Provider

  • IObservable<T> を実装したクラスを定義する。
  • クラスはラムダ式を使用したメソッドチェーンで実装できる。

開始

  • Create などの生成メソッドで生成、
# メソッド 概要
1 Create 監視内容を定義する。
2 Return 渡した値を単純に通知する。
3 Range 指定した範囲の値を通知する。
4 Repeat 第 1 引数で渡したデータを、
第 2 引数で指定した回数繰り返して通知する。
5 Generate for 文的に以下を定義する。
・初期値
・継続判定デリゲート
・インクリメント・デリゲート
・通知する値を生成するデリゲート
6 Case 複数用意された IObservable の中から、どれか１つを選択。
・IObservable 配列
・IObservable 選択デリゲート
7 Throw 例外を通知する。
8 Start 渡した値を非同期で通知する。
9 Defer Provider のファクトリを実装。

移行メモ（誤字）: Range の説明「正指定した範囲の値」は
指定した範囲の値の誤字と判断し修正した。

  • Create で生成する場合、以下のメソッドを使用して監視内容を定義する。
# メソッド 概要
1 OnNext 新しい値が発生したことを通知する。
2 OnError エラーが発生し、異常終了したことを通知する。
3 OnCompleted 正常に終了したことを通知する。

補足（Rx の文法規則）: この 3 つには厳密な順序の約束がある。

OnNext* (OnError | OnCompleted)?

つまり、

  • OnNext は 0 回以上流れる
  • OnError または OnCompleted は どちらか一方が、最後に 1 回だけ
  • 終了通知の後には、何も流れてはならない

自分で IObservable<T> を実装する場合、この規則を守る責任は
実装側にある。Observable.Create を使えば大部分は担保される。

  • 必要に応じて、LINQ メソッドを使用し通知内容を操作できる。

  • Subscribe で、上記の其々のメソッドからの
    Push（＝通知、発行、イベント）を受け取るSubscriberを設定する。

終了

  • Provider が OnCompleted を呼び出し、正常終了。
  • Provider が OnError を呼び出し、異常終了。
  • Subscriber側で Dispose する。

補足（購読解除の忘れがリークになる）: Subscribe は
IDisposable を返す。これを捨てると購読が残り続け、
Provider が Subscriber を参照し続けるためメモリ リークになる。

// ① using / Dispose で明示的に解除
using var sub = observable.Subscribe(...);

// ② CompositeDisposable にまとめて、画面破棄時に一括解除
_disposables.Add(observable.Subscribe(...));

// ③ 寿命を別のストリームに委ねる
observable.TakeUntil(_closed).Subscribe(...);

OnError で購読が終了する点も重要で、
エラー後も流し続けたい場合は Retry / Catch を挟む必要がある。

分類

大きく分けて Hot / Cold に分類される。

  • Cold

    • Subscriberとは１：１の関係
    • Subscriberが居なければ動作しない。
    • 監視が終わった後は OnCompleted で（自分で）終了する。
    • Observable.Interval など例外もある。

  • Hot

    • Subscriberとは１：ｎの関係
    • Subscriberがいなくても動作し続ける。
    • 監視が終わったかどうかSubscriberが判断して Dispose。

補足（Hot / Cold は Rx 最大のハマりどころ）: 原文が
後述のポイントで「概念を掴む」と強調している通り、
ここが最も事故が起きる箇所である。

Cold Hot
値の生成 購読のたびに最初から生成 1 本の流れを共有
例 Observable.Range、HTTP リクエスト イベント、Subject、株価
購読前の値 失われない（購読時に始まる） 購読前の値は受け取れない

典型的な事故:

var src = Observable.Create<int>(o => { Console.WriteLine("実行"); ... });
src.Subscribe(...);   // 「実行」
src.Subscribe(...);   // 「実行」← 2 回動いてしまう

HTTP リクエストなど副作用のある Cold を複数箇所で購読すると、
リクエストが購読数だけ飛ぶ。

対処は Cold を Hot に変換すること。

var shared = src.Publish().RefCount();  // 購読者間で 1 本を共有

Subject (Subscriber/Provider)

  • IObserver<T> と IObservable<T> を実装したクラスを定義できる。
  • Subject を使用すれば、Subscriber をラムダ式で定義できる。

補足（Subject の種類と、使いすぎへの注意）:

種類 挙動
Subject<T> 購読後に流れた値のみ受け取る
BehaviorSubject<T> 最新の 1 件を購読時に即座に受け取る（状態の表現に適する）
ReplaySubject<T> 過去 n 件を購読時に再生
AsyncSubject<T> 完了時に最後の 1 件だけ流す（Task に近い）

ただし、Subject を安易に使うと Rx の宣言的な利点が失われる
（手続き的に OnNext を呼ぶコードが散らばる）。
可能なら Observable.FromEvent や演算子で組み立て、
Subject は外部との境界に限定するのが定石である。

その他

合成

Provider の合成

  • Merge
  • SelectMany
  • Switch

値の合成

  • Concat
  • Zip
  • Amb
  • CombineLatest

補足（実務で使う頻度が高いもの）: 数百ある演算子のうち、
実際によく使うのは限られる。

演算子 用途
Throttle（RxJS の debounce） 入力が止まってから実行（インクリメンタル検索）
Sample 一定間隔で最新値を間引く
Switch 新しい要求が来たら、前の処理を捨てる（検索の競合防止）
CombineLatest 複数の最新値を組み合わせる（フォームの妥当性判定）
Retry / Catch エラー時の再試行
DistinctUntilChanged 同じ値の連続を無視

Throttle + Switch の組み合わせは、
「入力のたびに検索 API を叩き、古い結果が後から返って上書きする」
という古典的なバグを、宣言的に解決する。

textChanged
  .Throttle(TimeSpan.FromMilliseconds(300))   // 入力が落ち着いてから
  .DistinctUntilChanged()                     // 同じ語なら投げない
  .Select(q => SearchAsync(q).ToObservable())
  .Switch()                                   // 古い検索は破棄
  .Subscribe(UpdateUI);

Merge と Concat の違いにも注意。
Merge は並行、Concat は前が終わってから次である。

変換

イベント

補足（ドラッグが Rx の代表例である理由）: ドラッグは
「MouseDown してから MouseUp するまでの MouseMove」であり、
イベントの時間的な組み合わせそのものである。

var drag = from down in mouseDown
           from move in mouseMove.TakeUntil(mouseUp)
           select move.Location;

手続き的に書くと「ドラッグ中フラグ」と開始座標の状態管理が要るが、
Rx では状態変数を持たずに表現できる。

非同期

補足（現在は async/await が担う領域／最新化）: この節が扱う
APM（Begin～/End～）のネスト解消は、
Rx が登場した当時（.NET 4 以前）における最大の動機の一つだった。

しかし C# 5.0 の async/await により、
「非同期のネスト」は言語レベルで解決された。

// かつて Rx / SelectMany で解いていた問題
var a = await GetAAsync();
var b = await GetBAsync(a);

現在の Rx と Task の相互変換は次の通り。

変換 メソッド
Task → IObservable .ToObservable()
IObservable → Task await observable（最後の値）、.ToTask()
IObservable → IAsyncEnumerable .ToAsyncEnumerable()

なお FromAsyncPattern は
APM 自体が非推奨（Begin～/End～ は現在ほぼ使われない）のため、
新規コードでは Observable.FromAsync を使う。

ポイント

Hot と Cold の概念を掴む

Scheduler が実行スレッドを決定する

補足（Scheduler の要点）: Rx ではどのスレッドで動くかを
演算子ではなく Scheduler が決める。

メソッド 何を変えるか
ObserveOn 通知を受け取る（＝下流が動く）スレッド
SubscribeOn **購読処理（＝上流の開始）**が動くスレッド

UI アプリでは、
**「重い処理はバックグラウンド、UI 更新は UI スレッド」**を
この 2 つで宣言的に指定できる。

source
  .SubscribeOn(TaskPoolScheduler.Default)      // 取得はプールで
  .Select(Heavy)
  .ObserveOn(DispatcherScheduler.Current)      // 表示は UI スレッドで
  .Subscribe(UpdateUI);

既定では、Rx はスレッドを勝手に切り替えない
（OnNext を呼んだスレッドがそのまま下流を実行する）点に注意。

Subject でテストする

補足（テストの本命は TestScheduler）: 原文の言う
「Subject でテストする」（任意のタイミングで値を流し込む）に加え、
Rx には TestScheduler（Microsoft.Reactive.Testing）がある。

仮想時間を進められるため、
Throttle(5分) のような処理を
実際に 5 分待たずにテストできる。

var s = new TestScheduler();
var result = s.Start(() => source.Throttle(TimeSpan.FromMinutes(5), s));
// 5 分を「即座に」進めて検証できる

時間に依存する処理を確実にテストできるという点は、
手続き的な実装に対する Rx の明確な優位性である。

事件

2020/03/09 ちょっとした事件が起きたらしい。

FF

ファクト・ファインディング

感想

  • イイね。

  • SI っぽくなって来た。

  • dis っては無い。

「そう言う事も考える必要が出てきた。」

...と言う事で。

補足（この「事件」の論点）: 参照先は、
「Rx は強力だが、チーム全体が使いこなせるとは限らない」
という、技術選定における習熟度格差の話題である。

Rx は学習曲線が急で、

  • Hot / Cold を理解していないと、副作用が多重に走る
  • 購読解除を忘れるとリークする
  • 演算子が数百あり、レビューで妥当性を判断しづらい
  • スタック トレースが追いにくく、デバッグが難しい

といった性質を持つ。原文の「SI っぽくなって来た」という感想は、
個人の技量に依存する技術を、多人数の開発に持ち込む難しさを
指したものと読める。

現在の妥当な判断としては、

  • 単発の非同期は async/await、
  • 有限の非同期列は IAsyncEnumerable<T>、
  • 本当に「時間とイベントの合成」が要る箇所に限って Rx、

と切り分けるのが、この論点への実務的な回答になる。

参考

移行メモ（リポジトリ移管）: 原文の
Reactive-Extensions/Rx.NET は、現在 dotnet/reactive に移管
（.NET Foundation 配下）されているため、URL を更新した。

＠IT

Build Insider

Qiita

neue cc

xin9le.net

かずきのBlog@hatena

present

Reactive Extensions 入門

Muhammad Rehan Saeed

Reactive Extensions (Rx)

Tags: 移行, .NET開発

このWikiは「Open棟梁Project」,「OSSコンソーシアム 開発基盤部会」によって運営されています。

NetDevInfraWiki

マイクロソフト系技術情報 Wiki
Open 棟梁 Wiki

（未着手）

開発基盤部会 Wiki

移行管理: DONE ／ TODO

Clone this wiki locally