はじめに

UniTask勉強会を開いたのでその流れでUniRxの記事も書こうと思い3か月経過したのでさすがに執筆しようと思い立ちました。
細かく仕組みや概念の解説をするというよりかは、私がUniRx使ってコーディングするようになるまでのUniRxに対しての認識や考え・必要な知識をまとめていきます。
UniRxを勉強しようと調べると様々な記事が出てきますが、実際どういうときに使えるのか分からない内容が多いと思います。
実用例に焦点を置いていくので、これからUniRxを使っていこうと考えている方の参考になれば幸いです。

UniRxとは

ReactiveExtensions(Rx)がC#にもともとありますが、Unityではそのまま使えないのでUnity用に拡張したものになります。
そもそもRxが何なのかという話も本記事を通して書いていきます。

UniRxとは何なのか

まず本題に入る前に同じ処理をコールバックとRxを使用した場合で比較してみましょう

コールバックを使用した場合

イベントを送る側

using System;
using UnityEngine;

public class Sample : MonoBehaviour
{
    public Action callback;

    void Start()
    {
        callback?.Invoke();
    }
}

イベントを受け取る側

using UnityEngine;

public class OtherClass : MonoBehaviour
{
    [SerializeField] private Sample _sample;

    void Awake()
    {
        _sample.callback = () =>
        {
            Debug.Log("HelloUniRx");
        };
    }
}

SampleクラスのStartが呼ばれたら、OtherClassでログを出すというシンプルなサンプルです。

UniRxを使用した場合

イベントを送る側

using System;
using UniRx;
using UnityEngine;

public class Sample : MonoBehaviour
{
    private Subject<Unit> _subject = new Subject<Unit>();
    public IObservable<Unit> observable => _subject;

    void Start()
    {
        _subject.OnNext(Unit.Default);
    }
}

宣言の部分について解説すると、

private Subject<Unit> _subject = new Subject<Unit>();
public IObservable<Unit> observable => _subject;

Subjectはイベントの通知と購読を担うオブジェクトで、Unit部分はコールバックで言う引数の型です。
通知する際に値を渡さなくてよい場合はUnitを指定して、必要な場合は渡したい方を指定しています。

observableはsubjectの購読する機能(Subscribe)のみを外に公開するために用意しています。
逆に通知する機能(OnNext)のみを外に公開させたい場合はIObserverで公開します。
両方公開したい場合はSubjectをそのまま公開しても大丈夫です。(getterのみで)

_subject.OnNext(Unit.Default);

で、イベントを通知しています。
Unit指定の場合はUnit.Defaultで、それ以外の場合は対象の型のオブジェクトを渡します。

イベントを受け取る側

using UniRx;
using UnityEngine;

public class OtherClass : MonoBehaviour
{
    [SerializeField] private Sample _sample;

    void Awake()
    {
        _sample.observable.Subscribe(_ =>
        {
            Debug.Log("HelloUniRx");
        }).AddTo(this);
    }
}

SubscribeでOnNextが呼ばれたときに実行する処理を書きます。
Subjectで指定した型の値が第一引数に渡されてきます。

AddTo(this)の部分はOtherClassが破棄されたときに購読をやめるという意味です。
もしこれを書かなかった場合はOtherClassが破棄(Destroy)された後もOnNextが呼ばれたときログが出力されてしまいます。
もし中の処理でOtherClassの座標を変えるといった処理が書かれていた場合はNullReferenceが発生してしまうのでAddTo(this)は割と大事な処理です。(後ほど詳しく解説します)

UniRxで書く必要ある？

上記のサンプルではUniRxを使う必要性は感じません。
私も最初UniRxを使う利点が全く理解できませんでした。

HelloWorld的なサンプルでは実感が湧かないのでもう少し具体的なサンプルをUniRxで書いてみましょう。
戦車ゲームで砲弾を発射したことを通知する例です。

using System;
using UniRx;
using UnityEngine;

