Почему IEnumerable.ToObservable такой медленный?

В этом разница между хорошо себя зарекомендовавшей наблюдаемой и наблюдаемой «самокрутись, потому что ты думаешь, что чем быстрее, тем лучше, но это не так».

Когда вы углубитесь в источник, вы обнаружите эту прекрасную строчку:

scheduler.Schedule(this, (IScheduler innerScheduler, _ @this) => @this.LoopRec(innerScheduler));

Эффективно вызывает hasNext = enumerator.MoveNext(); один раз за запланированную рекурсивную итерацию.

Это позволяет вам выбрать планировщик для вашего .ToObservable(schedulerOfYourChoice) вызова.

С другими выбранными вами вариантами вы создали простую серию вызовов .OnNext, которые практически ничего не делают. У Method2 нет даже .Subscribe звонка.

Оба из Method2 и Method1 выполняются с использованием текущего потока и оба выполняются до завершения до завершения подписки. Они блокируют звонки. Они могут вызвать состояние гонки.

Method1 — единственная, которая хорошо ведет себя как наблюдаемая. Он асинхронный и может работать независимо от подписчика.

Имейте в виду, что наблюдаемые объекты — это коллекции, которые выполняются во времени. Обычно они имеют асинхронный источник, таймер или реакцию на внешний стимул. Они не часто убегают от простого перечисления. Если вы работаете с перечислимым, следует ожидать, что синхронная работа будет работать быстрее.

Скорость не является целью Rx. Целью является выполнение сложных запросов к временным, отправленным значениям.

Потому что Тема ничего не делает.

Похоже, что производительность оператора цикла различна для двух случаев:

for(int i=0;i<1000000;i++)
    total++;

or

for(int i=0;i<1000000;i++)
    DoHeavyJob();

Если использовать другой Subject с медленной реализацией OnNext, результат будет более приемлемым.

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 = 100;
        Method1(COUNT);
        Method2(COUNT);
        Method3(COUNT);
    }

    class My_Slow_Subject : SubjectBase<int>
    {

        public override void OnNext(int value)
        {
            //do a job which spend 3ms
            System.Threading.Thread.Sleep(3);
        }


        bool _disposed;
        public override bool IsDisposed => _disposed;
        public override void Dispose() => _disposed = true;
        public override void OnCompleted() { }
        public override void OnError(Exception error) { }
        public override bool HasObservers => false;
        public override IDisposable Subscribe(IObserver<int> observer) 
                => throw new NotImplementedException();
    }

    static SubjectBase<int> CreateSubject()
    {
        return new My_Slow_Subject();
    }

    static void Method1(int count)
    {
        var source = Enumerable.Range(0, count);
        var subject = CreateSubject();
        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 = CreateSubject();
        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 = CreateSubject();
        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: 434 msec
Loop & Notify: 398 msec
Observable.Create: 394 msec

Поддержка ToObservable System.Reactive.Concurrency.IScheduler

Это означает, что вы можете реализовать свой собственный IScheduler и решать, когда запускать каждую задачу.

Надеюсь это поможет

С Уважением