如何将任务列表转换为 Observable 并在完成后处理元素？

Question

给定一个集合Tasks：

var americanAirlines = new FlightPriceChecker("AA");

...

var runningTasks = new List<Task<IList<FlightPrice>>>

{

    americanAirlines.GetPricesAsync(from, to),

    delta.GetPricesAsync(from, to),

    united.GetPricesAsync(from, to)

};

我想GetPricesAsync()以它们到达的任何顺序处理结果。目前，我正在使用 while 循环来实现这一点：

while (runningTasks.Any())

{

    // Wait for any task to finish

    var completed = await Task.WhenAny(runningTasks);

    // Remove from running list   

    runningTasks.Remove(completed);

    // Process the completed task (updates a property we may be binding to)

    UpdateCheapestFlight(completed.Result);

}

这是一个可以使用 Rx 更优雅地解决的问题吗？我尝试使用类似下面的代码但卡住了，因为我必须在某个地方await每个getFlightPriceTask都会阻塞，然后才执行下一个，而不是执行第一个完成的然后等待下一个：

runningTasks

  .ToObservable()

  .Select(getFlightPriceTask => .???.)

Answer 1

尝试这个：runningTasks&nbsp; .Select(getFlightPriceTask => getFlightPriceTask.ToObservable())&nbsp; .Merge()&nbsp; .Subscribe(flightPrices => UpdateCheapestFlight(flightPrices))

Answer 2

（使用Merge()就是诀窍！），我想对此发表评论并提出替代解决方案。评论 Shlomo 的解决方案这个解决方案非常简单，表达了 Rx 的优雅。唯一的问题是不能等待完成。这在生产代码中通常不是问题，我们只关心更新然后绑定到 UI 的属性。我的另一个评论是计算是在Subscribe()- 有些人喜欢保持订阅超轻量级的，但我认为这主要是个人喜好。runningTasks&nbsp; // Get all tasks and turn them into Observables.&nbsp; .Select(getFlightPriceTask => getFlightPriceTask.ToObservable())&nbsp; // Merge all tasks (in my case 3) into one "lane". Think of cars trying&nbsp; // to leave a three lane highway and going for a one lane exit.&nbsp; .Merge()&nbsp; // For every task "leaving the highway" calculate the minimum price.&nbsp; .Subscribe(flightPrices => UpdateCheapestFlight(flightPrices))备选方案 1：使用 Do()这根本没有使用Subscribe()，这有点违背 Rx 的想法，但它可以等待，因此表现得像原始版本。await runningTasks&nbsp; &nbsp; .Select(getFlightPriceTask => getFlightPriceTask.ToObservable())&nbsp; &nbsp; .Merge()&nbsp; &nbsp; // Process result of each task.&nbsp; &nbsp; .Do(flightPrices => UpdateCheapestFlight(flightPrices))&nbsp; &nbsp; // Taking all elements will only complete if all three tasks have completed.&nbsp; &nbsp; .Take(runningTasks.Count);备选方案 2：消除 UpdateCheapestFlight()最后，我认为做这种更具 Rx 风格的方法是根本不使用原始的辅助方法，而是讲述一个易于阅读的“Rx-story”。var minFlightPrice = await runningTasks&nbsp; &nbsp; // Get all the tasks and turn them into Observables&nbsp;&nbsp; &nbsp; .Select(getFlightPriceTask => getFlightPriceTask.ToObservable())&nbsp; &nbsp; // Merge all three into one "lane".&nbsp; &nbsp; .Merge()&nbsp; &nbsp; // Get local minimum value of each airline&nbsp; &nbsp; .Select(x => x.Min())&nbsp; &nbsp; // Take all the local minimums...&nbsp; &nbsp; .Take(runningTasks.Count)&nbsp; &nbsp; // ...and find the minimum of them.&nbsp; &nbsp; .Min();

Answer 3

这是另一个解决方案：await runningTasks&nbsp; &nbsp; .ToObservable()&nbsp; &nbsp; .Merge()&nbsp; &nbsp; .Do(result => UpdateCheapestFlight(result))&nbsp; &nbsp; .DefaultIfEmpty();它看起来类似于 Shlomo 的解决方案，但有一个细微的区别：任务不是投影到嵌套的 observable ( IObservable<IObservable<TResult>>)，而是投影到任务的 observable ( IObservable<Task<TResult>>)。Rx 包含Merge对这两种结构都起作用的运算符的重载。后者稍微更有效，因为它避免了创建任务的一些一次性包装器。当我们从异步委托而不是已经物化的任务开始时，前者更强大，因为它允许控制并发级别（通过不一次启动所有任务），还因为它可以处理任何挂起任务的自动取消以防结果 observable 因任何原因（包括任何任务中发生的错误）在任何时间取消订阅。该Do运营商用于处理的任务的结果在他们的完成，一次一个结果的顺序。最后DefaultIfEmpty需要操作员，以防止InvalidOperationException在初始任务列表为空的情况下发生。这是因为等待生成的 observable，并且等待 observable 需要返回一个值（最后发出的值）。以下是上述示例中使用的 Rx 运算符的签名：// Converts an enumerable sequence to an observable sequence.public static IObservable<TSource> ToObservable<TSource>(&nbsp; &nbsp; this IEnumerable<TSource> source);// Merges results from all source tasks into a single observable sequence.public static IObservable<TSource> Merge<TSource>(&nbsp; &nbsp; this IObservable<Task<TSource>> sources);// Invokes an action for each element in the observable sequence, and propagates// all observer messages through the result sequence. This method can be used for// debugging, logging, etc. of query behavior by intercepting the message stream// to run arbitrary actions for messages on the pipeline.public static IObservable<TSource> Do<TSource>(this IObservable<TSource> source,&nbsp; &nbsp; Action<TSource> onNext);// Returns the elements of the specified sequence or the type parameter's default// value in a singleton sequence if the sequence is empty.public static IObservable<TSource> DefaultIfEmpty<TSource>(&nbsp; &nbsp; this IObservable<TSource> source);