[UniRx]ObserveOnとSubscribeOn

UnityのAPIはメインスレッド上でしか扱えないという制限があります。
しかし、重い処理やネットワークを通した処理などは別のスレッドで実行したほうがよいこともあります。
そこでUniRxのObserveOnとSubscribeOnを使用すれば、簡単に処理を実行するスレッドを変更することができます。

ObserveOn

ObserveOnを使用するとobserver.OnNextやobserver.OnCompletedなどの操作が指定したスケジューラ上で実行されるようになります。
これを使えばSelectやSubscribeに渡す関数を特定のスレッド上で動かすことができます。
ObserveOnMainThreadを使用すればメインスレッドで動かすことができます。

使用例

void Start()
{
    // Scheduler.MainThreadにメインスレッドでアクセスしておく必要がある(初めてアクセスしたときに内部の変数が初期化されるため)
    var s = Scheduler.MainThread;
    new Thread(() =>
    {
        // Debug.Log(this.gameObject.name);  // get_gameObject can only be called from the main thread.
        Observable.Create<string>(observer =>
        {
            observer.OnNext("");
            return null;
        })
        .ObserveOnMainThread()
        .Select(_ => this.gameObject.name)
        .Subscribe(Debug.Log);
    }).Start();
}

注:Threadをそのまま使用しているのはサンプルだからです。普段はObsavable.Startとかを使用したほうがいいです。
ObserveOnMainThreadを使用した後のSelectやSubscribeに渡した関数はメインスレッドで実行されるためthis.gameObject.nameにアクセスしてもエラーになりません。

SubscribeOn

SubscribeOnを使用するとSubscribeが指定したスケジューラ上で実行されるようになります。
内部のSubscribeがスケジューラ上で実行されるようになるだけでSubscribeに渡した関数がスケジューラ上で実行されるわけではないことに注意が必要です。
SubscribeOnMainThreadを途中に挟んでもSubscribeに渡した関数はメインスレッドで実行されないこともあります。
(基本的にはIObserver.OnNextを実行したスレッド上でSubscribeに渡した関数が実行される。変更したい場合はObserveOnを使用する。)

使用例

void Start()
{
    Observable.Create<string>(observer =>
    {
        // 重い処理, メインスレッド上で実行するとゲームが止まってしまう
        Thread.Sleep(3000);
        observer.OnNext("");
        return null;
    })
    // Subscribeをスレッドプール上で実行する
    .SubscribeOn(Scheduler.ThreadPool)
    // UnityAPIを使用するためにメインスレッドに切り替える
    .ObserveOnMainThread()
    .Select(_ => this.gameObject.name)
    .Subscribe(Debug.Log);
}

Obsavable.Createに渡した関数内ではThread.Sleep(3000)を実行しており、メインスレッドで実行するとゲームが止まってしまいます。
SubscribeOn(Scheduler.ThreadPool)を使用することでスレッドプール上でObsavable.Createに渡した関数を実行しています。

失敗例

自分が試行錯誤したときのコードです。

スケジューラの初期化がメインスレッド以外で起こってしまう

void Start()
{
    var subject = new Subject<int>();
    new Thread(() =>
    {
        subject
            // Exception: UniRx requires a MainThreadDispatcher component created on the main thread. Make sure it is added to the scene before calling UniRx from a worker thread.
            .ObserveOnMainThread()
            .Subscribe(_ => Debug.Log(this.gameObject.name));
    }).Start();
}

MainThreadDispatcherが動き出すより前にOnNextしてしまう

void Start()
{
    var subject = new Subject<int>();
    subject
        .SubscribeOnMainThread()
        .Subscribe(_ => Debug.Log("hoge"));
    // OnNextしてもDebug.Log("hoge")が実行されない
    // SubscribeOnMainThread()するとMainThreadDispatcher.Updateが実行されるまではSubscribeされないため
    // 1フレーム待つと正常に動く
    subject.OnNext(0);
}

SubscribeOnMainThreadしてもSubscribeに渡した関数はメインスレッドで実行されるわけではない

void Start()
{
    IObserver<Unit> o = null;
    Observable.Create<Unit>(observer =>
    {
        o = observer;
        return null;
    })
    .SubscribeOnMainThread()
    // get_gameObject can only be called from the main thread.
    .Subscribe(_ => Debug.Log(this.gameObject.name));

    this.UpdateAsObservable()
        .Where(_ => o != null)
        .ObserveOn(Scheduler.ThreadPool)
        .Subscribe(_ => o.OnNext(Unit.Default));
}

最後に

どうしてもわからない部分はソースを読みましょう。
内部の処理を知ることでどのように書くとよいかがなんとなく見えてくると思います。