Параллельные коллекции. Низкоуровневая синхронизация

ConcurrentStack

ConcurrentStack представляет собой потокобезопасную коллекцию, обслуживаемую по принципу "последним поступил - первым обслужен" (LIFO). Данная коллекция похожа на коллекцию ConcurrentQueue, но с другими методами доступа к элементам. Класс ConcurrentStack определяет методы Push(), PushRange(), TryPeek(), TryPop() и TryPopRange(). Более подробно методы класса ConcurrentStack описаны в Табл. 15.4.

Пример использования коллекции ConcurrentStack представлен ниже:

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Collections.Concurrent;
using System.Collections;
namespace ConcurrentStackExample
{
    class Program
    {
    
        static void Main(string[] args)
        {
            // создаем коллекцию
            ConcurrentStack<int> sharedStack = new ConcurrentStack<int>();
            // заполняем коллекцию в цикле с помощью метода Push
            for (int i = 0; i < 1000; i++)
            {
                sharedStack.Push(i);
            }
            // объявляем переменную-счетчик количества обработанных элементов
            int itemCount = 0;
            // создаем список задач
            Task[] tasks = new Task[10];
            for (int i = 0; i < tasks.Length; i++)
            {
                // создаем задачу
                tasks[i] = new Task(() =>
                {
                    while (sharedStack.Count > 0)
                    {
                        Thread.Sleep(10);
                        int queueElement;
                        // удаляем элемент из коллекции с помощью метода TryPop
                        bool gotElement = sharedStack.TryPop(out queueElement);
                        // увеличиваем значение переменной и сохраняем результат
                        if (gotElement)
                        {                   
                        Interlocked.Increment(ref itemCount); 
                        } 
                    }
                });
                // запускаем новую задачу
                tasks[i].Start();
            }
                // ожидаем завершения всех задач
                Task.WaitAll(tasks);
                // выводим на экран отчет о количестве обработанных элементов
                Console.WriteLine("Обработанно элементов: {0}", itemCount);
                 Console.ReadLine();
        }
    }
}

ConcurrentBag

Коллекция СoncurrentBag представляет собой потокобезопасную неупорядоченную коллекцию объектов и реализует концепцию отображения потоков на используемые внутренне массивы, и старается избежать блокировок. Для доступа к элементам применяются методы Add(), TryPeek() и TryTake() (Табл. 15.5). СoncurrentBag можно представить как набор очередей с двусторонним доступом (deque). Каждый поток при работе с коллекцией обращается к своей собственной очереди, добавляя и удаляя элементы с ее начала. Когда случается так, что очередь одного потока пуста, а ему нужно извлечь элемент, то он извлекает его из очереди соседнего потока, но уже не сначала очереди, а с противоположного конца очереди. Такой подход позволяет практически не пересекаться разным потокам по данным.

Пример использование коллекции СoncurrentBag представлен ниже:

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Collections.Concurrent;
using System.Collections;
namespace ConcurrentCollection
{
    class Program
    {
    
        static void Main(string[] args)
        {
            // создаем коллекцию
            ConcurrentBag<int> sharedBag = new ConcurrentBag<int>();
            // заполняем коллекцию в цикле с помощью метода Add
            for (int i = 0; i < 1000; i++)
            {
                sharedBag.Add(i);
            }
            // объявляем переменную-счетчик количества обработанных элементов
            int itemCount = 0;
            // создаем список задач
            Task[] tasks = new Task[10];
            for (int i = 0; i < tasks.Length; i++)
            {
                // создаем задачу
                tasks[i] = new Task(() =>
                {
                    while (sharedBag.Count > 0)
                    {
                        Thread.Sleep(10);
                        int queueElement;
                        // удаляем элемент из коллекции с помощью метода TryTake
                        bool gotElement = sharedBag.TryTake(out queueElement);
                        // увеливам значение переменной и сохраняем результат
                        if (gotElement)
                        {
                     
                      
                        Interlocked.Increment(ref itemCount);  } 
                    }
                });
                // запускаем новую задачу
                tasks[i].Start();
            }
                // ожидаем завершения всех задач
                Task.WaitAll(tasks);
                // выводим на экран отчет о количестве обработанных элементов
                Console.WriteLine("Обработанно элементов: {0}", itemCount);
                 Console.ReadLine();
    
        }
    }
}

