UniRx - Unity响应式编程插件

Posted 2021-04-30 洪流学堂

本文译者：郑洪智 - 你的技术探路者 
翻译日期2018年2月1日 
本文翻译自UniRx(https://github.com/neuecc/UniRx)插件的ReadMe 
这个插件是我特别喜欢的一个插件，希望能将这种技术思想传播给大家

UniRx - Unity响应式编程插件

插件作者Yoshifumi Kawai(neuecc) 
本文译者：郑洪智 - 你的技术探路者

UniRx是什么?

UniRx (Unity响应式编程插件) 重写了.Net的响应式扩展。.Net官方的Rx很棒，但是在Unity中无法使用，并且与ios的IL2CPP有兼容性问题。这个库这些问题并且添加了一些Unity专属的工具类。 支持的平台有：PC/Mac/android/iOS/WP8/WindowsStore/等等，并且支持Unity4.6之后的所有版本。

演讲PPT - http://www.slideshare.net/neuecc/unirx-reactive-extensions-for-unityen

更新博客 - https://medium.com/@neuecc

在Unity论坛中提供支持 - http://forum.unity3d.com/threads/248535-UniRx-Reactive-Extensions-for-Unity

更新日志 UniRx/releases(https://github.com/neuecc/UniRx/releases)

UniRx 包含 Core Library (Port of Rx) + Platform Adaptor (MainThreadScheduler/FromCoroutine/etc) + Framework (ObservableTriggers/ReactiveProeperty/etc)

为什么用Rx?

一般来说，网络操作需要用到 WWW 和 Coroutine。但是使用 Coroutine 对于异步操作来说不是一个好的选择，原因如下：

1. 协程不能有返回值，因为它返回类型必须是IEnumerator

2. 协程不能处理异常，因为 yield return 语句没办法被 try-catch

会造成代码大面积的强耦合。

Rx就是为了解决异步问题而来的。Rx可以让异步操作更优雅，使用事件驱动编程，使用LINQ操作。

游戏循环 (every Update, OnCollisionEnter, etc), 传感器数据 (Kinect, Leap Motion, VR Input, etc.) 都是事件。Rx将事件转化为响应式的序列，通过LINQ操作可以很简单地组合起来，还支持时间操作。

Unity通常是单线程，但是UniRx可以让多线程更容易。

UniRx 可以简化 uGUI 的编程，所有的UI事件 (clicked, valuechanged, etc) 可以转化为 UniRx 的事件流。

简介

介绍 Rx 的非常棒的文章: 《The introduction to Reactive Programming you've been missing》(https://gist.github.com/staltz/868e7e9bc2a7b8c1f754).

下面的代码实现了双击的检测：

var clickStream = Observable.EveryUpdate()
    .Where(_ => Input.GetMouseButtonDown(0));
clickStream.Buffer(clickStream.Throttle(TimeSpan.FromMilliseconds(250)))
    .Where(xs => xs.Count >= 2)
    .Subscribe(xs => Debug.Log("DoubleClick Detected! Count:" + xs.Count));

这个例子仅用5行代码，展示出了下面的特性：

• 将游戏循环为 (Update) 变成事件流

• 组合事件流

• 合并自身事件流

• 基于时间的操作非常简单

网络操作

使用 ObservableWWW 进行异步网络操作。它的 Get/Post 方法返回可订阅(Subscribe)的 IObservables:

ObservableWWW.Get("http://google.co.jp/")
    .Subscribe(
        x => Debug.Log(x.Substring(0, 100)), // onSuccess
        ex => Debug.LogException(ex)); // onError

Rx 可以组合和取消。你也可以通过LINQ表达式进行查询：

// composing asynchronous sequence with LINQ query expressions
var query = from google in ObservableWWW.Get("http://google.com/")
            from bing in ObservableWWW.Get("http://bing.com/")
            from unknown in ObservableWWW.Get(google + bing)
            select new { google, bing, unknown };

var cancel = query.Subscribe(x => Debug.Log(x));

// Call Dispose is cancel.
cancel.Dispose();

并行请求使用 Observable.WhenAll ：

// Observable.WhenAll is for parallel asynchronous operation
// (It's like Observable.Zip but specialized for single async operations like Task.WhenAll)
var parallel = Observable.WhenAll(
    ObservableWWW.Get("http://google.com/"),
    ObservableWWW.Get("http://bing.com/"),
    ObservableWWW.Get("http://unity3d.com/"));

parallel.Subscribe(xs =>
{
    Debug.Log(xs[0].Substring(0, 100)); // google
    Debug.Log(xs[1].Substring(0, 100)); // bing
    Debug.Log(xs[2].Substring(0, 100)); // unity
});

也可以获取进度信息：

// notifier for progress use ScheudledNotifier or new Progress<float>(/* action */)
var progressNotifier = new ScheduledNotifier<float>();
progressNotifier.Subscribe(x => Debug.Log(x)); // write www.progress

// pass notifier to WWW.Get/Post
ObservableWWW.Get("http://google.com/", progress: progressNotifier).Subscribe();

错误处理：

// If WWW has .error, ObservableWWW throws WWWErrorException to onError pipeline.
// WWWErrorException has RawErrorMessage, HasResponse, StatusCode, ResponseHeaders
ObservableWWW.Get("http://www.google.com/404")
    .CatchIgnore((WWWErrorException ex) =>
    {
        Debug.Log(ex.RawErrorMessage);
        if (ex.HasResponse)
        {
            Debug.Log(ex.StatusCode);
        }
        foreach (var item in ex.ResponseHeaders)
        {
            Debug.Log(item.Key + ":" + item.Value);
        }
    })
    .Subscribe();

和 IEnumerators 一起使用(协程)

IEnumerator (协程) 是Unity主要的异步工具。UniRx 集成了协程和 IObservables。 你可以用协程写异步代码，然后用 UniRx 来组织他们。这是控制异步流的最好的方法。

// two coroutines

IEnumerator AsyncA()
{
    Debug.Log("a start");
    yield return new WaitForSeconds(1);
    Debug.Log("a end");
}

IEnumerator AsyncB()
{
    Debug.Log("b start");
    yield return new WaitForEndOfFrame();
    Debug.Log("b end");
}

// main code
// Observable.FromCoroutine converts IEnumerator to Observable<Unit>.
// You can also use the shorthand, AsyncA().ToObservable()

// after AsyncA completes, run AsyncB as a continuous routine.
// UniRx expands SelectMany(IEnumerator) as SelectMany(IEnumerator.ToObservable())
var cancel = Observable.FromCoroutine(AsyncA)
    .SelectMany(AsyncB)
    .Subscribe();

// you can stop a coroutine by calling your subscription's Dispose.
cancel.Dispose();

在 Unity 5.3 或更新版本里, 可以使用 ToYieldInstruction 将 Observable 转为 Coroutine。

IEnumerator TestNewCustomYieldInstruction()
{
    // wait Rx Observable.
    yield return Observable.Timer(TimeSpan.FromSeconds(1)).ToYieldInstruction();

    // you can change the scheduler(this is ignore Time.scale)
    yield return Observable.Timer(TimeSpan.FromSeconds(1), Scheduler.MainThreadIgnoreTimeScale).ToYieldInstruction();

    // get return value from ObservableYieldInstruction
    var o = ObservableWWW.Get("http://unity3d.com/").ToYieldInstruction(throwOnError: false);
    yield return o;

    if (o.HasError) { Debug.Log(o.Error.ToString()); }
    if (o.HasResult) { Debug.Log(o.Result); }

    // other sample(wait until transform.position.y >= 100) 
    yield return this.transform.ObserveEveryValueChanged(x => x.position).FirstOrDefault(p => p.y >= 100).ToYieldInstruction();
}

通常，协程想要返回一个值需要使用回调callback。Observable.FromCoroutine 可以将协程转为可以取消的 IObservable[T]。

// public method
public static IObservable<string> GetWWW(string url)
{
    // convert coroutine to IObservable
    return Observable.FromCoroutine<string>((observer, cancellationToken) => GetWWWCore(url, observer, cancellationToken));
}

// IObserver is a callback publisher
// Note: IObserver's basic scheme is "OnNext* (OnError | Oncompleted)?" 
static IEnumerator GetWWWCore(string url, IObserver<string> observer, CancellationToken cancellationToken)
{
    var www = new UnityEngine.WWW(url);
    while (!www.isDone && !cancellationToken.IsCancellationRequested)
    {
        yield return null;
    }

    if (cancellationToken.IsCancellationRequested) yield break;

    if (www.error != null)
    {
        observer.OnError(new Exception(www.error));
    }
    else
    {
        observer.OnNext(www.text);
        observer.OnCompleted(); // IObserver needs OnCompleted after OnNext!
    }
}

这有一些例子。接下来是一个多OnNext模式。

public static IObservable<float> ToObservable(this UnityEngine.AsyncOperation asyncOperation)
{
    if (asyncOperation == null) throw new ArgumentNullException("asyncOperation");

    return Observable.FromCoroutine<float>((observer, cancellationToken) => RunAsyncOperation(asyncOperation, observer, cancellationToken));
}

static IEnumerator RunAsyncOperation(UnityEngine.AsyncOperation asyncOperation, IObserver<float> observer, CancellationToken cancellationToken)
{
    while (!asyncOperation.isDone && !cancellationToken.IsCancellationRequested)
    {
        observer.OnNext(asyncOperation.progress);
        yield return null;
    }
    if (!cancellationToken.IsCancellationRequested)
    {
        observer.OnNext(asyncOperation.progress); // push 100%
        observer.OnCompleted();
    }
}

// usecase
Application.LoadLevelAsync("testscene")
    .ToObservable()
    .Do(x => Debug.Log(x)) // output progress
    .Last() // last sequence is load completed
    .Subscribe();

未完结，敬请期待后续连载

---

---