C# Task, ForeEach, PLINQ Потоки ч.5

Приветствую всех, сегодня поговорим о продолжении темы связанной с потоками, это часть 5. Читать Потоки ч.1 Читать Потоки ч.2 Читать Потоки ч.3 Читать потоки ч.4

Эта часть будет посвящена параллельному программированию. Библиотека параллельных задач (TPL) предназначена для повышения производительности разработчиков за счет упрощения процесса добавления параллелизма в приложения. TPL динамически масштабирует степень параллелизма для наиболее эффективного использования всех доступных процессоров. В библиотеке параллельных задач осуществляется секционирование работы, планирование потоков в пуле ThreadPool, поддержка отмены, управление состоянием и выполняются другие низкоуровневые задачи. Используя библиотеку параллельных задач, можно повысить производительность кода, сосредоточившись на работе, для которой предназначена программа. Начиная с .NET Framework 4 библиотека параллельных задач является предпочтительным способом создания многопоточного и параллельного кода. Еще одним важным средством параллельного программирования, появившемся в .NET Framework 4.0, можно считать PLINQ. Язык PLINQ дает возможность без труда внедрять параллелизм в запросы, что, в свою очередь, позволяет легко (и безопаснее) использовать системные ресурсы.

Библиотека параллельных задач (TPL) представляет собой набор открытых типов и API-интерфейсов в пространствах имен System.Threading и System.Threading.Tasks в .NET Framework 4.

Класс Task — представляет асинхронную операцию.

Класс Task<TResult> — представляет асинхронную операцию, которая может вернуть значение.

Класс Parallel предоставляет поддержку параллельных циклов и областей.

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

Элементарная единица исполнения инкапсулируется в TPL средствами класса Task. В отличии от класса Thread, имеется в виду инкапсуляция асинхронной операции, а не потока исполнения. На системном уровне поток по-прежнему остается элементарной единицей исполнения, но при использовании задач соответствие между объектом класса Task и потоком исполнения уже не является взаимо-однозначным. Кроме того, исполнением задач управляет планировщик задач, который работает с пулом потоков. Это, например, означает, что несколько задач могут выполняться в одном потоке.

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

В отличие от потоков Thread, задачи не обладают уникальными именами. Вместо этого для идентификации задач используется свойство Id типа int. Идентификаторы задач уникальны, но не упорядочены, не стоит делать каких-либо предположений касательно очередности создания задач, основываясь на значениях идентификаторов. Идентификатор исполняемой в настоящий момент задачи можно получить с помощью свойства CurrentId. Это свойство принимает значение идентификатора задачи, если обращение к нему происходит из контекста задачи либо null, если обращающийся код не является исполняемым внутри какой-либо задачи.

Метод Wait позволяет приостановить выполнение вызывающего потока до момента завершения задачи. В момент использования этого метода могут быть сгенерированы два исключения: ObjectDisposedException, если задача освобождена посредством вызова Disposed; AggregateExecption, если задача сама генерирует исключение или же отменяется. Если у задачи существуют порожденные ею подзадачи и исключения происходят внутри них, все эти исключения будут собраны и упакованы в единое исключение AggregateExecption.

Дождаться завершения более одной задачи можно используя метода WaitAll. Он позволяет приостановить выполнения основного потока до тех пор, пока не завершаться все задачи, переданные ему в качестве аргументов params. Если же нужно дождаться завершения не всех задач, а любой, следует применять метод WaitAny. Метод WaitAny возвращает Id задачи, завершившейся первой.

Класс Task реализует интерфейс IDisposable, таким образом, для него определен метод Dispose. Этот метод нужно вызывать, если в процессе работы программы создается большое количество задач, оставляемых на произвол судьбы, или же необходимо явным образом освобождать ресурсы после завершения работы задачи. Вызов метода Dispose допустим только после завершения задачи, поэтому его следует дождаться, например, применяя метод Wait. Если попытаться вызвать Dispose для все еще работающей задачи, то будет сгенерировано исключение InvalidOperationException.

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

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

Продолжение – это одна задача, которая автоматически начинается после завершения другой задачи. Создать продолжение можно, например, используя метод ContinueWith. Очень часто в качестве продолжения задачи используются лямбда-выражения.

