控制SelectMany的最大线程数

Question

有没有一种方法可以用来设置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
    });

Answer 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);
});

另请参阅更高级的解决方案

Answer 2

SelectMany本身不会引入任何并发性，因此不会创建任何线程。是TPL做到了（因为您使用了Task.Run）。TPL通常在不创建太多线程来完成任务方面做得很好。如果您真的想限制线程的最大数量，请查看此处以及随后的此处。

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