为什么ParallelQuery<t>在转换为Observable时不起作用?

我有一个可观察的集合,我想并行处理,然后在过滤时观察处理的值,最后订阅一个接收过滤值的处理程序。

我的示例在语法上是正确的,编译也很好,当我运行代码时, Where 对执行筛选的语句进行了计算。但是订阅没有数据。如果我删除 AsParallel 以便在常规 IEnumerable ,数据通过,一切按预期工作。

这是我的示例,对字符串进行一些处理:

// Generate some data every second
var strings = Observable.Generate(() =>
    new TimeInterval<Notification<string>>(
        new Notification<string>
            .OnNext(DateTime.Now.ToString()), TimeSpan.FromSeconds(1)));

// Process the data in parallel
var parallelStrings = from value in strings.ToEnumerable().AsParallel()
                      select "Parallel " + value;

// Filter and observe
var data = String.Empty;
parallelStrings
    .Where(value => !String.IsNullOrEmpty(value))
    .ToObservable()
    .Subscribe(value => data = value);

下一个奇怪的事情是如果我使用 TakeWhile 操作符,在我看来,它在概念上类似于where,观察parallelquery的工作方式如预期的那样:

// Filter and observe
var data = String.Empty;
parallelStrings
    .TakeWhile(cs => !String.IsNullOrEmpty(cs))
    .ToObservable()
    .Subscribe(value => data = value);

向订阅中添加一些日志代码表明在 ToObservable 转换,但不在以下时间之后:

1.    var data = String.Empty;
2.    parallelStrings
3.        .Where(value => !String.IsNullOrEmpty(value))
4.        .Select(value => value)
5.        .ToObservable()
6.        .Select(value => value)
7.        .Subscribe(value => data = value);

第4行lambda中的断点被命中,而第6行lambda中的断点从未被命中。

为什么会 取而代之 使数据通过订阅服务器 在哪里? 不是吗?

如果这很重要,我将在Visual Studio 2010 RC中使用面向.NET 4.0框架客户端配置文件的项目开发代码。

更新 基于 @Sergeys answer 我修改了 在哪里? 过滤器。以下代码按预期工作:

var processedStrings = from value in strings
                       let processedValue = "Parallel " + value
                       where !String.IsNullOrEmpty(processedValue)
                       select processedValue;

var data = String.Empty;
processedStrings
    .ToEnumerable()
    .AsParallel()
    .ToObservable()
    .Subscribe(value => data = value );

首先转换最初的可观测数据还是有点尴尬 processedStrings 变成一个可枚举的,以便将其并行化,然后将其转换回一个可观测的,以便订阅最终的结果。