Создайте многопоточные приложения для запуска нескольких запросов в C#

Question

Я пытаюсь создать инструмент Windows Forms, который выполняет запросы асинхронно. Приложение имеет datagridview с 30 возможными запросами для запуска. Пользователь проверяет запросы, которые он хочет выполнить, например, 10 запросов и нажатие кнопки. Приложение имеет переменную maxthreads = 3 (для обсуждения), которая указывает, сколько потоков можно использовать для асинхронного запуска запросов. Запросы выполняются в рабочей среде, и мы не хотим перегружать систему слишком большим количеством потоков, запущенных в одно и то же время. Каждый запрос выполняется в среднем 30 секунд. (около 5 минут, другие 2 сек.) В datagridview есть столбец изображений, содержащий значок, который отображает состояние каждого запроса (0- Доступно для запуска, 1-Выбран для запуска, 2- Запуск, 3- Успешно завершено, -1 Ошибка) Мне нужно иметь возможность общаться с пользовательским интерфейсом каждый раз, когда запрос начинается и заканчивается. Как только запрос заканчивается, результаты отображаются в представлении datagridview, содержащемся в Tabcontrol (одна вкладка для каждого запроса).

Подход: Я думал создать несколько рабочих фонарей maxthread и позволить им запускать запросы. По завершении работы фонового работ он взаимодействует с пользовательским интерфейсом и назначается новому запросу и так далее, пока не будут выполнены все запросы.

Я пробовал использовать jobWorker, который отправил бы работу работникам фона, но не знает, как ждать завершения всех потоков. После завершения bgw он сообщает о прогрессе в событии RunWorkerCompleted в jobWorker, но тот уже завершен.

В потоке пользовательского интерфейса я называю работник присваивания со всеми запросами, которые необходимо выполнить:

private void btnRunQueries_Click(object sender, EventArgs e) 
    { 
     if (AnyQueriesSelected()) 
     { 
      tcResult.TabPages.Clear(); 

      foreach (DataGridViewRow dgr in dgvQueries.Rows) 
      { 
       if (Convert.ToBoolean(dgr.Cells["chk"].Value)) 
       { 
        Query q = new Query(dgr.Cells["ID"].Value.ToString(), 
         dgr.Cells["Name"].Value.ToString(), 
         dgr.Cells["FileName"].Value.ToString(), 
         dgr.Cells["ShortDescription"].Value.ToString(), 
         dgr.Cells["LongDescription"].Value.ToString(), 
         dgr.Cells["Level"].Value.ToString(), 
         dgr.Cells["Task"].Value.ToString(), 
         dgr.Cells["Importance"].Value.ToString(), 
         dgr.Cells["SkillSet"].Value.ToString(), 
         false, 
         new Dictionary<string, string>() 
         { { "#ClntNb#", txtClntNum.Text }, { "#Staff#", "100300" } }); 

        qryList.Add(q); 
       } 
      } 
      assignmentWorker.RunWorkerAsync(qryList); 
     } 
     else 
     { 
      MessageBox.Show("Please select at least one query.", 
          "Warning", 
          MessageBoxButtons.OK, 
          MessageBoxIcon.Information); 
     } 
    } 

Вот AssignmentWorker:

private void assignmentWorker_DoWork(object sender, DoWorkEventArgs e) 
    { 
     foreach (Query q in (List<Query>)e.Argument) 
     { 
      while (!q.Processed) 
      { 
       for (int threadNum = 0; threadNum < maxThreads; threadNum++) 
       { 
        if (!threadArray[threadNum].IsBusy) 
        { 
         threadArray[threadNum].RunWorkerAsync(q); 
         q.Processed = true; 
         assignmentWorker.ReportProgress(1, q); 
         break; 
        } 
       } 

       //If all threads are being used, sleep awhile before checking again 
       if (!q.Processed) 
       { 
        Thread.Sleep(500); 
       } 
      } 
     } 
    } 

Все BGW запустить то же событие:

private void backgroundWorkerFiles_DoWork(object sender, DoWorkEventArgs e) 
    { 
     try 
     { 
      Query qry = (Query)e.Argument; 

      DataTable dtNew = DataAccess.RunQuery(qry).dtResult; 

      if (dsQryResults.Tables.Contains(dtNew.TableName)) 
      { 
       dsQryResults.Tables.Remove(dtNew.TableName); 
      } 

      dsQryResults.Tables.Add(dtNew); 

      e.Result = qry; 
     } 
     catch (Exception ex) 
     { 

     } 
    } 

После того, как запрос вернулся и DataTable был добавлен в набор данных:

private void backgroundWorkerFiles_RunWorkerCompleted(object sender, 
                RunWorkerCompletedEventArgs e) 
    { 
     try 
     { 
      if (e.Error != null) 
      { 
       assignmentWorker.ReportProgress(-1, e.Result); 
      } 
      else 
      { 
       assignmentWorker.ReportProgress(2, e.Result); 
      } 
     } 
     catch (Exception ex) 
     { 
      int o = 0; 
     } 
    } 

У меня есть проблема в том, что работник присваивание заканчивается до финиша BGW и призыв к assignmentWorker.ReportProgress идти в ад (простите мой французский). Как я могу дождаться завершения всех запущенных bgw до завершения работника задания?

