Archives: Java Lessons

fork/join framework – для разбиения задачи на подзадачи и чтобы эти подзадачи выполнялись в отдельных потоках.

Также эти подзадачи могут тоже делиться на подзадачи и опять же чтобы эти подзадачи выполнялись в отдельных потоках.

Есть пул потоков. В один из его потоков ставиться большая задача.

Эта задача и подзадачи в будущем являют собой выполнения метода compile, который нужно переопределить

Деление на подзадачи в else происходит особым рекурсивным образом с помощью методов fork и join.

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

join ждет пока эта форкнутая подзадача выполнится.

join возвращает значение из compile, который fork вызвал.

Ясное дело, compile будет вызываться рекурсивно форком и соответственно джойны будут копиться пока не дойдет до наименьшей подзадачи.

И джойны, понятное дело, в обратном направлении начнут освобождаться.

Помещенные задачи форком в любом потоке в пул потоков могут быть взяты любым свободным потоком для выполнения.

Пример программы:

import java.util.concurrent.ExecutionException; import java.util.concurrent.ForkJoinPool; import java.util.concurrent.RecursiveTask; public class ForkJoinExample { static volatile int j; public static void main(final String[] arguments) throws InterruptedException, ExecutionException { //узнать количество ядер компьютера //(в каждом ядре эффективно выполнять 1 поток) int nThreads = Runtime.getRuntime().availableProcessors(); System.out.println(nThreads); int[] numbers = new int[1000]; //запишем в массив числа от 1 до 1000 for(int i = 0; i < numbers.length; i++) { numbers[i] = i+1; } //Будем складывать числа от 1 до 1000 //эффективным образом с помощью fork/join // сюда передаем количество // ядер (размер пула потоков) ForkJoinPool forkJoinPool = new ForkJoinPool(nThreads); // invoke запускает задачу на выполнение // то есть вызывает compile. Long result = forkJoinPool.invoke( new Sum(numbers,0,numbers.length)); //пока invoke не вернул значение дальше //в методе мейн ничего не выполняется System.out.println(result); } static class Sum extends RecursiveTask< Long > { int low; int high; int[] array; Sum(int[] array, int low, int high) { this.array = array; this.low = low; this.high = high; } protected Long compute() { //if задача или подзадача уже необходимо мала //возвращаем что-то из метода compile if(high – low <= 10) { long sum = 0; for(int i = low; i < high; ++i) sum += array[i]; return sum; } //else - делим задачу или подзадачу //на подзадачи (в данном случае на 2 половины) else { int mid = low + (high - low) / 2; Sum left = new Sum(array, low, mid); Sum right = new Sum(array, mid, high); // добавляем одну половину задачи (подзадачу) // в пул потоков left.fork(); // добавляем вторую половину задачи (подзадачу) // в пул потоков right.fork(); // ждем когда compile в форке завершиться long leftResult = left.join(); // ждем когда compile в форке завершиться long rightResult = right.join(); return leftResult + rightResult; } } } }

Вывод:

Как можно увидеть, в консоли сумма чисел от 1 до 1000 посчитана верна – 500500.

ThreadLocal – переменная ПОТОКА. Это не переменная объекта потока, а именно переменная ПОТОКА. То есть она принадлежит только одному потоку и существует только в одном потоке.

В примере ниже мы создаем объект потока threadDemo, который передаем на выполнение в три потока. То есть три потока будут работать с одним и тем же объектом.

Если переменная НЕ ThreadLocal и эта переменная в этом объекте меняется каким-то одним потоком, то эта переменная будет измененной и в других потоках, так как они работают с одним и тем же объектом.

Пример программы:

class SomeThread implements Runnable { // локальная переменная объекта класса int counter; // локальная переменная потока ThreadLocal threadLocalCounter = new ThreadLocal<>(); // то есть, как видим, есть переменные ОБЪЕКТА, // а есть переменные ПОТОКА. // Смотри более подробно после кода public void run() { counter++; if (threadLocalCounter.get() != null) { // запись в переменную ThreadLocal // через метод set threadLocalCounter.set(threadLocalCounter.get() + 1); } else { threadLocalCounter.set(0); } System.out.println(“Counter: ” + counter); System.out.println(“threadLocalCounter: ” + threadLocalCounter.get()); } } class ThreadLocalExample { public static void main(String[] args) { SomeThread threadDemo = new SomeThread(); Thread t1 = new Thread(threadDemo); Thread t2 = new Thread(threadDemo); Thread t3 = new Thread(threadDemo); t1.start(); try { Thread.sleep(200); } catch (InterruptedException e) {} t2.start(); try { Thread.sleep(200); } catch (InterruptedException e) {} t3.start(); } }

Вывод:

Как можно увидеть в консоли, переменную counter меняли все три потока, а переменная threadLocalCounter была у каждого потока своя личная и в каждом из потоков она только успела стать 0.

Как можно увидеть ThreadLocal иногда не заменим, например в этом случае, когда создать локальную переменную для потока можно только с помощью ThreadLocal, потому что объект один, это threadDemo и все потоки работают с ним, и соответственно, например, локальная переменная counter для всех потоков общая и все потоки изменяют ее в этом объекте, а чтобы создать личную для каждого потока работающего с объектом threadDemo необходимо использовать ThreadLocal.

---

ThreadLocal и потоконебезопасные обьекты

Объекты некоторых классов (например SimpleDateFormat) потокоНЕбезопасны.

То есть если к одному и тому же объекту SimpleDateFormat одновременно обращаются несколько потоков, то нужно применять синхронизацию в потоках (используя synchronized например), иначе будет ошибка. Но синхр-я тормозит работу программы. Поможет ThreadLocal.

То есть если объект реализует Runnable, и он будет использоваться многими потоками, и в этом объекте есть потокоНЕбезопасный объект, лучше не использовать синхронизацию, а размножить его на все потоки используя ThreadLocal.

То есть с помощью ThreadLocal можно сделать так, что у каждого потока будет свой объект SimpleDateFormat.

ReadWriteLock – класс содержащий 2 лока: один для чтения, другой для записи.

То есть один поток блокирует ресурс для записи, а другие потоки, желающие читать из ресурса, ждут пока он запишет, и после того, как записывающий закончил запись, потоки читают, и далее всё повторяется.

Пример программы:

import java.io.*; import java.util.*; import java.util.concurrent.locks.ReadWriteLock; import java.util.concurrent.locks.ReentrantReadWriteLock; public class RedWrtLockExample { static long numOfOperations=10000000000L; public static void main(String[] args) { CommonResource commonResource = new CommonResource(); //объект ресурса for (int i = 1; i < 4; i++){ //Запускаем 3 потока и передаем в каждый из них //ресурс общий для потоков. Thread t = new Thread(new CountThread(commonResource)); t.setName("Thread " + i); t.start(); } //5 раз записать 10 млрд в commonResource.value for (int i = 0; i < 5; i++) { //Здесь начинается запись в ресурс потоком мейн //и пока происходит запись в ресурс потоком мейн //другие потоки читать из ресурса не смогут. commonResource.write(); try{ Thread.sleep(100); } catch(InterruptedException e){} } } static class CommonResource { long value; private final ReadWriteLock lock = new ReentrantReadWriteLock(); void write() { //здесь происходит writeLock().lock() на время //записи десяти млрд блокируем все другие потоки, //которые дошли до readLock().lock() в этом объекте lock.writeLock().lock(); for (long i = 0; i < numOfOperations; i++){ value++; } //запускаются залоченные в read lock.writeLock().unlock(); } void read() { //Если в каком то из потоков в этом объекте //случился writeLock().lock() и пока еще не случился //writeLock().unlock() то readLock().lock() //останавливает потоки, которые дошли //до readLock().lock() в этом объекте ресурса. lock.readLock().lock(); System.out.println("Counter: " + value); lock.readLock().unlock(); } } static class CountThread implements Runnable { CommonResource res; CountThread(CommonResource res){ this.res=res; } public void run(){ for (int i = 0; i < 5; i++) { try { Thread.sleep(200); } catch(InterruptedException e){} //Пока объект заблокирован записью //чтение потоками не будет происходить. res.read(); } } } }

Вывод:

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

Создание потока дорогостоящая операция.

ExecutorService помогает поддерживать пул потоков, то есть поддерживает выполнение некоторого фиксированного количества потоков, которые одновременно выполняются.

Также он назначает задачи этим потокам в этом пуле (в пуле всё время находятся те же самые потоки, это важно).

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

Как уже было сказано ExecutorService создает некоторый пул потоков.

То есть потоки в пуле потоков не создаются заново, а переиспользуются.

То есть представим, что чтобы обработать один запрос клиента сервер создает отдельный поток. Когда серверной программой обрабатываются запросы клиентов, из-за большого количества потоков может увеличиваться время отклика (промежуток времени, который требуется серверу, чтобы обработать запрос извне). То есть требуется дополнительное время для постоянного создания нитей на сервере, что приводит к увеличению времени отклика и чтобы постоянно не создавались потоки на сервере дла обработки запросов, ExecutorService может помочь. То есть запросы будут выполняться в ExecutorService, в котором потоки переиспользуются для обработки запросов клиентов, и если в ExecutorService нет места, запросы становиться в очередь и будут ждать пока освободиться поток в ExecutorService.

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

Пример программы:

import java.io.*; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; class ServiceExecutorExample { //к этой переменной добавляется //в потоках SomeThread static volatile int j; public static void main(String[] args) { ExecutorService exeServ = //одновременно будет работать //только 10 потоков Executors.newFixedThreadPool(10); for (int i = 0; i < 1000; i++) { //в этом цикле 10 потоков будет //переполнены 100 раз. //Запустить поток в ExecutorService. exeServ.submit(new SomeThread()); //Если бы мы запускали здесь потоки //стандартным образом то создалось //бы 1000 потоков. } try { Thread.sleep(1000); } catch (InterruptedException e){} //завершить ExecutorService exeServ.shutdown(); System.out.println("Counter: " + j); } static class SomeThread implements Runnable{ @Override public void run(){ try{ for (int i = 0; i < 100; i++){ j++; } } catch(Exception e){} } } }

Вывод:

То есть ExecutorService нам нужен чтобы некоторое количество потоков, некоторое продолжительное время переиспользовалось для выполнения каких-либо задач.

До этого мы рассматривали создание потоков с помощью Thread и Runnable. Последний вариант того как можно создать класс потока это реализовать Callable.

FutureTask – для взаимодействия с потоками. Он для получения результата выполнения потока (того, что вернет call), еще имеет методы проверки состояния потока.

С помощью метода get() класса FutureTask можно получить результат выполнения потока когда он завершит свое выполнение.

Из также важных методов – с помощью isDone() можем проверить завершился ли уже поток или еще нет.

Пример программы:

import java.io.*; import java.util.*; import java.util.concurrent.Callable; import java.util.concurrent.FutureTask; class CallableFutureExample { public static void main(String[] args) throws Exception { FutureTask futTask = new FutureTask(new MyCallable()); Thread t = new Thread(futTask); t.start(); //main останавливается и get ждет пока //нить t не выполнится и после этого //возвращает значение метода call. System.out.println(“Counter: ” + futTask.get()); } } class MyCallable implements Callable { public Integer call() throws Exception { Integer j=0; try { for (int i = 0; i < 5; i++) { j++; System.out.println("j = " + j); Thread.sleep(1000); } } catch(Exception e) { return j; //возвращаем j, который //будет помещен в FutureTask } return j; } }

Вывод:

Как видим, get остановил мейн пока выполнялся поток t и когда t выполнился, мейн снова запустился и вывел Counter: 5

Чтобы резко остановить поток можно использовать метод Stop(), но резко останавливать не рекомендуется, это может быть опасно.

Используют более аккуратный способ interrupt() – этот метод не останавливает поток, а устанавливает статус потока, как прерван, но САМ ПОТОК НЕ ОСТАНАВЛИВАЕТСЯ. Статус потока теперь можно будет проверить в этом потоке в нужном нам месте с помощью метода isInterrupted() и исходя из того, что он вернет завершить выполнение каких-то действий.

Пример программы:

class SomeThread extends Thread { SomeThread(String name){ super(name); } public void run(){ System.out.printf(“%s started… \n”, Thread.currentThread().getName()); int counter=1; // счетчик циклов while(!isInterrupted()){ //если возвращает true //то цикл останавливается System.out.println(“Loop ” + counter++); } //то есть получается блок while работает //пока не придет сигнал его остановить //методом interrupt и в этом вся суть System.out.printf(“%s finished… \n”, Thread.currentThread().getName()); } } public class InterruptExample { public static void main(String[] args) { System.out.println(“Main thread started…”); SomeThread t = new SomeThread(“SomeThread”); t.start(); try{ Thread.sleep(150); //до вызова interrupt метод isInterrupted //в потоке возвращает false t.interrupt();//теперь статус потока t – прерван //и теперь в потоке t вызов метода isInterrupted //должен вернуть true Thread.sleep(150); } catch(InterruptedException e){ System.out.println(“Thread has been interrupted”); } System.out.println(“Main thread finished…”); } }

Вывод:

…

Из консоли видно, что цикл выполнялся в потоке пока он не получил статус прерван из main.

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

В примере ниже мы создаем переменную volatile и потоки добавляют к ней 1, потом в месте await они останавливаются и когда зависнет указанное в CyclicBarrier количество потоков потоки развисают и сразу после этого запускается метод run в классе Run. И в этом методе run класса Run будет использована эта ранее изменяемая зависающими потоками переменная.

Пример программы:

import java.io.*; import java.util.concurrent.CyclicBarrier; class CyclicBarrierExample { //к этой переменной добавляется 1 //в потоках SomeThread static volatile int j; public static void main(String[] args) { //Только когда на await зависнут три потока все //зависнувшие потоки развиснут. //и метод run в классе Run запуститься только //когда все три потока дойдут до await CyclicBarrier cycbar = new CyclicBarrier(6, new Run()); for (int i = 1; i < 7; i++){ SomeThread t = new SomeThread(); t.setName("Thread "+ i); t.cycbar = cycbar; t.start(); } } //здесь в Run выводиться результирующее j static class Run extends Thread { @Override public void run(){ System.out.println(j); } } static class SomeThread extends Thread{ CyclicBarrier cycbar; @Override public void run(){ try{ j++; System.out.println( Thread.currentThread().getName()+" "+j); cycbar.await(); //здесь поток зависает System.out.println( Thread.currentThread().getName() + " is alive again"); } catch(Exception e){} } } }

Вывод:

Как видим шесть потоков остановили свою работу, потом запустился поток Run и вывел 6 и потом потоки продолжили работу.