为什么IEnumerable.ToObservable这么慢？

Question

我试图一次枚举一个大的IEnumerable，并观察附带了各种运算符（IEnumerable，Count，Sum等）的枚举。一种明显的方法是使用方法Average将其转换为IObservable，然后为其指定观察者。我注意到，这比其他方法要慢得多，例如做一个简单的循环并在每次迭代时通知观察者，或者使用ToObservable方法而不是ToObservable。区别是巨大的：慢20到30倍。是这样的，还是我做错了？

Observable.Create

输出：

ToObservable

。NET Core 3.0，C＃8，System.Reactive 4.3.2，Windows 10，控制台应用程序，已发布版本

Answer 1

这是表现良好的可观察结果与“自己滚动，因为您认为更快更好，但是它不是”之间的观察结果。

当您深入探究源代码时，您会发现这条可爱的小字句：

using System;
using System.Diagnostics;
using System.Linq;
using System.Reactive.Disposables;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using System.Reactive.Threading.Tasks;

public static class Program
{
    static void Main(string[] args)
    {
        const int COUNT = 10_000_000;
        Method1(COUNT);
        Method2(COUNT);
        Method3(COUNT);
    }

    static void Method1(int count)
    {
        var source = Enumerable.Range(0, count);
        var subject = new Subject<int>();
        var stopwatch = Stopwatch.StartNew();
        source.ToObservable().Subscribe(subject);
        Console.WriteLine($"ToObservable: {stopwatch.ElapsedMilliseconds:#,0} msec");
    }

    static void Method2(int count)
    {
        var source = Enumerable.Range(0, count);
        var subject = new Subject<int>();
        var stopwatch = Stopwatch.StartNew();
        foreach (var item in source) subject.OnNext(item);
        subject.OnCompleted();
        Console.WriteLine($"Loop & Notify: {stopwatch.ElapsedMilliseconds:#,0} msec");
    }

    static void Method3(int count)
    {
        var source = Enumerable.Range(0, count);
        var subject = new Subject<int>();
        var stopwatch = Stopwatch.StartNew();
        Observable.Create<int>(o =>
        {
            foreach (var item in source) o.OnNext(item);
            o.OnCompleted();
            return Disposable.Empty;
        }).Subscribe(subject);
        Console.WriteLine($"Observable.Create: {stopwatch.ElapsedMilliseconds:#,0} msec");
    }
}

每个调度的递归迭代有效地调用一次

ToObservable: 7,576 msec
Loop & Notify: 273 msec
Observable.Create: 511 msec

。

这使您可以为scheduler.Schedule(this, (IScheduler innerScheduler, _ @this) => @this.LoopRec(innerScheduler)); 呼叫选择调度程序。

使用其他选项，您已经创建了一个从头到尾的hasNext = enumerator.MoveNext();调用序列，几乎没有任何作用。 .ToObservable(schedulerOfYourChoice)甚至没有.OnNext通话。

Method2和.Subscribe两者都使用当前线程运行，并且都运行到订阅完成之前的完成。他们正在阻止通话。它们可能导致比赛条件。

Method2是唯一表现良好的可观察对象。它是异步的，可以独立于订户运行。

请记住，可观察值是随时间推移而运行的集合。它们通常具有异步或计时器或对外部刺激的响应。他们通常不会遇到简单的枚举。如果您使用的是枚举，则应该期望同步运行速度更快。

速度不是Rx的目标。目标是对基于时间的推送值执行复杂的查询。