public class Tank : MonoBehaviour
{
    // 弾を発射した事を通知する
    private Subject<Unit> _fireSubject = new Subject<Unit>();
    public IObservable<Unit> observable => _fireSubject;

    private void Update()
    {
        if (Input.GetKeyDown(KeyCode.Space))
        {
            Fire();
        }
    }

    void Fire()
    {
        _fireSubject.OnNext(Unit.Default);
        // 弾を飛ばす処理
    }
}

シンプルな例ですが、戦車が弾を発射したことが影響する範囲は何が考えられますでしょうか？

残弾表示UI
カメラの演出（揺れなど）
SE
周囲にいる戦車が驚いてビックリマークが頭上にでるかもしれない

特にこのサンプルの戦車ゲームの仕様が決まっているわけではないですが、上記が考えられます。
コールバックでは１対１でしか紐づけられないので限界があります。

その時はイベントを使って複数登録できるようにすればいいじゃないの？

おっしゃる通りです。
しかし、以下の仕様がある場合はどうでしょうか？

最初の一回だけ敵の戦車が弾を発射したときに頭上にビックリマークを出す
いずれかの戦車が弾を発射したらカメラを揺らす
同じフレームで複数の戦車が発射した場合はSEは1回のみ再生する

別にそれぞれの処理の中でフラグなど使ってできるじゃん

おっしゃる通りですが、余計な変数が増えたり冗長な処理になってしまうと思います。
上記の3つの例は以下のUniRxの機能を利用するとシンプルに分かりやすく記述することが可能です。

最初の一回だけ敵の戦車が弾を発射したときに頭上にビックリマークを出す

→Firstフィルタで１回目の通知だけ処理して購読を破棄する。

いずれかの戦車が弾を発射したらカメラを揺らす

→すべての戦車のSubjectをMergeで合成してから購読する。

同じフレームで複数の戦車が発射した場合はSEは1回のみ再生する

→ThrottleFirstFrameで1フレーム内で一番最初に来た通知のみ処理

それぞれが何なのかはこの後解説します。

Rxはイベントの上位互換でイベントと同じことができるだけではなく、イベント以上に柔軟に処理を記述することが可能です。
コールバックやイベントを使わず、UniRxで書いておけば後から必要な処理が発生しても対応しやすいと思います。
(コールバックやイベントを使わないほうが良いという意味ではなく、コールバック以外に用途が考えられないようなものに対してはコールバックを使うなど、UniRxに囚われすぎても良くないです)

UniRxの基本知識

オペレータ

様々なオペレータが用意されていますが、本記事では私がよく使うものをピックアップしてまとめます。
@toRisouPさんがUniRxオペレータ一覧をまとめてくださっているので詳しくはこちらを参照すると良いと思います。

オペレータの説明に関しては実用例を入れると逆に分かりにくくなるので、実用例なしで解説します。

フィルタ

購読する条件または購読終了の条件を指定するオペレータです。

Where

条件を満たしたときに購読処理を行います。

using UniRx;
using UnityEngine;

public class Sample : MonoBehaviour
{
    void Start()
    {
        Subject<int> subject = new Subject<int>();
        subject.Where(v => v >= 2).Subscribe(v =>
        {
            Debug.Log(v);
        });
        
        subject.OnNext(1);
        subject.OnNext(2);
        subject.OnNext(3);
    }
}

結果

First, FirstOrDefault

最初の通知または条件を満たした通知を受け取ったときに購読を終了する

using UniRx;
using UnityEngine;

public class Sample : MonoBehaviour
{
    void Start()
    {
        Subject<int> subject = new Subject<int>();
        // 何かしら通知があったら一つ目だけ処理を行う
        subject.First().Subscribe(v =>
        {
            Debug.Log($"First{v}");
        });

        // 条件が合う物が通知されたらそこで通知を終了する
        subject.First(v => v == 2).Subscribe(v =>
        {
            Debug.Log($"First{v}");
        });
        
        subject.OnNext(1);
        subject.OnNext(2);
        subject.OnNext(3);
        subject.OnNext(2);
    }
}

