Нерест много потоков. Кажется, что очередь в очереди на темы

Question

Я очень новичок в потоковом режиме, поэтому моя реализация, без сомнения, очень рудиментарна.

Я пытаюсь создать 8 или более потоков, каждый из которых загружает файл, используя FTPWebRequest, для параллельной работы.

Все нити кажутся порожденными и выполняют свою работу, но только две когда-либо работают в любой момент времени. Кажется, они попали в очередь.

Может кто-нибудь предложить, что я могу делать неправильно и как решить проблему?

private void btnDownload_Click(object sender, EventArgs e) 
    { 
     int itemCount = lBoxList.Items.Count; 
     string[] listOfFilesToDownload = new string[itemCount]; 

     for (int i = 0; i < itemCount; i++) 
     { 
      listOfFilesToDownload[i] = (string)lBoxList.Items[i]; 
     } 

     FtpParallelDownload("ftp://ftp.somedomain.com/sub/sub2/sub3/", (int)this.nudNumberOfThreads.Value, listOfFilesToDownload, tBoxDownloadPath.Text); 

    } 

public static void FtpParallelDownload(string serverUri, int maxNumberOfThreads, string[] listOfFilesToDownload, string downloadPath) 
    { 
     int progressPercent = 0; 

     // Validate number of threads requested 
     if (!(maxNumberOfThreads >= 1)) 
     { 
      ArgumentException e = new ArgumentException(); 
      throw e; 
     } 

     // Calc number of files based on array length 
     int numberOfFiles = listOfFilesToDownload.Length; 

     // Don't spawn more threads than files 
     if (maxNumberOfThreads > numberOfFiles) 
     { 
      maxNumberOfThreads = numberOfFiles; 
     }    

     // Thread spawning 

     List<Thread> threadPool = new List<Thread>(); 

     int runningThreadCount = 0; 



     for (int i = 0; i < numberOfFiles; i++) 
     { 
      if (runningThreadCount < maxNumberOfThreads) 
      { 
       Thread workerThread = new Thread(() => FtpDownloadFile(serverUri, listOfFilesToDownload[i], downloadPath)); 
       threadPool.Add(workerThread); 
       workerThread.Start(); 
       runningThreadCount++;     
      } 
      else 
      { 
       i--; 
      } 

      List<Thread> removeThreadList = new List<Thread>(); 
      foreach (Thread t in threadPool) 
      { 
       if (!t.IsAlive) 
       { 
        removeThreadList.Add(t); 
       } 
      } 

      foreach (Thread t in removeThreadList) 
      { 
       threadPool.Remove(t); 
       runningThreadCount--; 
      } 
      removeThreadList.Clear(); 
     } 

     MessageBox.Show("DOWNLOADS COMPLETE!"); 
    } 

    private static void FtpDownloadFile(string serverUri, string fileName, string downloadPath) 
    { 
     try 
     { 
      FtpWebRequest request = (FtpWebRequest)FtpWebRequest.Create(new Uri(serverUri + fileName)); 
      request.Method = WebRequestMethods.Ftp.DownloadFile; 
      request.Credentials = new NetworkCredential("anonymous", "noreply@mydomain.com"); 
      FtpWebResponse response = (FtpWebResponse)request.GetResponse(); 
      Stream responseStream = response.GetResponseStream(); 
      FileStream writeStream = new FileStream(downloadPath + "/" + fileName, FileMode.Create); 

      int Length = 4096; 
      Byte[] buffer = new Byte[Length]; 
      int bytesRead = responseStream.Read(buffer, 0, Length); 
      while (bytesRead > 0) 
      { 
       writeStream.Write(buffer, 0, bytesRead); 
       bytesRead = responseStream.Read(buffer, 0, Length); 
       Thread.Sleep(10); 

      } 
      writeStream.Close(); 
      response.Close(); 
      responseStream.Close(); 
     } 

     catch (WebException wEx) 
     { 
      MessageBox.Show(wEx.Message, "Download Error"); 
     } 
     catch (Exception ex) 
     { 
      MessageBox.Show(ex.Message, "Download Error"); 
     } 
    } 

threadPool является Thread<List>, что я использую, чтобы контролировать количество потоков, которые живы так, чтобы метать икру больше, когда одна законченная работа.

Answer 1

Вы действительно должны использовать существующую библиотеку, а не сворачивать свои собственные.

Вот как это можно сделать с помощью Reactive Framework от Microsoft (NuGet «Rx-WinForms»).

public static void FtpParallelDownload(
    string serverUri, int maxNumberOfThreads, 
    string[] listOfFilesToDownload, string downloadPath) 
{ 
    listOfFilesToDownload 
     .ToObservable() 
     .Select(x => Observable.Start(() => FtpDownloadFile(serverUri, x, downloadPath))) 
     .Merge(maxNumberOfThreads) 
     .ObserveOn(this) 
     .ToArray() 
     .Subscribe(xs => { },() => { MessageBox.Show("DOWNLOADS COMPLETE!"); }); 
} 

Выполнено.