はじめに
- ReactiveX の概念を理解するためにまとめた資料です
- 社内クローズドな勉強会で話した内容を Qiita 向けに修正しています
- 前提としてデザインパターンの知識がちょっと必要です
- サンプルコードはほぼ Swift + RxSwift です
ReactiveXとは
ReactiveXとは
ReactiveX (以降 Rx) は Observable シーケンスを使った非同期でイベントベースのプログラムを構成するためのライブラリ。
デザインパターンの Observer パターン を下記の２点で拡張したもの。
- データやイベントのシーケンス をサポートする
- 宣言的にシーケンスを構成できるようにするための オペレータ を追加する
また、低レベルなスレッディング、同期、スレッドセーフ、並行データ構造、ノンブロッキングI/Oのような懸念事項を抽象化してくれる。
(引用元: http://reactivex.io/intro.html)
Rx は Observer パターンの拡張
- データやイベントのシーケンスをサポートする
- 宣言的にシーケンスを構成できるようにするためのオペレータを追加する
【復習】Observerパターン
Observerパターン
オブジェクトのイベントを他のオブジェクトへ通知するためのパターン。GoF による デザインパターン の一つ。
-
Observerable
通知する側のオブジェクト＝監視可能なオブジェクト。Observer に監視される。 -
Observer 通知される側のオブジェクト＝監視するオブジェクト。Observerable を 監視する。
Observerパターンの例 (Swift)
observer-pattern
import Foundation
// MARK: - Interfaces
protocol IObserver: class {
associatedtype Value
func notify(_ value: Value)
}
protocol IObservable: class {
associatedtype Value
associatedtype Observer: IObserver where Observer.Value == Value
func registerObserver(_ observer: Observer)
func unregisterObserver(_ observer: Observer)
func notifyObservers(_ value: Value)
}
// MARK: - Implementations
class IntObserver: IObserver {
typealias Value = Int
let name: String
init(name: String) {
self.name = name
}
func notify(_ value: Value) {
print("\(name) -> \(value)")
}
}
class IntObservable: IObservable {
typealias Value = Int
typealias Observer = IntObserver
func registerObserver(_ observer: Observer) {
observers.append(observer)
}
func unregisterObserver(_ observer: Observer) {
observers.removeAll(where: { $0 === observer })
}
func notifyObservers(_ value: Observer.Value) {
observers.forEach { $0.notify(value) }
}
private var observers: [Observer] = []
}
// MARK: - Examples
let observable = IntObservable()
// register observers
observable.registerObserver(IntObserver(name: "A"))
observable.registerObserver(IntObserver(name: "B"))
//notify
observable.notifyObservers(10)
observable.notifyObservers(20)
result
A -> 10
B -> 10
A -> 20
B -> 20
Observable ≒ データやイベントのシーケンス
observable.notifyObservers(10)
observable.notifyObservers(20)
- Observer には Observable から 10, 20 というデータが順番に非同期に通知される
- また、ここで通知されるものはデータに限らず、例えばボタンクリックのようなイベントかもしれない
- つまり、Observable は データやイベントのシーケンス とみなせる
データやイベントの非同期なシーケンスに対して順番に反応する Reactive Programing
【復習】Iterator パターン
Iterator パターン
- 集約したオブジェクトを列挙する手段を提供するデザインパターン。これも GoF の一つ。
- モダンなプログラミング言語のコレクションコンテナでは大抵実装されている。
Iterator パターンの例 (Swiftのコレクション)
Swift の配列などのコレクションは Iterator パターンを定義した Sequence というプロトコル（≒ Interface）に準拠している。
swift-sequence
protocol IteratorProtocol {
associatedtype Element
mutating func next() -> Element?
}
protocol Sequence {
associatedtype Element
associatedtype Iterator: IteratorProtocol
func makeIterator() -> Iterator
}
Iterator パターンの例 (Swiftのコレクション)
-
Sequenceプロトコルに準拠していると Iterator によりコレクションの要素を次のように列挙することができる。 -
next()がnil(null) を返すと列挙は完了する。
array-iterator
let array = [1, 2, 3, 4, 5]
var iterator = array.makeIterator()
while let next = iterator.next() {
print(next)
}
- （おまけ）上記のコードは Swift の
for in構文でシンプルに記述することができる。
for-in
let array = [1, 2, 3, 4, 5]
for element in array {
print(element)
}
Observer/Iterator パターンのまとめ
Observer/Iterator パターンのまとめ
- Observer パターンの Observable はデータやイベントの シーケンス とみなせる
- Iterator パターンはシーケンスの要素を 次へ (Next) により順番に得ることができる
- また、次の要素がない場合はシーケンスの 完了 (Complete) となる
コレクション と Observable (Rx)
コレクション と Observable (Rx) 👉 どちらもシーケンス
コレクションは Iterator パターンにより要素を順番に処理できる。
let array = [1, 2, 3, 4, 5]
var iterator = array.makeIterator()
while let next = iterator.next() {
print(next)
}
これを Rx の Observable の世界にするとこうなる。
let observable = Observable.of(1, 2, 3, 4, 5)
observable.subscribe(onNext: { next in
print(next)
}
👉 似てるよね！
コレクション と Observable (Rx) の違い
- コレクションはシーケンスの全ての要素を同時に処理する
- Rx の Observable はシーケンスの要素が非同期に放出され、そのタイミングで処理する
(引用元: https://www.atmarkit.co.jp/fdotnet/introrx/introrx_01/introrx_01_01.html)
👉 Rx の Observable は時間軸に乗る非同期なストリーム（小川）
コレクション と Observable (Rx) の違い (ソースコードで比べる)
コレクションでは next() で能動的に次の要素を取りに行く。
let array = [1, 2, 3, 4, 5]
var iterator = array.makeIterator()
while let next = iterator.next() {
print(next)
}
Rx の Observable では、それを 購読 (Subscribe) し、次の要素が放出されるのを onNext で待つ。
let observable = Observable.of(1, 2, 3, 4, 5)
observable.subscribe(onNext: { next in
print(next)
}
Rx のマーブルダイアグラム
Rx の Observable はマーブルダイアグラムにより図示される。
(引用元: http://reactivex.io/documentation/observable.html)
Rx とは Observer パターンの拡張
- データやイベントのシーケンスをサポートする
- 宣言的にシーケンスを構成できるようにするためのオペレータを追加する
オペレータ
コレクションのオペレータ
モダンなプログラミング言語ではコレクションに対する様々なオペレータとサポートする。
-
射影 :
map,flatMap, ... -
抽出 :
filter,take, ... -
合成 :
concat,zip, ... -
集計 :
reduce,distinct, ... -
並び替え :
sort,reverse, ...
(参考: https://qiita.com/amay077/items/9d2941283c4a5f61f302)
コレクションのオペレータの例 (Swift)
整数のコレクションに次のオペレーション行う。
- 全ての要素に3を足す
- そこから偶数のみを抽出する
let array = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
array.map { $0 + 3 } // [4, 5, 6, 7, 8, 9, 10, 11, 12, 13]
.filter { $0.isMultiple(of: 2) } // [4, 6, 8, 10, 12]
.subscribe(onNext: { print($0) }
Observable のオペレータ
ほぼ同じ！Observable もシーケンス。Rx からは様々なオペレータが提供されている。
let observable = Observable.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
observable.map { $0 + 3 } // [4, 5, 6, 7, 8, 9, 10, 11, 12, 13]
.filter { $0.isMultiple(of: 2) } // [4, 6, 8, 10, 12]
.subscribe(onNext: { print($0) }
Observable の具体的な例 (Spread Sheet)
みんな大好きスプレッドシート。
let a1 = Observable.of(1)
let b1 = Observable.of(2)
let c1 = Observable.combineLatest(a1, b1).map { $0 + $1 }
Observable の具体的な例 (UI)
検索バーにキーワードを入力するたびに検索結果が表示される（インクリメンタルサーチ）
// テキストフィールドからの入力が変化する度に、その入力内容が放出される
textField.rx.text.asObservable()
// 連続入力でAPIを叩きすぎないようにするために200msec以内の連続入力は無視する
.debounce(.milliseconds(200), scheduler: MainScheduler.instance)
// 入力内容で検索APIを実行
.flatMap { keyword in
API.search(by: keyword).asObservable()
}
// その結果を購読し、処理する（結果一覧を表示する、など）
.subscribe(onNext: { result in
print(result)
})
Rxの起源
- Microsoft Research の実験的なプロジェクトとして始まる
- 2009年に .NET Framework 向けの Rx.NET を提供開始
- 2011年に正式な製品に昇格し、以降各プラットフォームに移植される
- Rx 全体の開発を現在リードしているのは Netflix (MSでRxやってた人が転職した先)
- Rx のオペレータは 2007年にリリースされた C# 3.0 の新機能 LINQ の非同期/イベント版といえる
(参考: https://www.atmarkit.co.jp/fdotnet/introrx/introrx_01/introrx_01_01.html)
(参考: https://www.buildinsider.net/column/kawai-yoshifumi/004)
Rxの対応プラットフォーム
触れないプラットフォームなど、（ほぼ）ない！ Let's play with Rx!
| 言語/PF | Rx |
|---|---|
| Java | RxJava |
| Java (Android) | RxAndroid |
| JavaScript | RxJS |
| C# | Rx.NET |
| C#(Unity) | UniRx |
| Scala | RxScala |
| Clojure | RxClojure |
| C++ | RxCpp |
| Lua | RxLua |
| Ruby | Rx.rb |
| Python | RxPY |
| Go | RxGo |
| Groovy | RxGroovy |
| JRuby | RxJRuby |
| Kotlin | RxKotlin |
| Swift | RxSwift |
| Swift (iOS/macOS) | RxCocoa |
| PHP | RxPHP |
| Elixir | reaxive |
| Dart | RxDart |
おまけ
脱Rxの兆し
iOS/Android 界隈では脱 Rx の兆しが...
-
iOS : Apple が Swift 用 RP フレームワーク Combine を提供開始
- iOS13以降で利用可能。Swift UI のためのフレームワークでもある。
-
Android : Kotlin の Coroutine 機能に RP 的機能が増えてきている
- Channel, Flow, ...
Rx is not FRP
ReactiveX はたまに Functional Reactive Programming (FRP) と称されるが、それは 誤り 。ReactiveX は Functional で Reactive かもしれないが、FRP とは別のもの。
| Difference | |
|---|---|
| FRP | operates on values that change continuously over time 時間とともに継続的に変化する値を操作する |
| ReactiveX | operates on discrete values that are emitted over time 時間の経過とともに放出される離散的な値を操作する |
なるほど、わからん＼(^o^)／
(参考: https://twitter.com/ReactiveX/status/483625917491970048)