ConcurrentDictionary

Коллекция ConcurrentDictionary представляет собой потокобезопасную коллекцию пар "ключ-значение", доступ к которой могут одновременно получать несколько потоков. Для доступа к элементам коллекции в не блокирующем режиме служат методы TryAdd(), TryGetValue(), TryRemove() и TryUpdate() (Табл. 15.6). Поскольку элементы основаны на ключах и значениях, ConcurrentDictionary<TKey, TVal> не реализует интерфейс IProducerConsumerCollection<T>.

Методы, представленные в Табл. 15.6, следует использовать в следующих случаях:

Пример использования коллекции ConcurrentDictionary представлен ниже:

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Collections.Concurrent;
using System.Threading.Tasks;
namespace ConcurrentDictionaryExample
{
    class BankAccount
    {
        public int Balance
        {
            get;
            set;
        }
        class Program
        {
            static void Main(string[] args)
            {
                // создаем экземпляр банковского счета
                BankAccount account = new BankAccount();
                // создаем коллекцию
                ConcurrentDictionary<object, int> sharedDict
                    = new ConcurrentDictionary<object, int>();
                        // создаем список задач, которые возвращают целочисленный массив
                Task<int>[] tasks = new Task<int>[10];
                for (int i = 0; i < tasks.Length; i++)
                {
                    // помещаем начальные значения в словарь
                    sharedDict.TryAdd(i, account.Balance);
                    // создаем новую задачу
                    tasks[i] = new Task<int>((keyObj) =>
                    {
                        // создаем переменную для использования в цикле
                        int currentValue;
                        bool gotValue;
                        // создаем цикл для обновления баланса счета
                        for (int j = 0; j < 1000; j++)
                        {
                            // получаем текущее значение из словаря
                            gotValue = sharedDict.TryGetValue(keyObj, out currentValue);
                            // увеличиваем значение и обновляем словарь
                            sharedDict.TryUpdate(keyObj, currentValue + 1, currentValue);
                        }
                        // создаем переменную конечного результата
                        int result;
                        // получаем результат из словаря
                        gotValue = sharedDict.TryGetValue(keyObj, out result);
                        // возвращаем значение результата, если есть
                        if (gotValue)
                        {
                            return result;
                        }
                        else
                        {
                            // если нет результата - вызываем исключение
                            throw new Exception(
                                String.Format("Нет элементов данных доступных для объекта {0}", keyObj));
                        }
                    }, i);
                    // запускаем задачу
                    tasks[i].Start();
                }
                // обновляем баланс счета с помощью результатов выполнения задач
                for (int i = 0; i < tasks.Length; i++)
                {
                    account.Balance += tasks[i].Result;
                }
                // выводим значение счетчика
                Console.WriteLine("Ожидаемое значение: {0}, Баланс: {1}",
                    10000, account.Balance);
                Console.ReadLine();
            }
        }
    }
}

BlockingCollection

Коллекция BlockingCollection, представляет собой потокобезопасную коллекцию, которая осуществляет блокировку и ожидает, пока не появится возможность выполнить действие по добавлению или извлечению элемента. Пример использования коллекции BlockingCollection представлен ниже:

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Collections.Concurrent;
using System.Threading.Tasks;


