又見Rx——Rx via UniRx

zhuanlan.zhihu.com

)）。

以至於，Rx的源碼當中存在大量關於Dispose的代碼。

在調用SubscribeCore中需要傳入一個ICancelable，並且需要返回一個IDisopsable。

SubscribeCore的函數原型：

protected override IDisposable SubscribeCore(IObserver<T> observer, IDisposable cancel)

ICancelable的介面：

public interface ICancelable : IDisposable{ bool IsDisposed { get; }}

當然，並不是所有東西都需要實現，所以Rx通常會返回一個代表不需要Dispose的常量來代表不需要銷毀。

Linq操作

事實上，我們剛剛已經接觸了一部分包裝模式，只不過我們的數據源是直接創建出來的，那麼Linq操作又是如何實現的呢？

類似於.Where、.Amb等等操作是如何連起來讓整個數據流通過短短的幾個函數就可以被我們自由操作？

答案還是包裝模式。

我們進入我們找到Where的實現，實際上是創建了一個名叫WhereObservable類的實例，我們到裡面去看，與Range的實現大同小異。

它的SubscribeCore非常簡單：

protected override IDisposable SubscribeCore(IObserver<T> observer, IDisposable cancel){ if (predicate != null) { return source.Subscribe(new Where(this, observer, cancel)); } else { return source.Subscribe(new Where_(this, observer, cancel)); }}

只是對Observer進行簡單包裝之後直接將其監聽到傳進來的數據源中。

所以我們只需要看Where這個Observer即可：

class Where : OperatorObserverBase<T, T> { readonly WhereObservable<T> parent; public Where(WhereObservable<T> parent, IObserver<T> observer, IDisposable cancel) : base(observer, cancel) { this.parent = parent; } public override void OnNext(T value) { var isPassed = false; try { isPassed = parent.predicate(value); } catch (Exception ex) { try { observer.OnError(ex); } finally { Dispose(); } return; } if (isPassed) { observer.OnNext(value); } } }

我們看到它的代碼非常簡單，OnNext的時候判斷一下條件，如果條件通過則調用其包裝好的Observer，否則就直接跳過，出錯就OnError。

舉一反三，我們可以通過這種方法來做其他各類的操作符，類似於Select、Amb等等等，方式完全一樣！只不過是包裝了傳入的Observer而已！

Schedule調度器

事實上，我們大多數的時候都不會注意到調度器的存在，因為它大多數情況下以默認值的形式存在，當我們想要更加靈活地使用Rx的時候也許就會用到它。

調度器的介面如下：

public interface IScheduler{ DateTimeOffset Now { get; } // Interface is changed from official Rx for avoid iOS AOT problem (state is dangerous). IDisposable Schedule(Action action); IDisposable Schedule(TimeSpan dueTime, Action action);}

其中Now代表的是當前時間，下面兩個方法一個是對傳入的方法進行調度，而另一個是傳入時間對行為進行調度。

查看引用之後我們會發現很多與時間相關的Observer與Observable都會採用調度器來進行實現，例如Range可以使用自定義的調度器來創建，我們可以隔幾秒鐘之後將一個值出隊，其默認調度器是ImmediateScheduler也就是直接調用。

代碼如下

//直接進行調度if (scheduler == Scheduler.Immediate){ for (int i = 0; i < count; i++) { int v = start + i; observer.OnNext(v); } observer.OnCompleted(); return Disposable.Empty;}//其他的調度方式，傳入的Action類似於行為隊列的實現else{ var i = 0; return scheduler.Schedule((Action self) => { if (i < count) { int v = start + i; observer.OnNext(v); i++; self(); } else { observer.OnCompleted(); } });}

一些操作符也用到了調度器，例如Timer、Delay等。

以下為Delay中的代碼片段：

public override void OnNext(T value){ var next = parent.scheduler.Now.Add(delay); var shouldRun = false; lock (gate) { queue.Enqueue(new Timestamped<T>(value, next)); shouldRun = ready && !active; active = true; } if (shouldRun) { cancelable.Disposable = parent.scheduler.Schedule(delay, DrainQueue); }}

當有新值到達的時候先放入隊列，通過調度器來調用出隊函數以達到Delay的目的。

調度器實際上也只是將行為放入隊列，然後在Thread中進行wait，並沒有使用協程，畢竟UniRx大部分代碼是從http://Rx.Net改過來的吧。

對接Unity功能

講到這裡，事實上或許連http://Rx.Net與UniRx的區別都沒有提到，但是最核心的內容已經講完了，UniRx中提供了大量的方法來幫助開發者在Unity中使用Rx，實際上我們只需要將Unity中的各個回調包裝成事件流便可以使用Rx了。

類似我們剛剛提及的雙擊功能：

//創建一個點擊事件流var clickStream = Observable.EveryUpdate() .Where(_ => Input.GetMouseButtonDown(0));

每一次Update被看做是一次事件，從中選出有點擊事件的幀，即可獲得點擊事件流。

接下來的使用方法便和其他任何一種平台的Rx都一樣了。

這部分的源碼全部都在UnityEngineBridge文件夾下，都非常好理解，有需要的同學也可以擴展出自己的事件流，非常容易。

未提及的部分

既然我們已經知道了Observable、Linq以及Scheduler是怎麼一回事，事實上我們自己也能去實現一個簡單的Rx了，但是事實上還有一些我們沒有閱讀的部分，大家可以繼續深入閱讀。

* UniRx在多線程中的情形

* UniRx在IOS上避免JIT發生而進行的重寫

* 在MVVM中使用Rx

推薦資料

Rx官網：(