Из задачи можно получить возвращаемое значение. Во-первых, это означает, что с помощью задач можно вычислить некоторый результат. Во-вторых, вызывающий процесс окажется блокированным до тех пор, пока не будет получен результат. Задачи, возвращающие результат создаются с использованием делегата Func, а не Action, как обычные задачи.

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

Класс Parallel позволяет использовать параллельное программирование на основе задач, не прибегая к управлению задачами явным образом. Метод Invoke позволяет выполнять один или несколько методов параллельно. Он масштабирует исполнение кода, используя доступные процессоры, если имеется такая возможность. Выполняемые методы должны быть совместимы с делегатом Action.

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

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

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

Кроме цикла For, параллельно также можно выполнять и цикл Foreach. Для него действуют все те же ограничения, что и для цикла For

PLINQ представляет собой параллельный вариант языка интегрированных запросов LINQ и тесно связан с библиотекой TPL. PLINQ применяется, главным образом, для достижения параллелизма данных внутри запроса. Основу PLINQ составляет класс ParallelEnumerable, определенный в пространстве имен System.Linq.

Пример использования Task:

 public static void MyMethod()
        {
           for (int i=0; i<80; i++)
            {
                Thread.Sleep(80);
                Console.ForegroundColor = ConsoleColor.Blue;
                Console.Write("*");
            }
        }

        static void Main(string[] args)
        {
            //создаем делегат
            Action dl = new Action(MyMethod);
            //создаем экземпляр задачи и передаем делегат
            Task t = new Task(dl);
            //запускаем задачу
            t.Start();

            // Метод Main() остается активным до завершения задачи MyTask(). 
            for (int i = 0; i < 80; i++)
            {
                Thread.Sleep(100);
                Console.ForegroundColor = ConsoleColor.Yellow;
                Console.Write("-");
            }
            Console.ReadKey();
        }

public static void MyMethod()