namespace BlockingCollectionExample
{
     class BankAccount {
        public int Balance {
            get;
            set;
        }
    }
    class Deposit {
        public int Amount {
            get;
            set;
        }
    }
    class Program
    {
        static void Main(string[] args)
        {
             // создаем коллекцию BlockingCollection 
            BlockingCollection<Deposit> blockingCollection
                = new BlockingCollection<Deposit>();
            // создаем и запускаем задачу, которая будет генерировать депозиты и помещать их в коллекцию
            Task[] producers = new Task[3];
            for (int i = 0; i < 3; i++) {
                producers[i] = Task.Factory.StartNew(() => {
                    // создаем депозиты
                    for (int j = 0; j < 20; j++) {
                     // создаем перевод
                        Deposit deposit = new Deposit { Amount = 100 }; 
                     // помещаем перевод в коллекцию
                        blockingCollection.Add(deposit);
                    }
                });
            };
            // создаем продолжение, которое будет сигнализировать об окончании "поставки" 
            Task.Factory.ContinueWhenAll(producers, antecedents => {
                // создаем сигнал - "поставка" окончено
                Console.WriteLine("Сигнал об окончании производствки");
                blockingCollection.CompleteAdding();
            });
          // создаем банковский счет
            BankAccount account = new BankAccount();
          // создаем потребителя, который будет обновлять баланс, основанный на депозитах
            Task consumer = Task.Factory.StartNew(() => {
                while (!blockingCollection.IsCompleted) {
                    Deposit deposit;
                    // пытаемся получить следующий элемент коллекции 
                    if (blockingCollection.TryTake(out deposit)) {
                        // обновляем баланс с учетом суммы перевода
                        account.Balance += deposit.Amount;
                    }
                }
                // выводим финальный баланс
                Console.WriteLine("Итоговый баланс: {0}", account.Balance);
            });   
            consumer.Wait();
            Console.ReadLine();
        }
    }
} 

При использовании коллекции BlockingCollection следует:

Создание экземпляра класса BlockingCollection

Первым шагом, как говорилось ранее, является создание экземпляра класса BlockingCollection. Как показано в примере класс BlockingCollection - строго типизирован, и тип соответствует классу Debosit:

// создаем коллекцию BlockingCollection 
BlockingCollection<Deposit> blockingCollection = new BlockingCollection<Deposit>();

Конструктор по умолчанию использует неограниченный тип коллекции, это означает, что нет ограничений на количество "выдающихся" рабочих элементов в коллекции. Поставщики могут добавлять элементы в коллекцию гораздо быстрее, чем потребители использовать их. Для создания коллекции с ограничением используйте конструктор с аргументом int, как продемонстрировано ниже:

BlockingCollection<Deposit> blockingCollection = new BlockingCollection<Deposit>(5);

Этот пример создает экземпляр класса BlockingCollection, ограниченный пятью элементами, т.е. если поставщик попытается добавить элемент в коллекцию, которая уже будет содержать пять элементов, то метод Add() будет заблокирована, до тех пор, пока потребитель не использует элемент с помощью метода Take().

Создание поставщика

Следующий шаг - это создание процедуры выполняющую роль поставщика. В нашем примере, создается три задачи, которые создают двадцать экземпляров класса Deposit и добавляют их в коллекцию с помощью метода Add():

   Task[] producers = new Task[3];
            for (int i = 0; i < 3; i++) {
                producers[i] = Task.Factory.StartNew(() => {
                    // создаем депозиты
                    for (int j = 0; j < 20; j++) {
                     // создаем перевод
                        Deposit deposit = new Deposit { Amount = 100 }; 
                     // помещаем перевод в коллекцию
                        blockingCollection.Add(deposit);
                    }
                });
            };

Метод Add() блокируется, до тех пор, пока коллекция BlockingCollection не примет новый элемент данных, т.е если другая задача попытается добавить или взять элемент из коллекции, вызов метода Add() будет заблокирован, пока другая операция, над коллекцией, не будет завершена, это связанно с тем что все операции в BlockingCollection синхронизированы.

Класс BlockingCollection имеет ряд методов для добавления элементов в коллекцию, все эти методы описаны в Табл. 15.7.

Создание потребителя

Использование модели поставщик/потребитель позволяет создать "асимметрию" между поставщиком и потребителем. Использование данной модели необходимо в том случае, если производство и потребление элемента, занимает разное количество времени, Код ниже демонстрирует реализацию потребителя:

