Параллельные коллекции

Пример (Рис. 14.1) реализации интерфейса IProducerConsumerCollection<T> представлен в примере ниже:

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

namespace IProducerConsumerCollection
{
    //Реализуем интерфейс IProducerConsumerCollection<T> в классе ProducerConsumerCollection<T>
    class ProducerConsumerCollection<T> : IProducerConsumerCollection<T>
    { 
        private IProducerConsumerCollection<T> _queue = new ConcurrentQueue<T>(); 
        
        public void CopyTo(T[] array, int index)
        {
            Console.WriteLine("Вызов метода CopyTo");
            _queue.CopyTo(array, index);
        }
        public T[] ToArray()
        {
            Console.WriteLine("Вызов метода ToArray");
            return _queue.ToArray();
        }
        public bool TryAdd(T item)
        {
            Console.WriteLine("Вызов метода TryAdd " + item.ToString());
            return _queue.TryAdd(item);
        }
        public bool TryTake(out T item)
        {
            var ret = _queue.TryTake(out item);
            var val = "";
            if (item == null) 
            { val = "NULL"; }
            else
            { val = item.ToString(); }
            Console.WriteLine("Вызов метода TryTake возрвращает объект коллекции " + val + 
   " метод возвращает значение " + ret.ToString());
            return ret;
        }
        public void CopyTo(Array array, int index)
        {
            _queue.CopyTo(array, index);
        }
        public IEnumerator<T> GetEnumerator()
        {
            Console.WriteLine("GetEnumerator T..");
            return _queue.GetEnumerator();
        }
        IEnumerator System.Collections.IEnumerable.GetEnumerator()
        {
            Console.WriteLine("GetEnumerator..");
            return _queue.GetEnumerator();
        }
        public int Count
        {
            get
            {
                Console.WriteLine("Вызов метода Count...");
                return _queue.Count;
            }
        }
        public bool IsSynchronized
        {
            get { return _queue.IsSynchronized; }
        }
        public object SyncRoot
        {
            get { return _queue.SyncRoot; }
        }
    }
    
    class Program
    {
        static void Main(string[] args)
        {
            ProducerConsumerCollection<string> _prodConsumer = new ProducerConsumerCollection<string>(); 
//создаем объект класса ProducerConsumerCollection
            _prodConsumer.TryAdd("Значение 1 коллекции");//добавляем значения
            _prodConsumer.TryAdd("Значение 2 коллекции");//в коллекцию
            _prodConsumer.TryAdd("Значение 3 коллекции");
            _prodConsumer.TryAdd("Значение 4 коллекции");
            string s;
            _prodConsumer.TryTake(out s);
            _prodConsumer.TryTake(out s);
            var array = _prodConsumer.ToArray();
            for (int n = 0; n < array.Length; n++)
            {
                Console.WriteLine(array[n]);
            }
            Console.ReadLine();
    
        }
    }
}

Пример использования обычной коллекции с применением параллелизма

Один из наиболее распространенных способов разделения данных - это использование коллекций. Применение параллельного программирования к коллекциям содержащим большой массив данных: позволит в значительной степени увеличить скорость обработки алгоритма(ов). При этом разработчик может столкнуться с проблемой "соревнования" потоков за доступ к данным коллекции. Пример ниже демонстрирует использование коллекции Queue<int> которая содержит 1000 элементов и элементы которой обрабатывает массив из 10 задач. Каждая задача асинхронно удаляет первый элемент коллекции, и увеличивает значении переменной-счетчика, используя для синхронизации атомарную операцию Interlocked.Increment()

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)
        {
            // создаем коллекцию
            Queue<int> sharedQueue = new Queue<int>();
            // заполняем коллекцию в цикле с помощью метода Enqueue
            for (int i = 0; i < 1000; i++)
            {
                sharedQueue.Enqueue(i);
            }
            // объявляем переменную-счетчик количества обработанных элементов
            int itemCount = 0;
            // создаем список задач
            Task[] tasks = new Task[10];
            for (int i = 0; i < tasks.Length; i++)
            {
                // создаем задачу
                tasks[i] = new Task(() =>
                {
                    while (sharedQueue.Count > 0)
                    {
                        // удаляем элемент из коллекции с помощью метода Dequeue
                        int item = sharedQueue.Dequeue();
                        // увеливам значение переменной и сохраняем результат
                        Interlocked.Increment(ref itemCount);
                    }
                });
                // запускаем новую задачу
                tasks[i].Start();
            }
                // ожидаем завершения всех задач
                Task.WaitAll(tasks);
                // выводим на экран отчет о количестве обработанных элементов
                Console.WriteLine("Обработанно элементов: {0}", itemCount);
                 Console.ReadLine();
        }
    }
}

Коллекция Queue<int> не является потокобезапассной, поэтому в случае рассинхронизации потоков (метод Thread.Sleep()), может возникнуть исключение InvalidOperationException (Рис. 14.3). Это исключение возникнет, когда один из потоков попытается вызвать метод Dequeue(), при этом сама коллекция будет уже пустой.