結果

条件を満たすものが通知されずに購読が終了した場合

using UniRx;
using UnityEngine;

public class Sample : MonoBehaviour
{
    void Start()
    {
        Subject<int> subject = new Subject<int>();
        
        // 条件が合う物が通知されたらそこで通知を終了する
        subject.First(v => v == 4)
            .Subscribe(onNext: v =>
            {
                Debug.Log($"FirstWithCondition{v}");
            }, onError: e =>
            {
                Debug.Log($"FirstError:{e.Message}");
            }, onCompleted: () =>
            {
                Debug.Log("FirstComplete");
            });
        
        subject.FirstOrDefault(v => v == 4)
            .Subscribe(onNext: v =>
            {
                Debug.Log($"FirstOrDefaultWithCondition{v}");
            }, onError: e =>
            {
                Debug.Log($"FirstOrDefaultError:{e.Message}");
            }, onCompleted: () =>
            {
                Debug.Log("FirstOrDefaultComplete");
            });
        
        subject.OnNext(1);
        subject.OnNext(2);
        subject.OnNext(3);

        subject.OnCompleted();
    }
}

結果

Firstは一度も条件に当てはまる通知がないとエラーになり、FirstOrDefaultの方は指定された型のデフォルト値(intの場合0)が通知されて完了します。

ThrottleFirstFrame, ThrottleFirst

最初に通知を受け取ってから指定されたフレーム数または時間購読を無視する。

ThrottleFirstFrame
フレーム指定の場合

using UniRx;
using UnityEngine;

public class Sample : MonoBehaviour
{
    private Subject<long> _subject = new Subject<long>();
    
    void Start()
    {
        _subject.ThrottleFirstFrame(1).Subscribe(v =>
        {
          Debug.Log($"ThrottleFirstFrame:{v}");  
        });
    }

    private int _updateCount = 1;
    void Update()
    {
        Debug.Log($"-----{_updateCount}-----");
        _subject.OnNext(_updateCount);
        _updateCount++;
    }
}

結果

１フレームで指定しているので、最初の通知の次のフレームの通知は無視されます。
また同じフレームで複数回通知があった場合は最初の通知だけ通ります。

ThrottleFirst

時間指定の場合

using System;
using UniRx;
using UnityEngine;

public class Sample : MonoBehaviour
{
    private Subject<Unit> _subject = new Subject<Unit>();
    
    void Start()
    {
        _subject.ThrottleFirst(TimeSpan.FromSeconds(0.5f)).Subscribe(_ =>
        {
          Debug.Log($"ThrottleFirst:{Time.realtimeSinceStartup}");  
        });
    }
    
    void Update()
    {
        _subject.OnNext(Unit.Default);
    }
}

結果

0.5秒おきに購読を一回受け付けています。
正確には通知があってから次の通知を受け付けるまで0.5秒かかります。

ThrottleFrame

ThrottleFirstFrameのほうが使う頻度は多いと思いますが、Firstのついていないほうのオペレーターが若干意味が異なるので一応紹介しておきます。

Firstがついてるほうは一度通知が届いてから、指定されたフレームの間購読を無視するでしたが、
ついていないバージョンは通知が落ち着いてから指定したフレーム経過後に最後に受け取った通知を処理します。

using UniRx;
using UnityEngine;

public class Sample : MonoBehaviour
{
    private Subject<long> _subject = new Subject<long>();
    
    void Start()
    {
        _subject.ThrottleFrame(2).Subscribe(v =>
        {
          Debug.Log($"ThrottleFirstFrame:{v}");  
        });
    }

    private int _updateCount = 1;
    void Update()
    {
        if (_updateCount >= 10 && 20 >= _updateCount)
        {
            Debug.Log($"-----{_updateCount}-----");
            _updateCount++;
            return;
        }
        Debug.Log($"-----{_updateCount}-----");
        _subject.OnNext(_updateCount);
        _updateCount++;
    }
}