Task consumer = Task.Factory.StartNew(() => {
       while (!blockingCollection.IsCompleted) {
           Deposit deposit;
           // пытаемся получить следующий элемент коллекции 
           if (blockingCollection.TryTake(out deposit)) {
               // обновляем баланс с учетом суммы перевода
               account.Balance += deposit.Amount;
           }
       }
       // выводим финальный баланс
       Console.WriteLine("Итоговый баланс: {0}", account.Balance);
            });

Как видно из примера, задача-потребитель входит в цикл, который выполнится в том случае, если свойство BlockingCollection.IsCompleted возвращает значение false. Свойство IsCompleted возвращает значение true, когда вызывается метод CompleteAdding() и коллекция не содержит элементов, это будет означать, что производство и потребление элементов коллекции, было завершено. Обобщающая информация по данным свойствам и методам представлена в Табл. 15.8.

В то время как в цикле, потребитель вызывает метод TryTake() для получения элемента из коллекции. Этот метод возвращает значение true, если элемент был успешно получен и присваивает значение элемента параметру (out deposit). Можно также было использовать Take(), который блокируется, пока элемент коллекции не будет доступен для потребления. Обобщающая информация по данным свойствам и методам представлена в Табл. 15.9.

Создание приложений с использованием параллельных коллекций

В данном практическом задании будет рассмотрено два примера создания консольного приложения с использованием параллельных коллекций BlockingCollection и ConcurrentQueue.

BlockingCollection

Создадим консольное приложение и назовем его, к примеру, "BlockingCollectionApplication" (рис P.1)

В новом проекте, подключим директивы на использование потоков и параллельных коллекций:

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

Объявляем блокирующую коллекцию символов BlockingCollection<char>:

class Program
    {
      
            static BlockingCollection<char> bc;
….
     }

Создадим статичный метод Produce(), который "производит" и "поставляет" символы от А до Я:

static void Producer()
   {
       for (char ch = 'A'; ch <= 'Я'; ch++)
       {
           bc.Add(ch);
           Console.WriteLine("Производится символ " + ch);
       }
   }

Создадим статичный метод Consumer(), который будет "потреблять" символы метода Produce():

static void Consumer()
   {
       for (int i = 0; i < 34; i++)
       {
           Console.WriteLine("Потребляется символ "+bc.Take());
       }
   }

Примечание. Метод Take удаляет элемент из коллекции.

В методе Main() используем блокирующую коллекцию, ограниченную 4 элементами:

bc = new BlockingCollection<char>(4)

Создаем задачи "Поставщика" и "Потребителя":

Task Prod = new Task(Producer)
Task Con = new Task(Consumer)

Запускаем задачи:

Con.Start()
Prod.Start()

Ожидаем завершение задач "Поставщик" и "Потребитель" с помощью метода WaitAll():

try
     {
         Task.WaitAll(Con, Prod);
     }
     catch (AggregateException exc)
     {
         Console.WriteLine(exc);
     }
     finally
     {
         Con.Dispose();
         Prod.Dispose();
         bc.Dispose();
     }

В итоге получаем следующее содержание метода Main():

static void Main(string[] args)
    {
        bc = new BlockingCollection<char>(4);
        Task Prod = new Task(Producer);
        Task Con = new Task(Consumer);
        Con.Start();
        Prod.Start();
        try
        {
            Task.WaitAll(Con, Prod);
        }
        catch (AggregateException exc)
        {
            Console.WriteLine(exc);
        }
        finally
        {
            Con.Dispose();
            Prod.Dispose();
            bc.Dispose();
        }
        Console.ReadLine();
    }

Запустим программу. На экране появиться смешанный результат, которые выводятся методами Produce и Consumer: (рис P.2)

Смешанный результат объясняется тем, что коллекция ограничена 4-мя элементами, это означает, что в неё может быть добавлено только 4 элемента, прежде чем её придется сократить.

К примеру, если убрать ограничение в четыре символа из коллекции:

bc = new BlockingCollection<char>();

То получится следующий результат:

(рис P.3)

Модифицируем метод Producer, добавив в него метод CompleteAdding().:

