我只是想学习两者,以及如何一起使用它们.我明白,他们可以互相补充,我根本无法找到一个真正做到这一点的例子.
解决方法
让我从一点背景开始吧.
.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));
我认为这是你问题的一个相当基本的答案.是你期待的吗？
