c# - ReactiveExtensions Observable FromAsync 调用两次函数

Question

好吧，试图理解 Rx，有点迷失在这里。

FromAsyncPattern 现已弃用，因此我采用了 here 中的示例(使用 Rx 点亮任务部分)，它有效，我只是做了一些更改，不使用 await 只是等待可观察和订阅......

我不明白的是为什么调用函数 SumSquareRoots 的两倍？

 var res = Observable.FromAsync(ct => SumSquareRoots(x, ct))
                                      .Timeout(TimeSpan.FromSeconds(5));

            res.Subscribe(y => Console.WriteLine(y));

            res.Wait();


class Program
{
    static void Main(string[] args)
    {
        Samples();
    }

    static void Samples()
    {
        var x = 100000000;

        try
        {
            var res = Observable.FromAsync(ct => SumSquareRoots(x, ct))
                                      .Timeout(TimeSpan.FromSeconds(5));

            res.Subscribe(y => Console.WriteLine(y));

            res.Wait();
        }
        catch (TimeoutException)
        {
            Console.WriteLine("Timed out :-(");
        }
    }

    static Task<double> SumSquareRoots(long count, CancellationToken ct)
    {
        return Task.Run(() =>
        {
            var res = 0.0;
            Console.WriteLine("Why I'm called twice");
            for (long i = 0; i < count; i++)
            {
                res += Math.Sqrt(i);

                if (i % 10000 == 0 && ct.IsCancellationRequested)
                {
                    Console.WriteLine("Noticed cancellation!");
                    ct.ThrowIfCancellationRequested();
                }
            }

            return res;
        });
    }
}

Answer 1

两次调用 SumSquareRoots 的原因是因为您订阅了两次:

// Subscribes to res
res.Subscribe(y => Console.WriteLine(y));

// Also Subscribes to res, since it *must* produce a result, even
// if that result is then discarded (i.e. Wait doesn't return IObservable)
res.Wait();

Subscribe是 foreach Rx 的 - 就像你 foreach一个IEnumerable两次，你最终可能会做 2 倍的工作，多个 Subscribe s 表示工作的倍数。要撤消此操作，您可以使用不会丢弃结果的阻塞调用:

Console.WriteLine(res.First());

或者，您可以使用 Publish “卡住”结果并将其回放给 > 1 个订阅者(有点像您在 LINQ 中使用 ToArray 的方式):

res = res.Publish();
res.Connect();

// Both subscriptions get the same result, SumSquareRoots is only called once
res.Subscribe(Console.WriteLine);
res.Wait();

您可以遵循的一般规则是，任何不返回IObservable<T> 的Rx 方法。或 Task<T>将导致订阅(*)

* - 技术上不正确。但如果你这样想，你的大脑会感觉更好。