static void Producer()
    {
        for (char ch = 'А'; ch <= 'Я'; ch++)
        {
            bc.Add(ch);
            Console.WriteLine("Производится символ " + ch);
        }
    bc.CompleteAdding();
    }

Вызов метода CompleteAdding() означает, что в коллекцию не будет добавлено ни одного элемента, это приводит к тому, что свойство IsAddingComplete принимает логическое значение true. Если коллекция пуста, то свойство IsCompleted, принимает логическое значение true, и в этом случае вызовы метода Take() не блокируется.

Модифицируем метод Consumer, добавив в него метод TryTake.:

static void Consumer()
    {
        char ch;
            while(!bc.IsCompleted)
            {
         if(bc.TryTake(out ch))
            Console.WriteLine("Потребляется символ "+bc.Take());
            }       
    }

Теперь данный метод будет "потреблять" символы до тех пор, пока их будет производить метод-поставщик Consumer().

Примечание. Метод TryTake() возвращает логическое значение true, если элемент коллекции был удален.

Запустим программу. В окне запущенной программы отобразиться следующий результат: (рис P.4)

Данный вариант программы дает такой же результат, как и предыдущий, только с одним отличием: метод Producer() может "производить" и "поставлять" сколько угодно элементов. С этой целью он вызывает метод CompleteAdding() когда завершает создание элементов. Метод Consumer() лишь "потребляет" произведенные элементы до тех пор, пока свойство IsCompleted не примет значение true.

Листинг кода программы:

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Threading;
using System.Collections.Concurrent;

namespace BlockingCollectionApplication
{
    class Program
    {
        static BlockingCollection<char> bc;
        static void Producer()
        {
            for (char ch = 'А'; ch <= 'Я'; ch++)
            {
                bc.Add(ch);
                Console.WriteLine("Производится символ " + ch);
            }
       bc.CompleteAdding();
        }
        static void Consumer()
        {
            char ch;     
                while(!bc.IsCompleted)
               {
             if(bc.TryTake(out ch))
                Console.WriteLine("Потребляется символ "+bc.Take());
               }
            
        }
        static void Main(string[] args)
        {
            bc = new BlockingCollection<char>(4);
            Task Prod = new Task(Producer);
            Task Con = new Task(Consumer);
            Con.Start();
            Prod.Start();
            try
            {
                Task.WaitAll(Con, Prod);
            }
            catch (AggregateException exc)
            {
                Console.WriteLine(exc);
            }
            finally
            {
                Con.Dispose();
                Prod.Dispose();
                bc.Dispose();
            }
            Console.ReadLine();
        }
    }

}

ConcurrentQueue

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

Создадим консольное приложение и назовем его "ConcurrentQueueApplication": (рис P.5)

Создадим в классе Program, три статичных коллекции:

static List<int> collection1 = new List<int>();
       static Queue<int> collection2 = new Queue<int>();
       static ArrayList collection3 = new ArrayList();

Предварительно подключив директиву, для использования коллекции типа ArrayList:

using System.Collections

Далее, создадим перегруженный статичный метод, который помещает значения в созданные на предыдущем шаге коллекции:

