带插槽的 C# 多线程

Question

我有这个检查代理服务器的功能，目前它只检查一些线程并等待所有线程完成，直到下一组开始。是否可以在一个线程从允许的最大值完成后立即启动一个新线程？

for (int i = 0; i < listProxies.Count(); i+=nThreadsNum)
{                              
    for (nCurrentThread = 0; nCurrentThread < nThreadsNum; nCurrentThread++)
    {
        if (nCurrentThread < nThreadsNum)
        {
           string strProxyIP = listProxies[i + nCurrentThread].sIPAddress;
           int nPort = listProxies[i + nCurrentThread].nPort;
                    tasks.Add(Task.Factory.StartNew<ProxyAddress>(() => CheckProxyServer(strProxyIP, nPort, nCurrentThread)));
        }
     }                

     Task.WaitAll(tasks.ToArray());

     foreach (var tsk in tasks)
     {
        ProxyAddress result = tsk.Result;
        UpdateProxyDBRecord(result.sIPAddress, result.bOnlineStatus);
     }

     tasks.Clear();                
}

Answer 1

这看起来要简单得多：

int numberProcessed = 0;
Parallel.ForEach(listProxies,
  new ParallelOptions { MaxDegreeOfParallelism = nThreadsNum },
  (p)=> {
    var result = CheckProxyServer(p.sIPAddress, s.nPort, Thread.CurrentThread.ManagedThreadId);
    UpdateProxyDBRecord(result.sIPAddress, result.bOnlineStatus);
    Interlocked.Increment(numberProcessed);
});

带插槽：

var obj = new Object();
var slots = new List<int>();
Parallel.ForEach(listProxies,
  new ParallelOptions { MaxDegreeOfParallelism = nThreadsNum },
  (p)=> {
    int threadId = Thread.CurrentThread.ManagedThreadId;
    int slot = slots.IndexOf(threadId);
    if (slot == -1)
    {
      lock(obj)
      {
        slots.Add(threadId);
      }
      slot = slots.IndexOf(threadId);
    }
    var result = CheckProxyServer(p.sIPAddress, s.nPort, slot);
    UpdateProxyDBRecord(result.sIPAddress, result.bOnlineStatus);
});

我在那里采取了一些捷径来保证线程安全。您不必进行正常的检查-锁定-检查舞蹈，因为永远不会有两个线程尝试将相同的 threadid 添加到列表中，因此第二次检查将始终失败并且不需要。其次，出于同样的原因，我认为您也不需要锁定外部 IndexOf。这使得这是一个非常高效的并发例程，无论可枚举项中有多少项，它都很少锁定（它应该只锁定 nThreadsNum 次）。

Answer 2

另一种解决方案是使用SemaphoreSlim 或使用BlockinCollection<T> 的生产者-消费者模式。两种解决方案都支持取消。

SemaphoreSlim

private async Task CheckProxyServerAsync(IEnumerable<object> proxies)
{
  var tasks = new List<Task>();
  int currentThreadNumber = 0;
  int maxNumberOfThreads = 8;

  using (semaphore = new SemaphoreSlim(maxNumberOfThreads, maxNumberOfThreads))
  {
    foreach (var proxy in proxies)
    {
      // Asynchronously wait until thread is available if thread limit reached
      await semaphore.WaitAsync();

      string proxyIP = proxy.IPAddress;
      int port = proxy.Port;
      tasks.Add(Task.Run(() => CheckProxyServer(proxyIP, port, Interlocked.Increment(ref currentThreadNumber)))
        .ContinueWith(
          (task) =>
          {
            ProxyAddress result = task.Result;

            // Method call must be thread-safe!
            UpdateProxyDbRecord(result.IPAddress, result.OnlineStatus);

            Interlocked.Decrement(ref currentThreadNumber);

            // Allow to start next thread if thread limit was reached
            semaphore.Release();
          },
          TaskContinuationOptions.OnlyOnRanToCompletion));
    }

    // Asynchronously wait until all tasks are completed
    // to prevent premature disposal of semaphore
    await Task.WhenAll(tasks);
  }
}

---

生产者-消费者模式

// Uses a fixed number of same threads
private async Task CheckProxyServerAsync(IEnumerable<ProxyInfo> proxies)
{
  var pipe = new BlockingCollection<ProxyInfo>();
  int maxNumberOfThreads = 8;
  var tasks = new List<Task>();

  // Create all threads (count == maxNumberOfThreads)
  for (int currentThreadNumber = 0; currentThreadNumber < maxNumberOfThreads; currentThreadNumber++)
  {
    tasks.Add(
      Task.Run(() => ConsumeProxyInfo(pipe, currentThreadNumber)));
  }

  proxies.ToList().ForEach(pipe.Add);
  pipe.CompleteAdding();

  await Task.WhenAll(tasks);
}

private void ConsumeProxyInfo(BlockingCollection<ProxyInfo> proxiesPipe, int currentThreadNumber)
{
  while (!proxiesPipe.IsCompleted)
  {
    if (proxiesPipe.TryTake(out ProxyInfo proxy))
    {
      int port = proxy.Port;
      string proxyIP = proxy.IPAddress;
      ProxyAddress result = CheckProxyServer(proxyIP, port, currentThreadNumber); 

      // Method call must be thread-safe!
      UpdateProxyDbRecord(result.IPAddress, result.OnlineStatus);
    }
  }
}

Answer 3

如果我正确理解了您的问题，那么使用await Task.WhenAny 做这件事实际上相当简单。基本上，您保留所有正在运行的任务的集合。一旦达到一定数量的正在运行的任务，您就等待一个或多个任务完成，然后从您的集合中删除已完成的任务并继续添加更多任务。

下面是我的意思的一个例子：

        var tasks = new List<Task>();

        for (int i = 0; i < 20; i++)
        {
            // I want my list of tasks to contain at most 5 tasks at once
            if (tasks.Count == 5)
            {
                // Wait for at least one of the tasks to complete
                await Task.WhenAny(tasks.ToArray());

                // Remove all of the completed tasks from the list
                tasks = tasks.Where(t => !t.IsCompleted).ToList();
            }

            // Add some task to the list
            tasks.Add(Task.Factory.StartNew(async delegate ()
                {
                    await Task.Delay(1000);
                }));
        }

Answer 4

我建议稍微改变你的方法。不要启动和停止线程，而是将您的代理服务器数据放入一个并发队列中，每个代理服务器一个项目。然后创建一个固定数量的线程（或异步任务）来处理队列。在我看来，这更有可能提供流畅的性能（您不会一遍又一遍地启动和停止线程，这会产生开销）并且更容易编码。

一个简单的例子：

class ProxyChecker
{
    private ConcurrentQueue<ProxyInfo> _masterQueue = new ConcurrentQueue<ProxyInfo>();

    public ProxyChecker(IEnumerable<ProxyInfo> listProxies)
    {
        foreach (var proxy in listProxies)
        {
            _masterQueue.Enqueue(proxy);
        }
    }

    public async Task RunChecks(int maximumConcurrency)
    {
        var count = Math.Max(maximumConcurrency, _masterQueue.Count);
        var tasks = Enumerable.Range(0, count).Select( i => WorkerTask() ).ToList();
        await Task.WhenAll(tasks);
    }

    private async Task WorkerTask()
    {
        ProxyInfo proxyInfo;
        while ( _masterList.TryDequeue(out proxyInfo))
        {
            DoTheTest(proxyInfo.IP, proxyInfo.Port)
        }
    }
}