Спасибо!

Answer 1

Как указано в the comment above, у вас слишком сложный дизайн. Если у вас есть определенное максимальное количество задач (запросов), которые должны выполняться одновременно, вы можете и должны просто создать это число рабочих и заставить их использовать задачи из своей очереди (или списка) задач до тех пор, пока эта очередь не будет пуста.

Отсутствие хорошего Minimal, Complete, and Verifiable code example, который кратко и четко иллюстрирует ваш конкретный сценарий, не представляется возможным предоставить код, который будет непосредственно решать ваш вопрос. Но вот пример использования List<T> как исходный код делает, который будет работать, как я описал выше:

using System; 
using System.Collections.Generic; 
using System.Threading.Tasks; 

namespace TestSO42101517WaitAsyncTasks 
{ 
    class Program 
    { 
     static void Main(string[] args) 
     { 
      Random random = new Random(); 
      int maxTasks = 30, 
       maxActive = 3, 
       maxDelayMs = 1000, 
       currentDelay = -1; 
      List<TimeSpan> taskDelays = new List<TimeSpan>(maxTasks); 

      for (int i = 0; i < maxTasks; i++) 
      { 
       taskDelays.Add(TimeSpan.FromMilliseconds(random.Next(maxDelayMs))); 
      } 

      Task[] tasks = new Task[maxActive]; 
      object o = new object(); 

      for (int i = 0; i < maxActive; i++) 
      { 
       int workerIndex = i; 

       tasks[i] = Task.Run(() => 
       { 
        DelayConsumer(ref currentDelay, taskDelays, o, workerIndex); 
       }); 
      } 

      Console.WriteLine("Waiting for consumer tasks"); 

      Task.WaitAll(tasks); 

      Console.WriteLine("All consumer tasks completed"); 
     } 

     private static void DelayConsumer(ref int currentDelay, List<TimeSpan> taskDelays, object o, int workerIndex) 
     { 
      Console.WriteLine($"worker #{workerIndex} starting"); 

      while (true) 
      { 
       TimeSpan delay;  
       int delayIndex; 

       lock (o) 
       { 
        delayIndex = ++currentDelay; 
        if (delayIndex < taskDelays.Count) 
        { 
         delay = taskDelays[delayIndex]; 
        } 
        else 
        { 
         Console.WriteLine($"worker #{workerIndex} exiting"); 
         return; 
        } 
       } 

       Console.WriteLine($"worker #{workerIndex} sleeping for {delay.TotalMilliseconds} ms, task #{delayIndex}"); 
       System.Threading.Thread.Sleep(delay); 
      } 
     } 
    } 
} 

В вашем случае, каждый работник будет сообщать о прогрессе в какой-то глобальное состояние. Вы не показываете обработчик ReportProgress для своего рабочего задания, поэтому я не могу сказать конкретно, как это будет выглядеть.Но предположительно это должно было бы передать либо -1, либо 2 в какой-то метод, который знает, что делать с этими значениями (то есть то, что в противном случае было бы вашим обработчиком ReportProgress).

Обратите внимание, что код может быть несколько упрощен, особенно если отдельные задачи потребляются, если вы используете фактическую структуру данных очереди для задач. Этот подход будет выглядеть примерно так:

using System; 
using System.Collections.Concurrent; 
using System.Threading.Tasks; 

namespace TestSO42101517WaitAsyncTasks 
{ 
    class Program 
    { 
     static void Main(string[] args) 
     { 
      Random random = new Random(); 
      int maxTasks = 30, 
       maxActive = 3, 
       maxDelayMs = 1000, 
       currentDelay = -1; 
      ConcurrentQueue<TimeSpan> taskDelays = new ConcurrentQueue<TimeSpan>(); 

      for (int i = 0; i < maxTasks; i++) 
      { 
       taskDelays.Enqueue(TimeSpan.FromMilliseconds(random.Next(maxDelayMs))); 
      } 

      Task[] tasks = new Task[maxActive]; 

      for (int i = 0; i < maxActive; i++) 
      { 
       int workerIndex = i; 

       tasks[i] = Task.Run(() => 
       { 
        DelayConsumer(ref currentDelay, taskDelays, workerIndex); 
       }); 
      } 

      Console.WriteLine("Waiting for consumer tasks"); 

      Task.WaitAll(tasks); 

      Console.WriteLine("All consumer tasks completed"); 
     } 

     private static void DelayConsumer(ref int currentDelayIndex, ConcurrentQueue<TimeSpan> taskDelays, int workerIndex) 
     { 
      Console.WriteLine($"worker #{workerIndex} starting"); 

      while (true) 
      { 
       TimeSpan delay; 

       if (!taskDelays.TryDequeue(out delay)) 
       { 
        Console.WriteLine($"worker #{workerIndex} exiting"); 
        return; 
       } 

       int delayIndex = System.Threading.Interlocked.Increment(ref currentDelayIndex); 

       Console.WriteLine($"worker #{workerIndex} sleeping for {delay.TotalMilliseconds} ms, task #{delayIndex}"); 
       System.Threading.Thread.Sleep(delay); 
      } 
     } 
    } 
}