static void addNumber(int num)
    {
        try
        {
            collection1.Add(num);
            collection2.Enqueue(num);
            collection3.Add(num);
        
        }
        catch (AggregateException exc)
        {
            Console.WriteLine(exc);
        }

В главном методе Main, реализуем наполнение коллекций через цикл for с использованием метода addNumber:

for (int i = 0; i < 1000000; i++)
     {
         addNumber(i);
     }
     Console.WriteLine("Коллекция List<int> содержит: 
{0} элементов \nКоллекция Queue<int> содержит: {1} элементов \nКоллекция ArrayList содержит: 
{2} элементов", collection1.Count, collection2.Count, collection3.Count);
     Console.ReadLine();

Запустим программу. И получим следующий результат: (рис P.6)

Примечание. Как видно из результата выполнения программы все значения были добавлены в коллекции при прохождении цикла for.

Листинг кода программы:

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Collections;
using System.Threading.Tasks;
namespace ConcurrentQueueApplication
{
    class Program
    {
       static List<int> collection1 = new List<int>();
       static Queue<int> collection2 = new Queue<int>();
       static ArrayList collection3 = new ArrayList();
       static void addNumber(int num)
        {
            try
            {
                collection1.Add(num);
                collection2.Enqueue(num);
                collection3.Add(num);
             }
             catch (Exception exc)
            {
                Console.WriteLine(exc);
            }
        }
        static void Main(string[] args)
        {
            for (int i = 0; i < 1000000; i++)
            {
                addNumber(i);
            }
            Console.WriteLine("Коллекция List<int> содержит: 
  {0} элементов \nКоллекция Queue<int> содержит: 
   {1} элементов \nКоллекция ArrayList содержит: {2} элементов", 
    collection1.Count, collection2.Count, collection3.Count);
            Console.ReadLine();
        }
    }
}

Теперь используем параллельный цикл Parallel.For вместо цикла for для того, что бы посмотреть, как выполняется добавление записей с использованием многопоточного цикла. Для этого, сначала, подключим директиву для использования потоков:

using System.Threading.Tasks

Заменим цикл for на Parallel.For:

Parallel.For(0, 1000000, i =>
            {
                addNumber(i);
            });

Запустим программу. Программа выдаст приблизительно следующий результат: (рис P.7)

Так же программа может вызвать исключение при добавлении элемента в одну из коллекций:

(рис P.8)

Примечание. Как видно из результатов выполнения цикла Parallel.For, не целесообразно использовать стандартные коллекции для распараллеленного кода, это приводит либо к ошибкам в ходе выполнения программы, либо к тому, что не все элементы будут помещены в коллекцию, это связанно с тем, что несколько потоков могут обращаться к одной и той же позиции коллекции при добавлении элемента.

Теперь создадим коллекцию ConcurrentQueue. Для этого сначала подключим директиву для того, что бы использовать параллельную коллекцию:

using System.Collections.Concurrent

Создаем коллекцию ConcurrentQueue с именем collection4:

static ConcurrentQueue<int> collection4 = new ConcurrentQueue<int>()

Добавляем элементы в коллекцию:

collection4.Enqueue(num)

Выводим количество элементов коллекции collection4:

Console.WriteLine("Коллекция List<int> содержит: 
{0} элементов \nКоллекция Queue<int> содержит: 
{1} элементов \nКоллекция ArrayList содержит: 
{2} элементов \nКоллекция ConcurrentQueue содержит: 
{3} элементов", collection1.Count, collection2.Count, collection3.Count,collection4.Count)

Запустим программу. И если программа отработает безошибочно, то результат будет следующим: (рис P.9)

Примечание. Как видно из результата выполнения программы, все элементы (от 0 до 1000000) будут помещены в коллекцию ConcurrentQueue.

Листинг кода программы:

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Collections;
using System.Threading.Tasks;
using System.Collections.Concurrent;
namespace ConcurrentQueueApplication
{
    class Program
    {
       static List<int> collection1 = new List<int>();
       static Queue<int> collection2 = new Queue<int>();
       static ArrayList collection3 = new ArrayList();
       static ConcurrentQueue<int> collection4 = new ConcurrentQueue<int>();
       static void addNumber(int num)
        {
            try
            {
                collection1.Add(num);
                collection2.Enqueue(num);
                collection3.Add(num);
                collection4.Enqueue(num);
            }
            catch (Exception exc)
            {
                Console.WriteLine(exc);
            }
        }
        static void Main(string[] args)
        {
      
            Parallel.For(0, 1000000, i =>
            {
                addNumber(i);
            });
            Console.WriteLine("Коллекция List<int> содержит: 
  {0} элементов \nКоллекция Queue<int> содержит: 
  {1} элементов \nКоллекция ArrayList содержит: {2} элементов \nКоллекция ConcurrentQueue содержит:
      {3} элементов", collection1.Count, collection2.Count, collection3.Count,collection4.Count);
           Console.ReadLine();
        }
    }
}