結果

ThrottleFrameは2フレームで指定されていて、通知は10フレームから20フレームまで止まっていて、10フレームを含めて2フレーム目に9フレームの時の通知が処理されています。
２フレームの間に受け取った最後の通知を処理するわけではないので注意が必要です。

Throttleの説明はフレーム指定が時間に変わるだけなので割愛させていただきます。

変換

流れてきた値を加工したいときに使用するオペレーターです。
基本Selectだけ覚えていればいろいろ対応できると思います。

Select

強引なサンプルですが、通知するときはGameObjectで通知し、購読する手前でSampleに変換する例です。

using UniRx;
using UnityEngine;

public class Sample : MonoBehaviour
{
    void Start()
    {
        Subject<GameObject> subject = new Subject<GameObject>();
        subject
            .Select(go => go.GetComponent<Sample>())
            .Subscribe(comp =>
            {
                comp.Test();
            });
        subject.OnNext(this.gameObject);
    }

    void Test()
    {
        Debug.Log("ThisIsSample");
    }
}

結果

合成

合成オペレータに関しては種類が多く、奥が深いので本記事では一つだけ紹介します。

Merge

複数のストリームを一つにまとめて購読することが可能です。

using UniRx;
using UnityEngine;

public class Sample : MonoBehaviour
{
    void Start()
    {
        Subject<int> subject1 = new Subject<int>();
        Subject<int> subject2 = new Subject<int>();

        subject1.Merge(subject2).Subscribe(v =>
        {
            Debug.Log(v);
        }).AddTo(this);
        
        subject1.OnNext(1);
        subject2.OnNext(2);
    }
}

結果

購読破棄

購読(Subscribe)を途中でやめたいケースが出てくると思います。
購読破棄する方法を３つ紹介します。

また購読破棄をしっかりしないと、原因が分かりにくい不具合やメモリリークが発生してしてしまいます。
過去に痛い思いをした反省を記事にしています。

Dispose

手動でDispose

Subject<int> subject = new Subject<int>();
var disposable = subject.Subscribe(_ => { });
subject.OnNext(1);
disposable.Dispose();

Subscribe時にIDisposableが返されるのでそれを通して破棄できます。

AddTo

最初のサンプルにあったように一番大事な方法です。

using UniRx;
using UnityEngine;

public class Sample : MonoBehaviour
{
    void Start()
    {
        Subject<int> subject = new Subject<int>();
        subject.Subscribe(_ => { }).AddTo(this);
    }
}

上記のように書くと、SampleがDestroyされたときに購読も破棄されます。

購読がMonoBehaviourと関わっている場合はAddTo(this)は絶対書きましょう。

購読した側のオブジェクトが破棄されても処理が継続して呼ばれてしまいます。

CompositeDisposable

破棄するタイミングが複数ある場合、ComositeDisposableを使用することで一つにまとめることが可能です。

以下は_gameObjects のいずれかがDestroyされたときに購読を破棄する例です。

using System.Collections.Generic;
using UniRx;
using UnityEngine;

public class Sample : MonoBehaviour
{
    [SerializeField] private List<GameObject> _gameObjects = new List<GameObject>();
    
    void Start()
    {
        CompositeDisposable compositeDisposable = new CompositeDisposable();
        foreach (var go in _gameObjects)
        {
            compositeDisposable.AddTo(go);
        }
        Subject<int> subject = new Subject<int>();
        subject.Subscribe(_ => { }).AddTo(compositeDisposable);
    }
}

Subject

これまでSubjectを例に挙げて解説してきたので、Subjectの詳しい説明は要らないと思います。
イベントの通知と購読をするものという認識で実用上は大丈夫です。

ReactiveProperty

int型のhpという値が変更されたときに残りHPを通知したい場合、Subjectを使用すると以下のようになりますが無駄があるように見えます。

