.net - 您如何使用反应式框架将两个异步操作链接在一起？

Question

我真正想做的就是完成两个异步操作，一个接一个。例如

下载网站 X。完成后，下载网站 Y。

Answer 1

从两个 observable 中执行 SelectMany（使用 ToAsync 将操作作为 IObservable 获取），就像 LINQ 选择很多（或使用语法糖）：

var result = 
from x in X
from y in Y
select new { x, y }  

还有其他选择，但这取决于您的具体情况。

Answer 2

我自己不喜欢这样的建议，但是...

如果您需要按顺序执行两个操作，请按顺序执行（您仍然可以在与 main 不同的线程上执行它们）。

如果您仍然想在代码中分离任务，那么来自 System.Parallel 的新构造将是合适的：

var task1 = Task.Factory.StartNew (() => FirstTask());
var task2 = task1.ContinueWith (frst => SecondTask ());

如果这是一种了解 system.reactive 的方法，请尝试 Observable.GenerateInSequence - 但它肯定会过分。请记住，observable 是 enumerable 的对应物，最好以类似的方式使用（迭代数据）。它不是运行异步操作的工具。

编辑：我想承认我错了，理查德是对的——在我回复的时候，我对 RX 并不完全满意。现在我认为RX是启动异步操作最自然的方式。然而，理查德的回答有点吝啬，应该是：

var result =  
from x in XDownload.ToAsync()() 
from y in YDownload.ToAsync()() 
select y   

Answer 3

也许是以下几行：

    static IObservable<DownloadProgressChangedEventArgs> CreateDownloadFileObservable(string url, string fileName)
    {
       IObservable<DownloadProgressChangedEventArgs> observable =
           Observable.CreateWithDisposable<DownloadProgressChangedEventArgs>(o =>
        {
            var cancellationTokenSource = new CancellationTokenSource();
            Scheduler.TaskPool.Schedule
            (
                () =>
                {

                    Thread.Sleep(3000);
                    if (!cancellationTokenSource.Token.IsCancellationRequested)
                    {
                        WebClient client = new WebClient();
                        client.DownloadFileAsync(new Uri(url), fileName,fileName);


                        DownloadProgressChangedEventHandler prgChangedHandler = null;
                        prgChangedHandler = (s,e) =>
                            {                                                   
                                o.OnNext(e);
                            };


                        AsyncCompletedEventHandler handler = null;
                        handler = (s, e) =>
                        {
                            prgChangedHandler -= prgChangedHandler;                            
                            if (e.Error != null)
                            {
                                o.OnError(e.Error);
                            }
                            client.DownloadFileCompleted -= handler;
                            o.OnCompleted();
                        };
                        client.DownloadFileCompleted += handler;
                        client.DownloadProgressChanged += prgChangedHandler;
                    }
                    else
                    {
                        Console.WriteLine("Cancelling download of {0}",fileName);
                    }
                }
            );
            return cancellationTokenSource;
        }
        );
       return observable;
    }

    static void Main(string[] args)
    {

        var obs1 = CreateDownloadFileObservable("http://www.cnn.com", "cnn.htm");
        var obs2 = CreateDownloadFileObservable("http://www.bing.com", "bing.htm");

        var result = obs1.Concat(obs2);
        var subscription = result.Subscribe(a => Console.WriteLine("{0} -- {1}% complete ",a.UserState,a.ProgressPercentage), e=> Console.WriteLine(e.Message),()=> Console.WriteLine("Completed"));
        Console.ReadKey();
        subscription.Dispose();
        Console.WriteLine("Press a key to exit");
        Console.ReadKey();
}

您可以将 CreateDownloadFileObservable 转换为 WebClient 上的扩展方法，并WebClient client = new WebClient();在方法定义中删除，客户端将如下所示：

WebClient c = new WebClient();
c.CreateDownloadFileObservable ("www.bing.com","bing.htm");