C#并发编程经典实例（第2版）

互操作

｜

101

该异常传递到可观察流时，会被包装到

AggregateException

中。这与彼此关联的块传播错

误的方式类似。

使用数据流网格并把它当作可观察流的终点，这个过程会稍微复杂一点。下面的示例代码

调用了

AsObserver

，让数据流块能够订阅可观察流：

IObservable<DateTimeOffset> ticks =

Observable.Interval(TimeSpan.FromSeconds(1))

.Timestamp()

.Select(x => x.Timestamp)

.Take(5);

var

display =

new

ActionBlock<DateTimeOffset>(x => Trace.WriteLine(x));

ticks.Subscribe(display.AsObserver());

try

{

display.Completion.Wait();

Trace.WriteLine(

"

Done.");

}

catch

(Exception ex)

{

Trace.WriteLine(ex);

}

就像之前一样，可观察流的完成会转化成数据流块的完成，任何可观察流的错误都会转化

成数据流块的错误。

讨论

从概念角度来看，数据流块和可观察流存在诸多共通之处。它们都可以传递数据，都能理

解完成和错误。但是，它们是为不同场景设计的，

TPL

数据流适用于异步编程和并行编程

同时存在的情况，而

System.Reactive

适用于响应式编程。然而，这种概念上的交集让它们 ...