使用反应式框架观察threadsafe集合

Question

这将归结为一个相当高层次的问题。我有一个很大的数据集，需要(独立地)验证每一行。我想使用Parallel.Foreach，并在该行上调用一个验证方法。假设此方法为threadsafe。如果验证返回一个错误，我需要用这个错误更新数据行。我显然不能从后台线程来做这件事。然而，我不确定实现这种错误处理的最佳方式是什么。我的想法是将行in和错误存储在BlockingCollection中，因为这是用于写入的线程安全。然后，我将不断轮询(从后台线程)，当后台线程找到数据时，调用表单并更新当前行。

我想知道是否有更简单的方法来做这件事，使用反应式框架？基本上，我需要一个多生产者threadsafe集合，当一个新的值被添加到集合中时，它可以被“观察”，'OnNext‘将在主线程上执行-这是可能的吗？理想情况下，我还可以控制这种情况发生的频率(比如每2-3秒一次，这样主线程上的回调就会每2-3秒更新一次多行)，所以我不会经常调用主线程。

耽误您时间，实在对不起。

Answer 1

这是如何打动你的想象的：

IObservable<bool> ValidateAsync(Row item)
{
    return Observable.Start(() => {
        // TODO: Figure out if the row is valid
        return true;
    }, Scheduler.TaskPoolScheduler);
}

myBigDataTable.ToObservable()
    .Select(x => ValidateAsync(x).Select(y => new { Row = x, IsValid = y }))
    .Merge(10 /* rows concurrently */)
    .ObserveOn(SynchronizationContext.Current /*assuming WinForms */)
    .Subscribe(x => {
        Console.WriteLine("Row {0} validity: {1}", x.Row, x.IsValid);
    });

没有锁，没有愚蠢的容器，没有阻塞，100%线程安全。