RxSwiftによる実行スレッドの切り替え #Swift

この記事の前提環境は Xcode 7 / RxSwift 2.0.0-beta です

RxSwiftにおける実行スレッドの切り替えはシンプルです。実行スレッドの切り替えには observeOn と subscribeOn を使います。言葉で説明するより実際に使い方を見てしまうのがわかりやすい思うので、早速コードを見てしまうことにしましょう。

let disposeBag = DisposeBag()
let scheduler = SerialDispatchQueueScheduler(globalConcurrentQueuePriority: .Default)
[1,2,3].asObservable()
    .doOn(onNext: { _ in NSLog("doOn") })
    .observeOn(scheduler)
    .subscribeNext({ _ in NSLog("subscribeNext") })
    .addDisposableTo(disposeBag)
NSLog("main")

// => ログ出力
// 2015-11-08 07:56:00.821 xctest[11841:2748936] doOn
// 2015-11-08 07:56:00.821 xctest[11841:2748936] doOn
// 2015-11-08 07:56:00.821 xctest[11841:2748993] subscribeNext
// 2015-11-08 07:56:00.821 xctest[11841:2748936] doOn
// 2015-11-08 07:56:00.822 xctest[11841:2748993] subscribeNext
// 2015-11-08 07:56:00.822 xctest[11841:2748936] main
// 2015-11-08 07:56:00.822 xctest[11841:2748993] subscribeNext

observeOn より手前の doOn はメインスレッド(2748936)で実行され、subscribeNext はバックグラウンドのスレッド(2748993)で直列に実行されていることがわかります。

ここで、スケジューラを OperationQueueScheduler に差し替えてみましょう。

let disposeBag = DisposeBag()
let scheduler = OperationQueueScheduler(operationQueue: NSOperationQueue())
[1,2,3].asObservable()
    .doOn(onNext: { _ in NSLog("doOn") })
    .observeOn(scheduler)
    .subscribeNext({ _ in NSLog("subscribeNext") })
    .addDisposableTo(disposeBag)
NSLog("main")

// => ログ出力
// 2015-11-08 08:02:46.161 xctest[12542:2757188] doOn
// 2015-11-08 08:02:46.161 xctest[12542:2757188] doOn
// 2015-11-08 08:02:46.161 xctest[12542:2757188] doOn
// 2015-11-08 08:02:46.162 xctest[12542:2757188] main
// 2015-11-08 08:02:46.161 xctest[12542:2757256] subscribeNext
// 2015-11-08 08:02:46.163 xctest[12542:2757248] subscribeNext
// 2015-11-08 08:02:46.163 xctest[12542:2757256] subscribeNext

すると、subscribeNext の実行スレッドがバラバラ(2757256, 2757248)になって並列に実行されていることがわかると思います。

このように observeOn はそのあとの処理を、引数で指定されたスケジューラを利用するように指定するときに用いることができます。subscribeOn は逆にその前の処理を、引数で指定されたスケジューラを利用するように指定したいときに用いることができます。

しばしば iOS アプリでは通信処理を走らせて、その結果をいい感じに加工してビューに反映させるという処理をすることがあると思います。ビューに反映させるときはメインスレッドから行わなければならないので、observeOn でメインスレッドに割り当てるスケジューラを指定してあげる必要があります。雑な例ではありますが、以下のような形になるでしょう。

let url = NSURL(string: "http://gochiusa.com")!
let request = NSURLRequest(URL: url)
NSURLSession.sharedSession().rx_response(request)
    .subscribeOn(backgroundWorkScheduler)
    .observeOn(MainScheduler.sharedInstance)
    .subscribeNext({ (data, response) -> Void in
        webView.loadData(data, MIMEType: "text/html", textEncodingName: "utf-8", baseURL: url)
    })
    .addDisposableTo(disposeBag)

まとめると以下の３点のような内容となります。

スレッドの指定には observeOn と subscribeOn を用いる
subscribeOn はそれ以前の処理を実行するスケジューラを指定できる
observeOn はそれ以後の処理を実行するスケジューラを指定できる

各々のスケジューラについてはRxSwiftの実装コードコメントに記載してありますので、そちらを参照するのが良いでしょう。スケジューラの実装コードは https://github.com/ReactiveX/RxSwift/tree/master/RxSwift/Schedulers のあたりにありますので是非目を通しておきましょう。