using UniRx;
using UnityEngine;

public class Sample : MonoBehaviour
{
    private int _hp = 50;
    private Subject<int> _hpSubject = new Subject<int>();

    void Start()
    {
        _hpSubject.Subscribe(v =>
        {
            Debug.Log($"HpChanged{v}");
        }).AddTo(this);
    }

    void Update()
    {
        if (Input.GetKeyDown(KeyCode.Space))
        {
            _hp--;
            _hpSubject.OnNext(_hp);
        }
    }
}

ReactivePropertyを用いるともっとシンプルに書けます

using UniRx;
using UnityEngine;

public class Sample : MonoBehaviour
{
    private IntReactiveProperty _hp = new IntReactiveProperty(50);

    void Start()
    {
        _hp.Subscribe(v =>
        {
            Debug.Log($"HpChanged{v}");
        }).AddTo(this);
    }

    void Update()
    {
        if (Input.GetKeyDown(KeyCode.Space))
        {
            _hp.Value--;
        }
    }
}

このように値が変更されたときに自動的にOnNextを発行してくれる便利なものがReactivePropertyです。
IntReactivePropertyのほかにも

StringReactiveProperty
Vector3ReactiveProperty
BoolReactiveProperty
ColorReactiveProperty

など様々な型が用意されており、
用意されていないものに関してはReactiveProperty
ListはReactiveCollection、
DictionaryはReactiveDictionary
を利用することができます。

MessageBroker

Subject,ReactivePropertyは通知するオブジェクトは１種類だけでしたが、
MessageBrokerは複数の型のオブジェクトの通知を一つのオブジェクトで行うことができます。

using UniRx;
using UnityEngine;

public class Sample : MonoBehaviour
{
    public class Event1 { }

    public class Event2 { }
    
    private MessageBroker _messageBroker = new MessageBroker();

    void Start()
    {
        _messageBroker.Receive<Event1>().Subscribe(_ =>
        {
            Debug.Log("Event1");
        }).AddTo(this);
        
        _messageBroker.Receive<Event2>().Subscribe(_ =>
        {
            Debug.Log("Event2");
        }).AddTo(this);
    }

    void UpDate()
    {
        if (Input.GetKeyDown(KeyCode.A))
        {
            _messageBroker.Publish(new Event1());
        }
        
        if (Input.GetKeyDown(KeyCode.B))
        {
            _messageBroker.Publish(new Event2());
        }
    }
}

また、アプリ全体に対してイベントを通知したい時のためにMessageBroker.Defaultが用意されています。

MessageBroker.Default.Receive<Event1>().Subscribe(_ => { });
MessageBroker.Default.Publish(new Event1());

どこでも通知出来てどこでも通知を受け取ることが可能ですが、あまり多様すべきものではないです。

トリガー系

MonoBehaviourのイベント用のトリガーが用意されているので購読することが可能です。

以下で少しだけ紹介します。

using UniRx;
using UniRx.Triggers;
using UnityEngine;

public class Sample : MonoBehaviour
{
    void Start()
    {
        this.OnDestroyAsObservable().Subscribe(_ => { });
        this.OnCollisionEnterAsObservable().Subscribe(collision => { });
        this.OnTriggerEnterAsObservable().Subscribe(collider => { });
        this.OnEnableAsObservable().Subscribe(_ => { });
        this.OnDisableAsObservable().Subscribe(_ => { });
    }
    
}

こちらにTrigger系のスクリプトがあります。

ほとんど網羅されてますね。

UniRxの利用例

UniRxの利用例を紹介します。

通信による値の変更通知

例えばソーシャルゲームの残りスタミナ数で考えてみましょう。
スタミナの数は、クエスト開始時・アイテム受け取り時・時間経過回復など様々なタイミングで変化します。
ヘッダーに表示されているスタミナ数はスタミナ数が変わるたびに更新しないといけません。
Updateで毎フレーム値を監視しても対応可能ですが、無駄なループ処理になってしまいます。
通信終了後にUniRxを通してスタミナ変更通知を送る仕組みを作っておけば必要なタイミングで、
さらにヘッダー以外のUIも変更通知を受け取ることが可能になります。

