Многопоточность в java
Содержание:
- Что такое многопоточность в Java?
- Ограничения классического подхода
- 1 Класс OutputStream
- Java Thread Join(). Теория
- Прерывание потоков
- Как выбирать производителя твердотельных накопителей
- Пример создания потока. Наследуем класс Thread
- Конкуренция и параллелизм
- 1 Потоки данных
- Методы wait и notify
- Новые возможности пакета java.uti.concurrent
- 1 Нововведения в Java 8: Функциональное программирование
- Настройка размера пула
- Запуск задач с помощью java.util.concurrent.ExecutorService
Что такое многопоточность в Java?
MULTITHREADING или многопоточность в Java — это процесс одновременного выполнения двух или более потоков с максимальной загрузкой ЦП. Многопоточные приложения выполняют одновременно два или более потоков. Следовательно, он также известен как параллелизм в Java. Каждый поток проходит параллельно друг другу.
Несколько потоков не выделяют отдельную область памяти, следовательно, они экономят память. Кроме того, переключение контекста между потоками занимает меньше времени.
Пример:
package demotest;
public class GuruThread1 implements Runnable
{
public static void main(String[] args) {
Thread guruThread1 = new Thread("Guru1");
Thread guruThread2 = new Thread("Guru2");
guruThread1.start();
guruThread2.start();
System.out.println("Thread names are following:");
System.out.println(guruThread1.getName());
System.out.println(guruThread2.getName());
}
@Override
public void run() {
}
}
Преимущества многопоточности:
- Пользователи не заблокированы, потому что потоки независимы, и мы можем выполнять несколько операций за раз
- Поскольку такие потоки независимы, другие потоки не будут затронуты, если один поток встретит исключение.
Ограничения классического подхода
Когда программист только начинает работать с многопоточными возможностями Java-платформы, то на первых порах он может даже впасть в состояние «эйфории», особенно, если у него уже был негативный опыт по созданию многопоточных приложений в других языках программирования
Действительно, система управления потоками в Java организованна крайне удачно, так как этот компонент платформы был детально проработан еще на стадии проектирования самых первых версий виртуальной Java-машины и языка программирования Java, и сейчас ему по-прежнему уделяется большое внимание. Однако со временем «розовые очки» спадают, и программист начинает замечать «раздражающие» моменты, которые усложняют организацию параллельного исполнения задач в Java
Первым, что бросается в глаза, оказывается слияние низкоуровневого кода, отвечающего за многопоточное исполнение, и высокоуровневого кода, отвечающего за основную функциональность приложения (так называемый «спагетти-код»). В листинге 1 показано, что бизнес—код и поточный код вообще находятся в одном классе, но даже в более удачном варианте из листинга 2 для выполнения задачи все равно требуется создать объект Thread и запустить его. Подобное перемешивание снижает качество архитектуры приложения и может затруднить его последующее сопровождение.
Но даже если удалось отделить поточный код от основного, то возникает проблема, связанная уже с управлением самими потоками. Потоки в Java запускаются только путем вызова метода start и останавливаются после вызова соответствующих методов или самостоятельно после завершения работы метода run. Также после того, как поток остановился, его нельзя запустить второй раз, что и приводит к следующим негативным моментам:
- поток занимает относительно много места в куче, так что после его завершения необходимо проследить, чтобы память, занимаемая им, была освобождена (например, присвоить ссылке на поток значение null);
- для выполнения новой задачи потребуется запустить новый поток, что приведет к увеличению «накладных расходов» на виртуальную машину, так как запуск потока – это одна из самых требовательных к ресурсам операций.
Если же удалось превратить поток в «многоразовый», то программист сталкивается с проблемой, как понять, что поток уже закончил выполнение задачи и ему можно выдавать следующую. Необходимо еще учитывать тот факт, что выполнение задачи может завершиться неудачно, например, возникновением исключительной ситуации, и подобная ситуация не должна повлиять на выполнение других задач.
Важно сказать, что «среднестатистический» программист будет отнюдь не первым, кто сталкивается с подобными проблемами в многозадачных приложениях. Все эти проблемы были давно проанализированы Java-сообществом и нашли решение в признанных шаблонах проектирования (например, ThreadPool (пул потоков) и WorkerThread (рабочий поток))
Но скорее всего, у рядового программиста, ограниченного временными рамками проекта, просто не будет времени или ресурсов, чтобы подготовить и самое главное протестировать полноценную реализацию данных шаблонов. А неточное или неполное внедрение этих шаблонов (да и вообще любых шаблонов проектирования) может в будущем негативно сказаться на этапе сопровождения продукта.
1 Класс OutputStream
С потоками ввода мы только что разобрались. Настало время поговорить о потоках вывода.
Класс является классом-родителем для всех классов, которые поддерживают байтовый вывод. Это абстрактный класс, который сам ничего не делает: для этого у него есть классы-наследники на все случаи жизни.
Сложновато звучит. Если попроще, этот класс оперирует байтами, а не, например, символами или другими типами данных. А то, что он абстрактный, значит, что мы обычно используем не его, а один из его классов-наследников. Например, и ему подобные.
Но вернемся к классу . У этого класса есть методы, которые обязаны реализовывать все его классы-наследники. Вот основные из них:
| Методы | Описание |
|---|---|
| Записывает один байт (не ) в поток. | |
| Записывает массив байт в поток | |
| Записывает часть массива байт в поток | |
| Записывает в поток все данные, которые хранятся в буфере | |
| Закрывает поток |
При создании объекта класса-наследника обычно указывается объект-источник, из которого читает данные. При создании объекта класса-наследника также обычно указывается целевой объект или целевой поток, в который будут записываться данные.
Вкратце пройдемся по всем методам класса :
Метод
Этот метод записывает в поток вывода один байт (не ). Переданное значение приводится к типу байт, три первые байта отбрасываются.
Метод
Записывает в поток вывода переданный массив байтов. Все.
Метод
Записывает в поток вывода часть переданного массива байтов. Переменная offset задает номер первого элемента массива, — длина записываемого фрагмента.
Метод
Метод используется, чтобы принудительно записать в целевой поток данные, которые могут кэшироваться в текущем потоке. Актуально при использовании буферизации и/или нескольких объектах-потоках, организованных в цепочку.
Метод
Записывает в целевой объект все незаписанные данные. Метод можно не вызывать, если вы используете -with-resources.
Пример — копирование файла
| Код | Примечание |
|---|---|
|
для чтения из файла для записи в файл Буфер, в который мы будем считывать данные Пока данные есть в потоке Считываем данные в буфер Записываем данные из буфера во второй поток |
Java Thread Join(). Теория
. Этот метод приостановит выполнение текущего потока до тех пор, пока другой поток не закончит свое выполнение. Если поток прерывается, бросается .
: Этот метод приостановит выполнение текущего потока на указанное время в миллисекундах. Выполнение этого метода зависит от реализации ОС, поэтому Java не гарантирует, что текущий поток будет ждать указанное вами время.
: Этот метод приостановит выполнение текущего потока до тех пор, пока другой поток не закончит свое выполнение на время заданное в миллисекундах плюс наносекундах.
Вот простой пример, показывающий использование метода . Цель программы: убедиться в том, что третий поток начнет работу только тогда, когда первый закончит выполнение.
Java
package ua.com.prologistic;
public class ThreadJoinExample {
public static void main(String[] args) {
Thread t1 = new Thread(new MyRunnable(), «t1»);
Thread t2 = new Thread(new MyRunnable(), «t2»);
Thread t3 = new Thread(new MyRunnable(), «t3»);
t1.start();
//стартуем второй поток только после 2-секундного ожидания первого потока (или когда он умрет/закончит выполнение)
try {
t1.join(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
t2.start();
//стартуем 3-й поток только после того, как 1 поток закончит свое выполнение
try {
t1.join();
} catch (InterruptedException e) {
e.printStackTrace();
}
t3.start();
//даем всем потокам возможность закончить выполнение перед тем, как программа (главный поток) закончит свое выполнение
try {
t1.join();
t2.join();
t3.join();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(«Все потоки отработали, завершаем программу»);
}
}
class MyRunnable implements Runnable{
@Override
public void run() {
System.out.println(«Поток начал работу:::» + Thread.currentThread().getName());
try {
Thread.sleep(4000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(«Поток отработал:::» + Thread.currentThread().getName());
}
}
|
1 |
packageua.com.prologistic; publicclassThreadJoinExample{ publicstaticvoidmain(Stringargs){ Thread t1=newThread(newMyRunnable(),»t1″); Thread t2=newThread(newMyRunnable(),»t2″); Thread t3=newThread(newMyRunnable(),»t3″); t1.start(); //стартуем второй поток только после 2-секундного ожидания первого потока (или когда он умрет/закончит выполнение) try{ t1.join(2000); }catch(InterruptedExceptione){ e.printStackTrace(); } t2.start(); //стартуем 3-й поток только после того, как 1 поток закончит свое выполнение try{ t1.join(); }catch(InterruptedExceptione){ e.printStackTrace(); } t3.start(); //даем всем потокам возможность закончить выполнение перед тем, как программа (главный поток) закончит свое выполнение try{ t1.join(); t2.join(); t3.join(); }catch(InterruptedExceptione){ e.printStackTrace(); } System.out.println(«Все потоки отработали, завершаем программу»); } } classMyRunnableimplementsRunnable{ @Override publicvoidrun(){ System.out.println(«Поток начал работу:::»+Thread.currentThread().getName()); try{ Thread.sleep(4000); }catch(InterruptedExceptione){ e.printStackTrace(); } System.out.println(«Поток отработал:::»+Thread.currentThread().getName()); } } |
Результат выполнения программы:
Java
Поток начал работу:::t1
Поток начал работу:::t2
Поток отработал:::t1
Поток начал работу:::t3
Поток отработал:::t2
Поток отработал:::t3
Все потоки отработали, завершаем программу
|
1 |
Потокначалработу::t1 Потокначалработу::t2 Потокотработал::t1 Потокначалработу::t3 Потокотработал::t2 Потокотработал::t3 Всепотокиотработали,завершаемпрограмму |
Следите за обновлениями раздела Многопоточность и параллелизм в Java
Прерывание потоков
Для прерывания выполнения нити, если это необходимо, используется метод interrupt(), который устанавливает переменную isInterrupt в значение true. К коде пользовательского класса, унаследованного от Runnable/Thread, это переменная должна проверяться. Отсюда следует, что на самом деле в Java нет возможности прервать поток извне, поток может остановиться только сам.
С другой стороны, в метод sleep() уже встроена проверка переменной isInterrupt, поэтому проверку вручную опускают. Если sleep() считывает наличие прерывания, то генерирует исключение.
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
public class ThreadInterrupt {
public static void main(String args) throws IOException {
Thread thread = new InterruptedClass();
thread.start();
BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
reader.readLine();
thread.interrupt();
}
}
class InterruptedClass extends Thread {
@Override
public void run() {
while (true) {
System.out.print("a");
try {
sleep(1000);
}
catch (InterruptedException e) {
break;
}
}
}
}
В примере основной поток ожидает ввод данных, в это время выполняется вторая нить. Но как только вы нажмете Enter, выполнится метод interrupt(). В свою очередь метод sleep() прочитает значение переменной isInterrupt класса Thread и сгенерирует исключение InterruptedException.
Если sleep() не используется, то isInterrupt проверяется вручную методом isInterrupted(). Следующий пример содержит ошибку, приводящую к зацикливанию:
public class ThreadInterrupt2 {
static int a = 1;
public static void main(String args) throws InterruptedException {
Thread thread = new WithoutSleep();
thread.start();
Thread.sleep(2000);
thread.interrupt();
}
}
class WithoutSleep extends Thread {
@Override
public void run() {
while (true) {
System.out.println("hi");
}
}
}
Мы могли бы ожидать, что через 2 секунды сработает метод interrupt(), который прервет дочернюю нить. Однако, поскольку в ней не проверяется значение isInterrupt, цикл продолжает работать. Корректный код может выглядеть так:
class WithoutSleep extends Thread {
@Override
public void run() {
while (!this.isInterrupted()) {
System.out.println("hi");
}
}
}
При наследовании от Runnable текущий поток через this получить нельзя. Его получают, вызывая соответствующий метод класса Thread:
Как выбирать производителя твердотельных накопителей
Пример создания потока. Наследуем класс Thread
Мы можем наследовать класс для создания собственного класса Thread и переопределить метод . Тогда мы можем создать экземпляр этого класса и вызвать метод для того, чтобы выполнить метод .
Вот простой пример того, как наследоваться от класса Thread:
Java
package ua.com.prologistic;
public class MyThread extends Thread {
public MyThread(String name) {
super(name);
}
@Override
public void run() {
System.out.println(«Стартуем наш поток » + Thread.currentThread().getName());
try {
Thread.sleep(1000);
// для примера будем выполнять обработку базы данных
doDBProcessing();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(«Заканчиваем наш поток » + Thread.currentThread().getName());
}
// метод псевдообработки базы данных
private void doDBProcessing() throws InterruptedException {
Thread.sleep(5000);
}
}
|
1 |
packageua.com.prologistic; publicclassMyThreadextendsThread{ publicMyThread(Stringname){ super(name); } @Override publicvoidrun(){ System.out.println(«Стартуем наш поток «+Thread.currentThread().getName()); try{ Thread.sleep(1000); // для примера будем выполнять обработку базы данных doDBProcessing(); }catch(InterruptedExceptione){ e.printStackTrace(); } System.out.println(«Заканчиваем наш поток «+Thread.currentThread().getName()); } // метод псевдообработки базы данных privatevoiddoDBProcessing()throwsInterruptedException{ Thread.sleep(5000); } } |
Вот тестовая программа, показывающая наш поток в работе:
Java
package ua.com.prologistic;
public class ThreadRunExample {
public static void main(String[] args){
Thread t1 = new Thread(new HeavyWorkRunnable(), «t1»);
Thread t2 = new Thread(new HeavyWorkRunnable(), «t2»);
System.out.println(«Стартуем runnable потоки»);
t1.start();
t2.start();
System.out.println(«Runnable потоки в работе»);
Thread t3 = new MyThread(«t3»);
Thread t4 = new MyThread(«t4»);
System.out.println(«Стартуем наши кастомные потоки»);
t3.start();
t4.start();
System.out.println(«Кастомные потоки в работе»);
}
}
|
1 |
packageua.com.prologistic; publicclassThreadRunExample{ publicstaticvoidmain(Stringargs){ Thread t1=newThread(newHeavyWorkRunnable(),»t1″); Thread t2=newThread(newHeavyWorkRunnable(),»t2″); System.out.println(«Стартуем runnable потоки»); t1.start(); t2.start(); System.out.println(«Runnable потоки в работе»); Thread t3=newMyThread(«t3»); Thread t4=newMyThread(«t4»); System.out.println(«Стартуем наши кастомные потоки»); t3.start(); t4.start(); System.out.println(«Кастомные потоки в работе»); } } |
Конкуренция и параллелизм
найдешь
- Конкуренция — это способ одновременного решения множества задач
- Параллелизм — это способ выполнения разных частей одной задачи
тут
- Наличие нескольких потоков управления (например Thread в Java, корутина в Kotlin), если поток управления один, то конкурентного выполнения быть не может
- Недетерминированный результат выполнения. Результат зависит от случайных событий, реализации и того как была проведена синхронизация. Даже если каждый поток полностью детерминированный, итоговый результат будет недетерминированным
- Необязательно имеет несколько потоков управления
- Может приводить к детерминированному результату, так например, результат умножения каждого элемента массива на число, не изменится, если умножать его по частям параллельно
- битов (например в 32-разрядных машинах сложение происходит в одно действие, параллельно обрабатывая все 4 байта 32-разрядного числа)
- инструкций (на одном ядре, в одном потоке процессор может выполнять инструкции параллельно, несмотря на то что код последовательный)
- данных (существуют архитектуры с параллельной обработкой данных (Single Instruction Multiple Data), способные выполнять одну инструкцию на большом наборе данных)
- задач (подразумевается наличие нескольких процессоров или ядер)
параллельного
1 Потоки данных
Любая программа редко существует сама по себе. Обычно она как-то взаимодействует с «внешним миром». Это может быть считывание данных с клавиатуры, отправка сообщений, загрузка страниц из интернета или, наоборот, загрузка файлов на удалённый сервер.
Все эти вещи мы можем назвать одним словом — процесс обмена данными между программой и внешним миром. Хотя это уже не одно слово.
Сам процесс обмена данными можно разделить на два типа: получение данных и отправка данных. Например, вы считываете данные с клавиатуры с помощью объекта — это получение данных. И выводите данные на экран с помощью команды — это отправка данных.
Для описания процесса обмена данными в программировании используется термин поток. Откуда вообще взялось такое название?
В реальной жизни им может быть поток воды или поток людей (людской поток). В программировании же под потоком подразумевают поток данных.
Потоки — это универсальный инструмент. Они позволяют программе получать данные откуда угодно (входящие потоки) и отправляют данные куда угодно (исходящие потоки). Делятся на два вида:
- Входящий поток (Input): используется для получения данных
- Исходящий поток (Output): используется для отправки данных
Чтобы потоки можно было «потрогать руками», разработчики Java написали два класса: и .
У класса есть метод , который позволяет читать из него данные. А у класса есть метод , который позволяет записывать в него данные. У них есть и другие методы, но об этом после.
Байтовые потоки
Что же это за данные и в каком виде их можно читать? Другими словами, какие типы данных поддерживаются этими классами?
О, это универсальные классы, и поэтому они поддерживают самый распространённый тип данных — . В можно записывать байты (и массивы байт), а из объекта можно читать байты (или массивы байт). Все — никакие другие типы данных они не поддерживают.
Поэтому такие потоки еще называют байтовыми потоками.
Особенность потоков в том, что данные из них можно читать (писать) только последовательно. Вы не можете прочитать данные из середины потока, не прочитав все данные перед ними.
Именно так работает чтение с клавиатуры через класс : вы читаете данные с клавиатуры последовательно: строка за строкой. Прочитали строку, прочитали следующую строку, прочитали следующую строку и т.д. Поэтому метод чтения строки и называется (дословно — «следующая срока»).
Запись данных в поток тоже происходит последовательно. Хороший пример — вывод на экран. Вы выводите строку, за ней еще одну и еще одну. Это последовательный вывод. Вы не можете вывести 1-ю строку, затем 10-ю, а затем вторую. Все данные записываются в поток вывода только последовательно.
Символьные потоки
Недавно вы изучали, что строки — второй по популярности тип данных, и это действительно так. Очень много информации передается в виде символов и целых строк. Компьютер отлично бы передавал все в виде байт, но люди не настолько идеальны.
Java-программисты учли этот факт и написали еще два класса: и . Класс — это аналог класса , только его метод читает не байты, а символы — . Класс соответствует классу , и так же, как и класс , работает с символами (), а не байтами.
Если сравнить эти четыре класса, мы получим такую картину:
| Байты (byte) | Символы (char) | |
|---|---|---|
| Чтение данных | ||
| Запись данных |
Практическое применение
Сами классы , , и в явном виде никто не использует: они не присоединены ни к каким внешним объектам, из которых можно читать данные (или в которые можно писать данные). Однако у этих четырех классов много классов-наследников, которые умеют очень многое.
Методы wait и notify
Последнее обновление: 27.04.2018
Иногда при взаимодействии потоков встает вопрос о извещении одних потоков о действиях других. Например, действия одного потока зависят от результата действий другого потока,
и надо как-то известить один поток, что второй поток произвел некую работу. И для подобных ситуаций у класса Object определено ряд методов:
-
wait(): освобождает монитор и переводит вызывающий поток в состояние ожидания до тех пор, пока другой поток не вызовет метод
-
notify(): продолжает работу потока, у которого ранее был вызван метод
-
notifyAll(): возобновляет работу всех потоков, у которых ранее был вызван метод
Все эти методы вызываются только из синхронизированного контекста — синхронизированного блока или метода.
Рассмотрим, как мы можем использовать эти методы. Возьмем стандартную задачу из прошлой темы — «Производитель-Потребитель» («Producer-Consumer»):
пока производитель не произвел продукт, потребитель не может его купить. Пусть производитель должен произвести 5 товаров, соответственно потребитель
должен их все купить. Но при этом одновременно на складе может находиться не более 3 товаров.
Для решения этой задачи задействуем методы и :
public class Program {
public static void main(String[] args) {
Store store=new Store();
Producer producer = new Producer(store);
Consumer consumer = new Consumer(store);
new Thread(producer).start();
new Thread(consumer).start();
}
}
// Класс Магазин, хранящий произведенные товары
class Store{
private int product=0;
public synchronized void get() {
while (product<1) {
try {
wait();
}
catch (InterruptedException e) {
}
}
product--;
System.out.println("Покупатель купил 1 товар");
System.out.println("Товаров на складе: " + product);
notify();
}
public synchronized void put() {
while (product>=3) {
try {
wait();
}
catch (InterruptedException e) {
}
}
product++;
System.out.println("Производитель добавил 1 товар");
System.out.println("Товаров на складе: " + product);
notify();
}
}
// класс Производитель
class Producer implements Runnable{
Store store;
Producer(Store store){
this.store=store;
}
public void run(){
for (int i = 1; i < 6; i++) {
store.put();
}
}
}
// Класс Потребитель
class Consumer implements Runnable{
Store store;
Consumer(Store store){
this.store=store;
}
public void run(){
for (int i = 1; i < 6; i++) {
store.get();
}
}
}
Итак, здесь определен класс магазина, потребителя и покупателя. Производитель в методе добавляет в объект Store с помощью его метода
6 товаров. Потребитель в методе в цикле обращается к методу объекта Store для получения
этих товаров. Оба метода Store — и являются синхронизированными.
Для отслеживания наличия товаров в классе Store проверяем значение переменной . По умолчанию товара нет, поэтому переменная равна .
Метод — получение товара должен срабатывать только при наличии хотя бы одного товара. Поэтому в методе
проверяем, отсутствует ли товар:
while (product<1)
Если товар отсутсвует, вызывается метод . Этот метод освобождает монитор объекта Store и блокирует выполнение метода get, пока для этого же монитора не будет вызван
метод .
Когда в методе добавляется товар и вызывается , то метод получает монитор и выходит из
конструкции , так как товар добавлен. Затем имитируется получение покупателем товара. Для этого
выводится сообщение, и уменьшается значение product: . И в конце вызов метода дает сигнал методу продолжить работу.
В методе работает похожая логика, только теперь метод должен срабатывать, если в магазине не более трех товаров. Поэтому в цикле проверяется наличие товара, и если товар уже есть,
то освобождаем монитор с помощью и ждем вызова в методе .
И теперь программа покажет нам другие результаты:
Производитель добавил 1 товар Товаров на складе: 1 Производитель добавил 1 товар Товаров на складе: 2 Производитель добавил 1 товар Товаров на складе: 3 Покупатель купил 1 товар Товаров на складе: 2 Покупатель купил 1 товар Товаров на складе: 1 Покупатель купил 1 товар Товаров на складе: 0 Производитель добавил 1 товар Товаров на складе: 1 Производитель добавил 1 товар Товаров на складе: 2 Покупатель купил 1 товар Товаров на складе: 1 Покупатель купил 1 товар Товаров на складе: 0
Таким образом, с помощью в методе мы ожидаем, когда производитель добавит новый продукт. А после добавления
вызываем , как бы говоря, что на складе освободилось одно место, и можно еще добавлять.
А в методе с помощью мы ожидаем освобождения места на складе. После того, как место освободится, добавляем товар и
через уведомляем покупателя о том, что он может забирать товар.
НазадВперед
Новые возможности пакета java.uti.concurrent
Платформа Java постоянно развивается, и поэтому к существующей функциональности все время добавляются новые возможности. Иногда новая функциональность берется из уже существующих сторонних библиотек, при этом речь не идет о банальном копировании, а скорее о переосмыслении и доработке уже существующих решений. Подобным способом в версию Java 5 был добавлен пакет java.util.concurrent, включающий в себя множество уже проверенных и хорошо зарекомендовавших себя приемов для параллельного выполнения задач (этот пакет — только одно из множества важных нововведений, представленных в Java 5).
В рамках этой статьи интерес представляют уже готовые к использованию реализации шаблонов WorkerThread и ThreadPool, а также еще один способ реализации задач для параллельного выполнения, кроме упоминавшихся класса Thread и интерфейса Runnable. Ещё в пакете java.util.concurrent находятся два подпакета: java.util.concurrent.locks и java.util.concurrent.atomic, с которыми тоже стоит ознакомиться, так как они значительно упрощают организацию взаимодействия между потоками и параллельного доступа к данным.
1 Нововведения в Java 8: Функциональное программирование
Вместе с выходом Java 8 в ней появилась мощная поддержка функционального программирования. Можно даже сказать, долгожданная поддержка функционального программирования. Код стал писаться быстрее, хотя читать его стало сложнее
Перед изучением функционального программирования в Java, рекомендуем хорошо разобраться в трех вещах:
- ООП, наследование и интерфейсы (1-2 уровни квеста Java Core).
- Дефолтная реализация методов в интерфейсе.
- Внутренние и анонимные классы.
Хорошая новость заключается в том, что без знания всего этого можно пользоваться многими возможностями функционального программирования в Java. Плохая новость — понять, как именно все устроено и как все работает, без тех же внутренних анонимных классов уже сложно.
В ближайших лекциях мы сосредоточимся на том, как легко и просто пользоваться возможностями функционального программирования в Java, без глубокого понимания, как оно устроено.
Чтобы разобраться во всех нюансах функционального программирования в Java, нужны месяцы. Читать же такой код можно научиться за несколько часов. Поэтому предлагаем начать с малого. Да хоть с тех же потоков ввода-вывода.
Настройка размера пула
Настраивая размер пула потоков, важно избежать двух ошибок: слишком мало потоков или слишком много потоков. К счастью, для большинства приложений спектр между слишком большим и слишком малым количеством потоков довольно широк
Если вы помните, есть два основных преимущества в организации поточной обработки сообщений в приложениях: возможность продолжения процесса во время ожидания медленных операций, таких, как I/O (ввод — вывод), и использование возможностей нескольких процессоров. В приложениях с ограничением по скорости вычислений, функционирующих на N-процессорной машине, добавление дополнительных потоков может улучшить пропускную способность, по мере того как количество потоков подходит к N, но добавление дополнительных потоков свыше N не оправдано. Действительно, слишком много потоков разрушают качество функционирования из-за дополнительных издержек переключения процессов
Оптимальный размер пула потоков зависит от количества доступных процессоров и природы задач в рабочей очереди. На N-процессорной системе для рабочей очереди, которая будет выполнять исключительно задачи с ограничением по скорости вычислений, вы достигните максимального использования CPU с пулом потоков, в котором содержится N или N+1 поток.
Для задач, которые могут ждать осуществления I/O (ввода — вывода) — например, задачи, считывающей HTTP-запрос из сокета – вам может понадобиться увеличение размера пула свыше количества доступных процессоров, потому, что не все потоки будут работать все время. Используя профилирование, вы можете оценить отношение времени ожидания (WT) ко времени обработки (ST) для типичного запроса. Если назвать это соотношение WT/ST, для N-процессорной системе вам понадобится примерно N*(1+WT/ST) потоков для полной загруженности процессоров.
Использование процессора – не единственный фактор, важный при настройке размера пула потоков. По мере возрастания пула потоков, можно столкнуться с ограничениями планировщика, доступной памяти, или других системных ресурсов, таких, как количество сокетов, дескрипторы открытого файла, или каналы связи базы данных
Запуск задач с помощью java.util.concurrent.ExecutorService
Облегчив с помощью интерфейса Callable создание задач для параллельного выполнения, пакет java.util.concurrent также берет на себя работу по запуску и остановке потоков. Вместо объекта Thread предлагается использовать объект типа ExecutorService, с помощью которого пользователь может просто поместить задачу в очередь на выполнение и ждать получения результата. Можно сказать, что ExecutorService – это значительно усовершенствованная реализация шаблона WorkerThread.
ExecutorService – это интерфейс, поэтому для выполнения задач используются его конкретные потомки, адаптированные под требования разрабатываемого приложения. Однако программисту нет необходимости создавать собственную реализацию ExecutorService, так как в пакете java.util.concurrent уже присутствуют различные варианты реализации ExecutorService. Доступ к ним можно получить через статические методы служебного класса Executors, метод которого newFixedThreadPool возвращает объект типа ExecutorService со встроенной поддержкой шаблона ThreadPool. Также в классе Executors есть и другие методы для создания объектов ExecutorService с различными свойствами.
Наибольший интерес в ExecutorService представляет метод submit, через который задача ставится в очередь на выполнение. На вход этот метод принимает объект типа Callable или Runnable, а возвращает некий параметризованный объект типа Future. Этот объект можно использовать для доступа к результату выполнения задачи, который будет возвращен из метода call соответствующего Callable-объекта. При этом через объект Future можно проверить, закончено ли уже выполнение задачи – с помощью метода isDone и через метод get получить доступ к результату или исключительной ситуации, если в процессе выполнения задачи произошла ошибка.
Таким образом, при запуске задач с помощью классов из пакета java.util.concurrent не требуется прибегать к низкоуровневой поточной функциональности класса Thread, достаточно создать объект типа ExecutorService с нужными свойствами и передать ему на исполнение задачу типа Callable. Впоследствии можно легко просмотреть результат выполнения этой задачи с помощью объекта Future, как показано в листинге 4.
Листинг 4. Запуск задачи с помощью классов пакета java.util.concurrent
1 public class ExecutorServiceSample {
2 public static void main(String[] args) {
3 //создать ExecutorService на базе пула из пяти потоков
4 ExecutorService es1 = Executors.newFixedThreadPool(5);
5 //поместить задачу в очередь на выполнение
6 Future<String> f1 = es1.submit(new CallableSample());
7 while(!f1.isDone()) {
8 //подождать пока задача не выполнится
9 }
10 try {
11 //получить результат выполнения задачи
12 System.out.println("task has been completed : " + f1.get());
13 } catch (InterruptedException ie) {
14 ie.printStackTrace(System.err);
15 } catch (ExecutionException ee) {
16 ee.printStackTrace(System.err);
17 }
18 es1.shutdown();
19 }
20}
Стоит обратить внимание на строку 18, где происходит остановка объекта ExecutorService с помощью метода shutdown. Дело в том, что потоки в объекте ExecutorService не останавливаются сами, как обычно, поэтому их необходимо явно остановить с помощью этого метода, при этом если в ExecutorService находятся невыполненные задачи, то потоки будут остановлены только, когда завершится последняя задача
