.net – TPL数据流和Rx组合示例[已关闭]

.NET框架具有许多特殊类型 – 适当的类或接口 – 对于向…提供特殊功能的任何特定类型或接口 – Task< T>,IObservable< T>,Nullable< T>,IEnumerable< T>,Lazy&底层类型T.

TPL使用任务< T>代表单个T值的异步计算

Rx使用IObservable< T>代表T的零个或多个值的异步计算.

这两者的“异步计算”方面将TPL和Rx组合在一起.

现在,TPL还使用类型Task来表示Action lambda的异步执行,但是这可以被认为是Task的特殊情况.其中T无效.非常像c#中的标准方法返回void像这样：

public void MyMethod() { }

Rx也允许使用称为单元的特殊类型使用相同的特殊情况.

TPL和Rx之间的差异是返回值的数量. TPL是唯一的,而Rx是零或更多.

所以,如果你以特殊的方式处理Rx,只能使用返回单个值的可观察序列,就可以像TPL那样进行一些计算.

例如,在TPL中,我可以写：

Task.Factory
    .StartNew(() => "Hello")
    .ContinueWith(t => Console.WriteLine(t.Result));

而在Rx中,相当于：

Observable
    .Start(() => "Hello")
    .Subscribe(x => Console.WriteLine(x));

我可以在Rx中进一步说明,TPL应该用于执行如下计算：

Observable
    .Start(() => "Hello",Scheduler.TaskPool)
    .Subscribe(x => Console.WriteLine(x));

(默认使用线程池.)

现在我可以做一些“混合配对”.如果我添加对System.Reactive.Threading.Tasks命名空间的引用,我可以很容易地在任务和可观察器之间移动.

Task.Factory
    .StartNew(() => "Hello")
    .ToObservable()
    .Subscribe(x => Console.WriteLine(x));

Observable
    .Start(() => "Hello")
    .ToTask()
    .ContinueWith(t => Console.WriteLine(t.Result));

注意ToObservable()& .ToTask()调用,并且从一个库到另一个库的转换.

如果我有一个observable返回多个值,我可以使用observable .ToArray()扩展方法将多个序列值转换为可以转换为任务的单个数组值.像这样：

Observable
    .Interval(TimeSpan.FromSeconds(1.0))
    .Take(5) // is IObservable<long>
    .ToArray()
    .ToTask() // is Task<long[]>
    .ContinueWith(t => Console.WriteLine(t.Result.Length));

我认为这是你问题的一个相当基本的答案.是你期待的吗？