限制并发产生和等待工作进程

Question

我有一个 C# 程序，它将一组工作项排队。对于每个工作项，我生成另一个应用程序（我无法更改）来处理该项目。此过程是相对 CPU 密集型的。我想限制此应用程序的实例数。

我考虑过使用 PLINQ：

Parallel.ForEach(
    workItems,
    new ParallelOptions { MaxDegreeOfParallelism = 4 },
    x => Process.Start("worker", x).WaitForExit());

但我担心的是，每个并行工作人员只会用完一个线程来等待相应的进程。

我还看到 PLINQ 尝试对返回的项目进行批处理，这意味着它可能会在等待适当大小的批处理时停止。那么，也许使用BlockingCollection 的单生产者/多消费者模式会起作用？这样做的问题是每个并行工作线程都有一个线程，这会（？）比 PLINQ 解决方案更糟糕。

鉴于上述内容已简化，并且我实际上有一个 TaskCompletionSource 连接到每个工作进程（通过 Exited 事件），我可以使用 TPL 中的某些东西来执行此操作而不阻塞任何后台线程吗？

Answer 1

如果您可以使用Process.Exited 事件将Process 开始和结束包装在Task 中：

Task WrapExternalProcess( WorkItem workItem ) { ... }

您可以使用延续来完全消除阻塞。这样的事情会做：

Task DoAllWork( IEnumerable<WorkItem> workItems )
{
  int THREAD_COUNT = 4;

  var bag = new ConcurrentBag<WorkItem>( workItems );
  var ce = new CountdownEvent( THREAD_COUNT );
  var tcs = new TaskCompletionSource<bool>();

  for ( int i = 0 ; i < THREAD_COUNT ; i++ ) Work( bag, ce, tcs );

  return tcs.Task;
}

void Work(
  ConcurrentBag<WorkItem> bag,
  CountdownEvent ce,
  TaskCompletionSource<bool> tcs
)
{
    WorkItem workItem;
    if ( bag.TryTake( out workItem) )
    {
        WrapExternalProcess( workItem )
          .ContinueWith( t => Work( bag, ce, tcs ) );
    }
    else // no more work
    {
        // If I'm the last thread to finish
        if ( ce.Signal() ) tcs.SetResult( true );
    }
}

Answer 2

我已经实现了以下内容：

var sem = new SemaphoreSlim(4);
foreach (var item in workItems)
{
    sem.Wait();

    ProcessAsync(item).ContinueWith(_ => sem.Release());
}

这给了我一个单一的线程，可以同时保持有限数量的后台进程在运行。