１フレーム内でとりあえずイベントを受け取り、１回だけ処理したい

最初にあげた例がまさにこちらのケースです。

同じフレームで複数の戦車が発射した場合はSEは1回のみ再生する
→ThrottleFirstFrameで1フレーム内で一番最初に来た通知のみ処理

また、ダメージを受けたときにHPゲージを振動させたい場合、
同一フレームで複数回ダメージを受けるタイミング可能性があります。
また、振動が終了するまで振動させたくないケースも考えられると思います。
例えば振動が１５フレーム続く場合、ThrottleFirstFrame(15)と指定するだけでシンプルに対応できます。

依存関係が離れている場所にイベントを送りたい

MessageBroker.Defaultがこちらのケースで利用できます。
もちろん多用するのはいけないですが、staticなクラスや変数を増やしたり、複数のオブジェクトをはしごして処理を伝えるよりは潔くMessageBroker.Default利用したほうが良いと思います。

複雑な入力受付に対応する

フリック・ダブルタップやVRコントローラーを一回振ったときなど複雑な入力にUniRxを使うことができます。
合成オペレーターを駆使するとコンパクトに入力判定をまとめることができます。
こちらに関しては結構難易度は高めですが、Updateでループを何回も回して判定する方法よりはメンテナンスがしやすいです。

データとUIの表示を同期させる

データとUIの表示を同期させるのがRxの王道な使い方だと思います。
複雑なUIだと所々で表示の更新処理が必要で、それぞれのタイミングで更新処理を呼び出すとどこかで処理を忘れ、表示が正しくない・おかしいことになってしまうことは避けられないでしょう。
UnityのGUIはMVPというパターンで実装するのが良いといわれています。

こちらに関しては@toRisouPさんの記事がとても参考になります。

また、ボタンのOnClickなどのUnityEventをUniRxに変換できるようにも拡張されていて便利です。

Button bt;
bt.onClick.AsObservable().Subscribe(_ => {});

UniTaskとの組み合わせ

UniRxを利用したawait

ストリームがCompleteになるまでawaitすることができるので、それを利用して入力待ちを非同期で行うことなどができます。

using Cysharp.Threading.Tasks;
using UniRx;
using UnityEngine;

public class Sample : MonoBehaviour
{
    async void Start()
    {
        await Observable.EveryUpdate().First(_ => Input.GetKeyDown(KeyCode.A)).ToUniTask(cancellationToken: this.GetCancellationTokenOnDestroy());
        Debug.Log("AKeyPressed");
    }
    
}

UniTaskPubSub

若干番外編ですが、MessageBrokerのUniTaskバージョンです。

イベントを発行したときに非同期の購読処理がすべて完了するまで、awaitすることができます。

var messageBus = new AsyncMessageBus();

// Subscriber

messageBus.Subscribe<FooMessage>(async msg =>
{
    await DoSomething1Async(msg.Id);
});

messageBus.Subscribe<FooMessage>(async msg =>
{
    await DoSomething2Async(msg.Id);
});


// Publisher

// Await for all subscribers.
await messageBus.PublishAsync(new FooMessage { Id = 1 });

// After PublishAsync awaited, DoSomething1Async and DoSomething2Async have been completed.

最後に

UniRxに焦点を当てて普段私がよく使うものや使い方などをまとめてみました。

この記事にまとめきれない内容や、私がまだ知らない概念もUniRxにはあるので本当に奥が深いです。

奥が深い故に、UniRxを使う敷居が高くなってしまっていると思いますが、

最初のうちは難しい概念を理解する必要はなく、本記事で解説した内容を使えるようになればある程度は実用的に使えるはずです。

本記事がUniRxを使うきっかけになれば嬉しいです！！