我怎样才能连续QueueUserWorkItems，但一次不排队呢？

Question

我正在为一个网站和一个不同的问题，我已经决定使用带有QueueUserWorkItem（）的ThreadPool的多线程的刮板。

我该如何持续排队工作项目而不一次排队呢？我需要排队> 300k项目（每个用户ID一个），如果我循环排队它们，我将耗尽内存。

所以，我想的是：

// 1 = startUserID, 300000 = endUserID, 25 = MaxThreads 
Scraper webScraper = new Scraper(1, 300000, 25); 

webScraper.Start(); 
// return immediately while webScraper runs in the background 

在此期间，webScraper是continuouslly将所有30万个工作项的线程变得可用。

这是我到目前为止有：

public class Scraper 
    { 
     private int MaxUserID { get; set; } 
     private int MaxThreads { get; set; } 
     private static int CurrentUserID { get; set; } 
     private bool Running { get; set; } 
     private Parser StatsParser = new Parser(); 


     public Scraper() 
      : this(0, Int32.MaxValue, 25) 
     { 
     } 

     public Scraper(int CurrentUserID, int MaxUserID, int MaxThreads) 
     { 
      this.CurrentUserID = CurrentUserID; 
      this.MaxUserID = MaxUserID; 
      this.MaxThreads = MaxThreads; 
      this.Running = false; 

      ThreadPool.SetMaxThreads(MaxThreads, MaxThreads); 
     } 

     public void Start() 
     { 
      int availableThreads; 

      // Need to start a new thread to spawn the new WorkItems so Start() will return right away? 
      while (Running) 
      { 

       // if (!CurrentUserID >= MaxUserID) 
       // { 
       //  while (availableThreads > 0) 
       //  { 
       //   ThreadPool.QueueUserWorkItem(new WaitCallBack(Process)); 
       //  } 
       // } 
       // else 
       // { Running = false; } 
      } 
     } 

     public void Stop() 
     { 
      Running = false; 
     } 

     public static void process(object state) 
     { 
      var userID = Interlocked.Increment(ref CurrentUserID); 
      ... Fetch Stats for userID 
     } 
    } 

这是正确的做法？

任何人都可以指向正确的方向来处理创建工作项目，而在后台调用Start（）时，并且不会一次创建所有工作项目？

Answer 1

这会更好地实施更少的工作项目从工作队列中窃取工作？仅仅因为你有30万件工作来做这件事并不意味着你需要30万名工作者才能做到这一点。显然，由于您只有几个核心，所以这些工作中只有少数能够同时进行，所以为什么不把大量工作交给更少的工作人员呢？

根据每项工作所需时间的持续时间，您可以将其均匀分配给每个工作人员，或者拥有一个中央队列（您必须锁定），并且每个工作人员可以抓取一些工作因为它耗尽。

编辑：

乔·达菲似乎有一系列有关编写工作窃取队列的位置：http://www.bluebytesoftware.com/blog/2008/08/12/BuildingACustomThreadPoolSeriesPart2AWorkStealingQueue.aspx。它看起来也像.Net 4的Threadpool会更聪明一些。但我认为你不需要这种情况下特别复杂的东西。

Answer 2

我认为创建队列中的项目似乎并不完全正确，因此如何让WorkItem在完成后再次将自己排队？

你的启动方法可以排队，比如MaxThreads项目的3倍（在你的例子中是75），然后你的Process方法在完成时排队。这样，你的启动方法迅速恢复，但触发关闭了一些工作项目，这就像我说的那么火自己：

    public class Scraper 
    { 
     private int MaxUserID { get; set; } 
     private int MaxThreads { get; set; } 
     private int currentUserID; 
     private bool Running { get; set; } 
     private Parser StatsParser = new Parser(); 

     private int Multiplier { get; set; } 

     public Scraper() 
      : this(0, Int32.MaxValue, 25) 
     { 
     } 

     public Scraper(int currentUserID, int maxUserID, int maxThreads) 
     { 
      this.currentUserID = currentUserID; 
      this.MaxUserID = maxUserID; 
      this.MaxThreads = maxThreads; 
      this.Running = false; 

      ThreadPool.SetMaxThreads(maxThreads, maxThreads); 
      Multiplier = 3; 
     } 

     public void Start() 
     { 
      Running = true; 
      for (int i = 0; i < MaxThreads * Multiplier; i++) 
      { 
       ThreadPool.QueueUserWorkItem(Process); 
      } 
     } 

     public void Stop() 
     { 
      Running = false; 
     } 

     public void Process(object state) 
     { 
      if (Running == false) 
      { 
       return; 
      } 
      if (currentUserID < MaxUserID) 
      { 
       Interlocked.Increment(ref currentUserID); 
       //Parse stats for currentUserID 
       ThreadPool.QueueUserWorkItem(Process); 
      } 
      else 
      { Running = false; } 
     } 
    } 

我敢肯定，正在运行的标志应采用互锁安全进行设定。我已经把乘数变成了一个属性，可以传递给构造函数 - 我相当肯定可以调整它来调整性能，这取决于这些统计信息需要解析多长时间。

Answer 3

我绝对不会使用ThreadPool.SetMaxThreads - 请记住，线程池在所有进程之间共享 - 设置线程的最大数量将简单地杀死性能。线程池背后的全部想法是，您不需要指定最大线程数量等事情 - .Net框架计算出要分配的最佳线程数量 - 您无需这样做。

请注意，排队300 000个项目不会导致300 000个线程产生 - ThreadPool类将为您管理线程数并根据需要重新使用线程。如果你只是担心会消耗太多的资源，我会建议你优化你的过程 - 也许创建一个'Spawner'类，然后运行1000个刮板实例？

Answer 4

看起来你需要一个主控过程控制类来管理正在触发并保持队列已满的工作人员的数量。

您可以用两个队列，然后工作：

1. 拥抱你需要刮
2. 其次要做的工作

这个硕士/调速对象，然后将保持所有项目一个循环，直到队列＃1中的所有项目都消失，并且在有可用循环时它将继续添加到队列＃2。

Answer 5

您可以使用不同的线程池。这里是一个：http://www.codeplex.com/smartthreadpool 它允许您一次排队所有的项目。您可以分配最大数量的线程来创建。假设你有1000个工作项目，并且你分配了100个线程。它会立即采取前100件物品，让其他人等待。只要其中一个项目完成并且一个线程释放，下一个排队的项目就会启动。它管理所有的工作，但不会使线程和内存饱和。此外，它不使用.net线程池中的线程。