如何将IObservable<Task<T>>转换为仅输出最新T的IObservable<T>

你的需求本质是只保留流中最新产生的任务的结果，丢弃所有未完成的旧任务结果，在System.Reactive（Rx）中，Switch操作符正好是为这类场景设计的，完全可以替代你手写的命令式ID标记逻辑。

实现代码 #

直接通过两步转换即可完成：

// 需引用System.Reactive.Linq命名空间
IObservable<T> streamB = streamA
    .Select(task => task.ToObservable()) // 将每个Task<T>转换为Observable<T>
    .Switch(); // 仅订阅最新的Observable，自动丢弃旧的未完成任务的结果

原理说明 #

• task.ToObservable()：把单个Task<T>转换为Observable<T>，当Task完成时，这个Observable会发射Task的结果然后完成。
• Switch()：这个操作符会持续监听上游的IObservable<IObservable<T>>流，每当有新的子Observable（对应新到达的Task）出现时，它会立即取消订阅之前的子Observable，只保留对最新子Observable的订阅。这完全匹配你给出的大理石图预期——当t2到达后，t1的后续完成结果会被忽略，仅t2的结果出现在streamB中。

对比原实现的优势 #

不需要手动维护递增ID、对比ID的逻辑，完全利用Rx内置的操作符实现，代码更简洁、可读性更强，同时Rx已经处理了线程安全和订阅管理的细节，避免了手动实现可能出现的bug。

如果仅使用TPL（Task Parallel Library），没有直接对应的内置API，因为TPL主要聚焦于任务的并行执行，而非流的动态切换与旧结果丢弃，因此Rx是解决这类问题的更优选择。

内容的提问来源于stack exchange，提问作者Alexander Høst