{

for (int i=0; i<80; i++)

{

Thread.Sleep(80);

Console.ForegroundColor = ConsoleColor.Blue;

Console.Write("*");

}

static void Main(string[] args)

{

//создаем делегат

Action dl = new Action(MyMethod);

//создаем экземпляр задачи и передаем делегат

Task t = new Task(dl);

//запускаем задачу

t.Start();

// Метод Main() остается активным до завершения задачи MyTask().

for (int i = 0; i < 80; i++)

{

Thread.Sleep(100);

Console.ForegroundColor = ConsoleColor.Yellow;

Console.Write("-");

}

Console.ReadKey();

}

Еще один пример, того как можно выполнить параллельно метод из другого класса:

class MyClass
    {
        public static void MyMethod()
        {
            for (int i = 0; i < 80; i++)
            {
                Thread.Sleep(80);
                Console.ForegroundColor = ConsoleColor.Blue;
                Console.Write("*");
            }
        }
    }
    class Prog
    { 
        
        static void Main(string[] args)
        {
           
            //создаем экземпляр задачи и передаем делегат
            Task t = new Task(MyClass.MyMethod);
            //запускаем задачу
            t.Start();

            // Метод Main() остается активным до завершения задачи MyTask(). 
            for (int i = 0; i < 80; i++)
            {
                Thread.Sleep(100);
                Console.ForegroundColor = ConsoleColor.Yellow;
                Console.Write("-");
            }
            Console.ReadKey();
        }

    }

class MyClass

{

public static void MyMethod()

{

for (int i = 0; i < 80; i++)

{

Thread.Sleep(80);

Console.ForegroundColor = ConsoleColor.Blue;

Console.Write("*");

}

class Prog

{

static void Main(string[] args)

{

//создаем экземпляр задачи и передаем делегат

Task t = new Task(MyClass.MyMethod);

//запускаем задачу

t.Start();

// Метод Main() остается активным до завершения задачи MyTask().

for (int i = 0; i < 80; i++)

{

Thread.Sleep(100);

Console.ForegroundColor = ConsoleColor.Yellow;

Console.Write("-");

}

Console.ReadKey();

}

Пример ожидания завершения работы методов.

static void Main()
		{
			Console.WriteLine("Основной поток запущен.");

			var task1 = new Task(MyTask);
			var task2 = new Task(MyTask);
                        var task3 = new Task(MyTask);
			var task4 = new Task(MyTask);

			// Выполнение задач.
			task1.Start();
			task2.Start();
                        task3.Start();
			task4.Start();
// WaitAll() - Ожидает завершения выполнения всех указанных объектов Task.
			 Task.WaitAll(task1, task2);
// WaitAny() - Ожидает завершения выполнения любого из указанных объектов Task.
			 Task.WaitAny(task3, task4);
// Метод Main() остается активным до завершения задачи MyTask(). 
                         task1.Wait();
 // Освобождение задачи. 
                         task1.Dispose();
            
			Console.WriteLine("Основной поток завершен.");

			
			Console.ReadKey();
		}

static void Main()

{

Console.WriteLine("Основной поток запущен.");

var task1 = new Task(MyTask);

var task2 = new Task(MyTask);

var task3 = new Task(MyTask);

var task4 = new Task(MyTask);

// Выполнение задач.

task1.Start();

task2.Start();

task3.Start();

task4.Start();

// WaitAll() - Ожидает завершения выполнения всех указанных объектов Task.

Task.WaitAll(task1, task2);

// WaitAny() - Ожидает завершения выполнения любого из указанных объектов Task.

Task.WaitAny(task3, task4);

// Метод Main() остается активным до завершения задачи MyTask().

task1.Wait();

// Освобождение задачи.

task1.Dispose();

Console.WriteLine("Основной поток завершен.");

Console.ReadKey();

}

Пример создания продолжения задачи:

 public static void MyMethod()
        {
            for (int i = 0; i < 80; i++)
            {
                Thread.Sleep(80);
                Console.ForegroundColor = ConsoleColor.Blue;
                Console.Write("*");
            }
        }

        public static void ContinueTask(Task task)
        {
            Console.WriteLine("\nПродолжение отработало");
        }

        static void Main(string[] args)
        {

            Task task = new Task(MyMethod);
            //создания продолжения задачи
            Task continueTask = task.ContinueWith(ContinueTask);

            task.Start();
            
            //ожидаем завершения
            continueTask.Wait();
            //освобождаем занимаемые ресурсы
            task.Dispose();
            continueTask.Dispose();


            Console.WriteLine("Основной поток завершен");
            Console.ReadKey();
        }

public static void MyMethod()

{

for (int i = 0; i < 80; i++)

{

Thread.Sleep(80);

Console.ForegroundColor = ConsoleColor.Blue;

Console.Write("*");

}

public static void ContinueTask(Task task)

{

Console.WriteLine("\nПродолжение отработало");

}

static void Main(string[] args)

{

Task task = new Task(MyMethod);

//создания продолжения задачи

Task continueTask = task.ContinueWith(ContinueTask);

task.Start();

//ожидаем завершения

continueTask.Wait();

//освобождаем занимаемые ресурсы

task.Dispose();

continueTask.Dispose();

Console.WriteLine("Основной поток завершен");

Console.ReadKey();

}

Пример использования параллельного цикла For:

    static void MyMethod(int i)
        {
           Console.WriteLine("Параллельный цикл {0}",i);
        }

        static void Main(string[] args)
        {

            int[] n =new int [10];
           
            Parallel.For(0, n.Length, MyMethod);

            Console.ReadKey();
        }

static void MyMethod(int i)

{

Console.WriteLine("Параллельный цикл {0}",i);

}

static void Main(string[] args)

{

int[] n =new int [10];

Parallel.For(0, n.Length, MyMethod);

Console.ReadKey();

}

Пример использования ForeEach:

// Метод служащий в качестве тела параллельно выполняемого цикла.
// Переменной i передается значение элемента массива данных, а не индекс элемента.
        public static void Show(int i)
        {
            Console.WriteLine(i);
        }

        static void Main(string[] args)
        {

            int [] n =new int [10];

            Parallel.For(0, n.Length, i => n[i]=i);

            Parallel.ForEach(n,Show);
            
            Console.ReadKey();
        }

// Метод служащий в качестве тела параллельно выполняемого цикла.

// Переменной i передается значение элемента массива данных, а не индекс элемента.

public static void Show(int i)

{

Console.WriteLine(i);

}

static void Main(string[] args)

{

int [] n =new int [10];

Parallel.For(0, n.Length, i => n[i]=i);

Parallel.ForEach(n,Show);

Console.ReadKey();

}

Пример нахождения отрицательного числа в массиве с помощью PLINQ

  static void Main()
        {
            var data = new int[10000000];

            // Инициализация массива данных положительными значениями.
            for (int i = 0; i < data.Length; i++) 
                data[i] = i;

            // Заполнение массива отрицательными значениями.
            data[1000] = -32;
            data[11000] = -432;
            data[12000] = -223;
            data[32000] = -43;
            data[734500] = -521;
            data[957000] = -323;

            // Запрос PLINQ для поиска отрицательных значений.
            var negatives = from val in data.AsParallel() // ParallelEnumerable.AsParallel<int>(data)
                            where val < 0
                            select val;

            foreach (var v in negatives)
                Console.Write(v + " ");

            Console.ReadKey();
        }

static void Main()

{

var data = new int[10000000];

// Инициализация массива данных положительными значениями.

for (int i = 0; i < data.Length; i++)

data[i] = i;

// Заполнение массива отрицательными значениями.

data[1000] = -32;

data[11000] = -432;

data[12000] = -223;

data[32000] = -43;

data[734500] = -521;

data[957000] = -323;

// Запрос PLINQ для поиска отрицательных значений.

var negatives = from val in data.AsParallel() // ParallelEnumerable.AsParallel<int>(data)

where val < 0

select val;

foreach (var v in negatives)

Console.Write(v + " ");

Console.ReadKey();

}

Следующий пример, позволяет заполнить массив рандомными числами, выбрать не четные значения в отдельную коллекцию и осуществить вывод в консоль, все части выполнены на основе параллельного программирования:

   public static void Show(int i)
        {
            Console.WriteLine(i);
        }

        static void Main(string[] args)
        {

            int[] i = new int[100];
            Random r = new Random();
            Parallel.For(0,i.Length,n=>i[n]=r.Next());

            ParallelQuery<int> w = from f in i.AsParallel()
                    where f % 2 != 0
                    select f;

            Parallel.ForEach(w,Show);
            
            Console.ReadKey();
        }

public static void Show(int i)

{

Console.WriteLine(i);

}

static void Main(string[] args)

{

int[] i = new int[100];

Random r = new Random();

Parallel.For(0,i.Length,n=>i[n]=r.Next());

ParallelQuery<int> w = from f in i.AsParallel()

where f % 2 != 0

select f;

Parallel.ForEach(w,Show);

Console.ReadKey();

}

Пример работы Task.Factory, который демонстрирует работу 2х методов запуская 2 параллельных потока, без остановки основного потока:

  public static void Method1()
        {
            for(int i=0; i < 100; i++)
            {
                Thread.Sleep(100);
                Console.ForegroundColor = ConsoleColor.Blue;
                Console.Write("*");
            }

        }
        public static void Method2()
        {
            for (int i = 0; i < 100; i++)
            {
                Thread.Sleep(100);
                Console.ForegroundColor = ConsoleColor.Yellow;
                Console.Write("-");
            }
        }


     
        static void Main(string[] args)
        {
            Task.Factory.StartNew(()=>Parallel.Invoke(Method1,Method2));
            Console.WriteLine("Main завершился");
            Console.ReadKey();
        }

public static void Method1()

{

for(int i=0; i < 100; i++)

{

Thread.Sleep(100);

Console.ForegroundColor = ConsoleColor.Blue;

Console.Write("*");

}

public static void Method2()

{

for (int i = 0; i < 100; i++)

{

Thread.Sleep(100);

Console.ForegroundColor = ConsoleColor.Yellow;

Console.Write("-");

}

static void Main(string[] args)

{

Task.Factory.StartNew(()=>Parallel.Invoke(Method1,Method2));

Console.WriteLine("Main завершился");

Console.ReadKey();

}

Читать ч.6 async\await

Добавить комментарий Отменить ответ