控制 SelectMany 的最大线程数答案

【问题标题】：Controlling the max thread count of SelectMany控制 SelectMany 的最大线程数
【发布时间】：2015-06-26 19:30:42
【问题描述】：

有没有一种方法可以用来设置 IObservable.SelectMany 的最大线程数？

以下代码非常适合在处理项目时更新 UI，但目前我尝试执行的任务有点占用资源。我想将最大线程数设置为两个，以减少资源使用量。

AsyncCommand = ReactiveCommand.CreateAsyncObservable(_ => 
{
    // Set progress bar indicator to 0

    var set = new [] {...} // The set of items to process

    // Set the progress bar indicator max to the count of the items to process

    return set
        .ToObservable()
        .SelectMany((item, index) => Task.Run(() =>
        {
            // Process item

            return item;
        }), (item, index, processed) => item); 
});

AsyncCommand
    .ObserveOn(RxApp.MainThreadScheduler)
    .Subscribe(item => 
    {
        // Step the progress bar indicator
    });

【问题讨论】：

标签： c# system.reactive reactiveui

【解决方案1】：

Merge 有一个最大并行度参数：

AsyncCommand = ReactiveCommand.CreateAsyncObservable(_ => 
{
    // Set progress bar indicator to 0

    var set = new [] {...} // The set of items to process

    // Set the progress bar indicator max to the count of the items to process

    return set
        .ToObservable()
        .Select(item => Observable.FromAsync(() => DoAsyncProcess(item))))
        .Merge(2);
});

另见more advanced solution

【讨论】：

【解决方案2】：

SelectMany，在您的代码 sn-p 中使用，它本身不会引入任何并发，因此不会创建任何线程。它是 TPL（因为您使用 Task.Run）。 TPL 通常在不创建太多线程来完成工作方面做得很好。如果您真的想限制最大线程数，请查看 here 和随后的 here。

作为一个简单的替代方案，使用 Stephen Cleary 的出色 AsyncEx 包中的 AsyncSemaphore，并将处理代码放在 WaitAsync 和 Release 调用之间。

【讨论】：