4

わかりました、Rx を理解しようとしていますが、ここで少し迷っています。

FromAsyncPattern は非推奨になったため、ここ(Rx でタスクをライトアップするセクション) の例を使用しましたが、動作します。いくつかの変更を加えただけです。

私が理解していないのは、関数 SumSquareRoots が 2 回呼び出されるのはなぜですか?

 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;
        });
    }
}
4

1 に答える 1

12

これが SumSquareRoots を 2 回呼び出している理由は、2 回サブスクライブしているためです。

// 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はforeachRx の です - ちょうどあなたが 2 回した場合と同じようにforeach、IEnumerable2 倍の仕事をすることになるかもしれません。複数Subscribeの s は複数の仕事を意味します。これを元に戻すには、結果を破棄しないブロッキング呼び出しを使用できます。

Console.WriteLine(res.First());

または、Publish結果を「フリーズ」して、1 人以上のサブスクライバーに再生することもできます ( ToArrayLINQ で使用する方法と似ています)。

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

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

従うことができる一般的なルールは、返さないIObservable<T>、またはTask<T>サブスクリプションを返すRx メソッドは(*) であるということです。

* - 技術的に正しくありません。しかし、このように考えると、あなたの脳は気分が良くなります。

于 2012-11-26T22:30:30.177 に答える