Использование параллельной библиотеки задач

Question

Я только начинаю писать и использовать TPL в .NET 4.5 и задавался вопросом, можете ли вы мне помочь.

В принципе, моя программа WPF выполняет тяжелую работу над набором записей базы данных, которые могут находиться в диапазоне от 5 до 10 миллионов. Я хочу использовать библиотеку TPL для эффективного управления этой работой, сохраняя при этом пользовательский интерфейс.

Мой сценарий выглядит следующим образом. Я планировал использовать 3 отдельные задачи для извлечения данных, скажем, по 3 миллиона каждый, и передать их моему «работнику».

И я хочу отдельную задачу, которая берет каждый элемент из ВСЕХ элементов из вышеуказанных задач и выполняет некоторую работу с ним.

Итак, я хочу глобальный список элементов, в которые все 3 задачи могут вставлять элементы, моя четвертая задача должна затем выбирать элементы из этого главного списка по одному и выполнять некоторую работу с ним, он должен дождаться, если список пуст для его повторного заполнения и т. д.

Не могли бы вы дать некоторые рекомендации относительно того, как это сделать с помощью TPL? Я новичок с TPL. Некоторые примеры кода были бы хорошими.

Answer 1

Рассмотрите возможность хранения ваших товаров с каждого Task в глобальном, потокобезопасном объекте списка, например ConcurrentStack или ConcurrentQueue.

Для получения дополнительной информации см: http://msdn.microsoft.com/en-us/library/system.collections.concurrent(v=vs.110).aspx

Answer 2

Если я понимаю описание проблемы правильно, у вас есть производитель/потребитель сценарий, в котором есть три производителей, каждый из которых извлечения независимых записей из базы данных, и один потребитель.

Если это так, то мне кажется, что вы хотите начать с BlockingCollection<T>, где T - тип объекта, который производители извлекают из базы данных. Основная используемая коллекция зависит от того, как вы хотите обрабатывать данные. Наиболее распространенными сценариями являются FIFO и неупорядоченные, представленные ConcurrentQueue<T> и ConcurrentBag<T>, соответственно.

Вот простой пример кода, который демонстрирует, как это будет работать:

BlockingCollection<int> consumeFrom = new BlockingCollection<int>(); 
int producerCount = 3; 

for (int i = 0; i < producerCount; i++) 
{ 
    int taskValue = i; 

    // Make dummy task for example 
    Task.Run(() => 
    { 
     for (int j = 0; j < 10; j++) 
     { 
      Thread.Sleep(1000); 
      consumeFrom.Add(taskValue * 10 + j); 
     } 

     if (Interlocked.Decrement(ref producerCount) == 0) 
     { 
      consumeFrom.CompleteAdding(); 
     } 
    }); 
} 

foreach (int i in consumeFrom.GetConsumingEnumerable()) 
{ 
    Console.WriteLine(i); 
} 

Обратите внимание, что это использует по умолчанию базового источника данных ConcurrentQueue<T>. Вы можете предоставить другие реализации IProducerConsumerCollection<T> к BlockingCollection<T> конструктора, например .:

BlockingCollection<int> consumeFrom = 
    new BlockingCollection<int>(new ConcurrentBag<int>());