Archives: Java Lessons

CountDownLatch – нужен когда требуется чтобы какой-то поток ждал пока не выполняться какое-то количество потоков.

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

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

import java.io.*; import java.util.concurrent.CountDownLatch; public class CountDownLatchExample { public static void main(String[] args) { int sleeptime = 0; CountDownLatch resources = new CountDownLatch(3); //запускаем 4 потока. каждый поток будет //уменьшать на единицу число, которое указано //в CountDownLatch. у нас 3. for (int i = 1; i < 11; i++) { SomeThread t = new SomeThread(); t.setName("Thread " + i); t.resources = resources; sleeptime=200; t.sleeptime = sleeptime; t.start(); } try { System.out.println( Thread.currentThread().getName()+" is stopped"); //этот текущий поток (то есть поток мейна) //будет остановлен пока счетчик resources //не будет ноль resources.await(); } catch(InterruptedException e){} System.out.println( Thread.currentThread().getName()+" is alive again"); } } class SomeThread extends Thread{ int sleeptime; CountDownLatch resources; @Override public void run() { try{ Thread.sleep(sleeptime); } catch (InterruptedException e){} //поток выполнился. уменьшаем счетчик на 1 resources.countDown(); System.out.println( Thread.currentThread().getName()+" finished working"); } }

Вывод:

Из консоли видно, что три потока завершили свое выполнение и сразу после этого ожил мейн.

ReentrantLock ведет себя как synchronized, но есть некоторые отличия.

Мы помним, что synchronized блочит цельный кусок кода, то есть synchronized, это цельный метод, который не может быть выполняем параллельно несколькими потоками.

То есть если synchronized лочит для других потоков цельный метод то ReentrantLock например может залочить для других потоков часть одного метода и часть другого метода.

Лучше пояснить на примере.

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

import java.io.*; import java.util.*; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; public class ReentrantLockExample { public static void main(String[] args) { CommonResource commonResource = new CommonResource(); //объект ресурса for (int i = 1; i < 6; i++) { //запускаем 5 потоков передаем в каждый //из них ресурс общий для потоков Thread t = new Thread(new CountThread(commonResource)); t.setName("Thread " + i); t.start(); } } } class CommonResource { int x; Lock lock = new ReentrantLock(); void increment() { //Заблокируем кусок кода, который состоит //из части метода increment и части метода //increment1. То есть очевидно, что это не //цельный кусок кода, а из ранных частей нашего //класса. При synchronized блокируется весь код, //который содержится в методе от первой строки //в нем до последней. Здесь же достаем часть //метода increment, которая после lock.lock() //и часть метода increment1, которая до lock.unlock(), склеиваем эти две части и //блокируем этот склеиный кусок кода как цельный. //То есть очевидно, что все что было в методе increment до вызова lock.lock() //(хотя я данном случае, как видим, там ничего нет) не блокируется, блокируется кусок //посреди метода, а не на весь метод. lock.lock(); x=1; for (int i = 1; i < 5; i++) { System.out.printf("%s %d \n", Thread.currentThread().getName(), x); x++; try { Thread.sleep(300); } catch (InterruptedException e) {} } lock.unlock(); } void increment1(){ x=1; for (int i = 1; i < 5; i++){ System.out.printf("%s %d \n", Thread.currentThread().getName(), x); x++; try{ Thread.sleep(300); } catch(InterruptedException e){} } lock.unlock(); //разблокируем кусок кода //объекта CommonResource который был //заблокирован в другом методе объекта. //то есть в отличии от synchronized //можно залочить и разлочить //НЕЦЕЛЫЙ кусок кода объекта, то есть //в разных частях объекта. } } class CountThread implements Runnable{ CommonResource res; CountThread(CommonResource res){ this.res=res; } public void run(){ //Лочиться в методе increment res.increment(); //Разлочивается в методе increment1 res.increment1(); //то есть как уже говорилось лок //и разлок нецельного куска //кода в любом месте объекта } }

Вывод:

Видим, 5 потоков выполняются по очереди.

Также видим, что каждый из 5 потоков выводит 8 значений. То есть оба цикла в объекте попадают в тот самый не цельный кусок кода, который не может быть выполняем параллельно несколькими потоками благодаря Reentrantlock.

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

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

import java.io.*; import java.util.concurrent.Semaphore; public class SemaphoreExample { public static void main(String[] args) { Semaphore resources = new Semaphore(3); // три разрешения // То есть одновременно могут выполняться только // три потока, все остальные стоят в очереди пока // одно из трех мест не освободится. for (int i = 1; i < 10; i++) { SomeThread t = new SomeThread(); t.setName("Thread " + i); t.resources = resources; t.start(); } } } class SomeThread extends Thread { Semaphore resources; @Override public void run() { try { // Поток занимает одно из трех разрешений resources.acquire(); System.out.println( Thread.currentThread().getName() + " entered the resource" ); Thread.sleep(1000); System.out.println( Thread.currentThread().getName() + " left the resource" ); resources.release(); // Поток покидает одно // из трех разрешений, после этого оставленное // разрешение сразу занимает другой поток, // стоявший в очереди на занятие разрешения, // поскольку не мог занять разрешение, так как // все три были заняты другими потоками. } catch (InterruptedException e) {} } }

Вывод:

Из консоли видно, что одновременно выполняются только три потока.

То есть, очевидно, что параллельно друг другу всё время работают только три потока.

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

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

class YieldExample { public static void main(String[] args) { new SomeThread().start(); new SomeThread().start(); new SomeThread().start(); } } class SomeThread extends Thread { public void run() { System.out.println(Thread.currentThread().getName() + ” уступает свое место другим потокам”); Thread.yield(); System.out.println(Thread.currentThread().getName() + ” завершился”); } }

Вывод:

Мы запускаем три потока. Планировщик потоков может запустить эти потоки в разном порядке (например, 2-1-0 или 1-0-2)

На консоли мы видим именно такую последовательность выполнения.

wait/notify – используется если нам нужно приостановить один поток и чтобы он ждал пока другой поток даст разрешение на продолжение выполнения остановленного потока.

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

class DemoClass { synchronized void part1() { System.out.println(“Thread t1 started”); //с помощью notify даем разрешение на //продолжение работы потока, который был //остановлен с помощью wait. notify(); System.out.println( “Thread t2 now is not locked by wait method anymore”); for(int i=0;i<15;i++) { System.out.println("Thread t1 is working now..."); } System.out.println("Thread t1 finished his work."); } //wait/notify - вызываются в блоках synchronized //над одним и тем же объектом в данном случае //над объектом DemoClass synchronized void part2() { try { System.out.println("Thread t2 started"); System.out.println("Thread t2 waiting"); //с помощью wait останавливаем поток, который //зашел в этом synchronized метод и теперь //другие потоки могут заходить в другие //synchronized методы для того чтобы в них //с помощью notify дать разрешение на //продолжение работы потока, который //был заблокирован здесь с помощью wait. wait(); System.out.println( "Thread t2 running again"); } catch (Exception e) { System.out.println(e.getClass()); } System.out.println("Thread t2 finished his work."); } } public class WaitNotifyExample { public static void main(String[] args) { DemoClass obj = new DemoClass(); //реализуем run с помощью анонимного класса. //чтобы не создавать два полноценных класса Thread t1 = new Thread(new Runnable() { public void run() { obj.part1(); } }); Thread t2 = new Thread(new Runnable() { public void run() { obj.part2(); } }); t2.start(); try { Thread.sleep(100); } catch(InterruptedException e){} t1.start(); } }

Вывод:

synchronized static (блокировка на уровне класса), а просто synchronized (блокировка на уровне объекта).

В коде ниже у нас уже будут 2 объекта ресурса одного класса. В них уже будет synchronized static блок.

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

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

public class StaticSinchExample { public static void main(String[] args) { // объект ресурса номер 1 CommonResource commonResource1 = new CommonResource(); // объект ресурса номер 2 CommonResource commonResource2 = new CommonResource(); for (int i = 1; i < 6; i++) { // запускаем 5 потоков, передаем в каждый из них // первый ресурс, общий для 5 потоков Thread t1 = new Thread(new CountThread(commonResource1)); // запускаем 5 потоков, передаем в каждый из них // второй ресурс, общий для 5 потоков Thread t2 = new Thread(new CountThread(commonResource2)); t1.setName("Thread t1_" + i); t2.setName("Thread t2_" + i); t1.start(); t2.start(); } // То есть, если один из 10 запускаемых выше потоков // дойдет до блока synchronized static в каком-либо // из объектов ресурсов, будь то commonResource1 // или commonResource2, то блок synchronized static // будет заблокирован для всех остальных потоков // во всех объектах ресурсов. } } class CommonResource { int x; static int x1; synchronized static void incrementstatic() { // Верхнюю строчку можно было бы переписать // как synchronized (CommonResource.class) – // блокирует для остальных потоков блок кода // класса, а не блок кода конкретного объекта. x1 = 1; for (int i = 1; i < 5; i++) { System.out.printf("Static %s %d \n", Thread.currentThread().getName(), x1); x1++; try { Thread.sleep(1100); } catch (InterruptedException e) {} } } synchronized void increment() { // Верхнюю строчку можно было бы переписать // как synchronized (this) – блокирует для // остальных потоков блок кода // конкретного объекта. x = 1; for (int i = 1; i < 5; i++) { System.out.printf("%s %d \n", Thread.currentThread().getName(), x); x++; try { Thread.sleep(100); } catch (InterruptedException e) {} } } } class CountThread implements Runnable { CommonResource res; CountThread(CommonResource res) { this.res = res; } public void run() { CommonResource.incrementstatic(); } }

Вывод:

Как видно, все 10 потоков выводятся по очереди.

То есть 4 значения одного потока из 10 потоков, потом 4 значения другого из 10 потоков и так до десятого потока.

Что значит, что блокировка synchronized static распространяется на все объекты ресурсов.

---

Детали о разнице synchronized static и просто synchronized.

synchronized static (на уровне класса) и просто synchronized (на уровне объекта) это ДВЕ РАЗНЫЕ БЛОКИРОВКИ. Могут выполняться параллельно друг другу.

Один sinchronized метод блокирует доступ и к другим synchronized методам в объекте. То есть если поток зашел в один synchronized метод объекта, то заблокирован для других потоков не только этот synchronized метод в этом объекте, заблокированы для других потоков и другие synchronized методы в этом объекте пока поток не покинет synchronized метод в объекте.

Если же один поток зашел в просто synchronized метод (не static) в объекте, то другие потоки могут спокойно заходить в synchronized static методы в этом объекте не дожидаясь пока поток, который зашел в просто synchronized выйдет оттуда.

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

public class StaticSinchExample2 { public static void main(String[] args) { //Объект ресурса CommonResource commonResource = new CommonResource(); for (int i = 1; i < 6; i++) { //запускаем 5 потоков и передаем //в каждый из них ресурс общий для потоков Thread t = new Thread(new CountThread(commonResource)); t.setName("Thread " + i); t.start(); } } } class CommonResource { int x; static int x1; synchronized void increment() { x = 1; for (int i = 1; i < 5; i++) { System.out.printf("%s %d \n", Thread.currentThread().getName(), x); x++; try { Thread.sleep(250); } catch (InterruptedException e) {} } } synchronized static void incrementstatic() { x1 = 1; for (int i = 1; i < 5; i++) { System.out.printf("Static %s %d \n", Thread.currentThread().getName(), x1); x1++; try { Thread.sleep(700); } catch (InterruptedException e) {} } } } //Код synchronized метода и synchronized static метода //не должны пересекаться так как блоки synchronized static //и просто synchronized могут выполняться параллельно, //а если они могут выполняться параллельно значит //они не синхронизированы между собой. class CountThread implements Runnable { CommonResource res; CountThread(CommonResource res) { this.res = res; } public void run() { //один поток может выполнять incrementstatic метод CommonResource.incrementstatic(); //другой поток может в это же время //параллельно выполнять increment res.increment(); } }

Вывод:

Как видно, после выхода Thread 1 из incrementstatic метода Thread 1 входит в increment метод и выполнение этого метода, как видно, происходит параллельно потоку Thread 5, который зашел в incrementstatic метод как только из него вышел Thread 1.

То есть очевидно synchronized static блок и synchronized блок выполняются параллельно друг другу.

Если несколько параллельных потоков одновременно хотят выполнить один и тот же кусок кода одного и того же объекта, то можно сделать так чтобы они выполняли его поочереди.

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

public class SynchronizedExample { public static void main(String[] args) { // Объект ресурса (общий для потоков ресурс) CommonResource commonResource = new CommonResource(); // Запускаем 5 потоков, передавая в каждый из них // ресурс, общий для потоков. for (int i = 1; i < 6; i++) { Thread t = new Thread(new CountThread(commonResource)); t.setName("Thread " + i); t.start(); } // Когда выполнение кода одного из параллельных // потоков доходит до оператора synchronized, // доступ к блоку кода synchronized объекта // ресурса (commonResource) блокируется, и на // время его блокировки монопольный доступ к // блоку кода в этом объекте имеет только один // поток, который дошел до synchronized и который // произвел блокировку, и все прочие потоки, // которые используют commonResource будут ждать // пока поток, который первый дошел до блока // synchronized, закончит его выполнение. } } class CommonResource { int x; // Когда один из потоков доходит сюда, доступ // к блоку кода synchronized объекта // CommonResource блокируется, и другие потоки // которые дошли до synchronized останавливаются // и ждут, пока поток, дошедший сюда раньше них, // выполнит блок synchronized. Когда он все-таки // выполнил synchronized, в synchronized заходит // другой поток из очереди, и другие ждут, пока уже // этот другой поток выполнит код synchronized, // и так далее со всеми остальными потоками // в очереди. synchronized void increment() { x = 1; for (int i = 1; i < 5; i++) { System.out.printf("%s %d \n", Thread.currentThread().getName(), x); x++; try { Thread.sleep(700); } catch (InterruptedException e) {} } } } class CountThread implements Runnable { CommonResource res; CountThread(CommonResource res) { this.res = res; } public void run() { res.increment(); // вызываем синхронизированный метод } }

Вывод:

Благодаря этому потоки будут работать с ресурсом поочередно и от 1 до 4 сначала выведет первый дошедший до synchronized поток потом второй и так далее. Видим в консоли, что потоки выстроились в очередь в таком порядке – 1 5 2 4 3.

Метод join останавливает поток в котором этот метод был вызван пока не закончит выполнение поток к которому этот метод был вызван.

Более понятно на примере:

import java.io.*; class ThreadJoining extends Thread { @Override public void run(){ for (int i = 0; i < 2; i++){ try{ Thread.sleep(500); System.out.println("Current Thread: " + Thread.currentThread().getName()); } catch(Exception e){ System.out.println("Exception has been caught"+e); } System.out.println(i); } } } class JoinExample { public static void main (String[] args){ ThreadJoining t1 = new ThreadJoining(); ThreadJoining t2 = new ThreadJoining(); ThreadJoining t3 = new ThreadJoining(); t1.start(); try{ System.out.println("Current Thread: " + Thread.currentThread().getName()); //ниже join вызван в потоке мейн к потоку t1 t1.join();//приостанавливает текущий поток //(в данном случае поток в котором выполняется мейн) //пока не выполниться t1 } catch(Exception e){ System.out.println("Exception has been caught" + e); } t2.start(); try{ System.out.println("Current Thread: " + Thread.currentThread().getName()); //Опять таки останавливаем мейн //пока не выполниться t2. t2.join(); } catch(Exception e){ System.out.println("Exception has been caught" + e); } t3.start(); try{ System.out.println("Current Thread: " + Thread.currentThread().getName()); //Опять таки останавливаем мейн //пока не выполниться t3. t3.join(); } catch(Exception e){ System.out.println("Exception has been caught" + e); } } }

Вывод:

В консоли видно, что мейн останавливался, давая по очереди полностью выполниться потокам.

Для создания общей глобальной переменной, которую будут использовать несколько потоков используется слово volatile.

Создадим два потока. Первый будет добавлять 1 к глобальной переменной и как только он добавил 1 к глобальной переменной, во втором потоке будет происходить вывод на консоль нового значения глобальной переменной. Этот процесс будет продолжаться пока i не станет 8.

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

public class VolatileExample { static volatile int i;//объявим i volatile public static void main(String[] args) { new MyThread1().start();//запустим потоки new MyThread2().start(); } static class MyThread1 extends Thread { @Override public void run() { while(i<8){ System.out.println("inc i = " + (++i)); try{ //Sleep для остановки текущего потока //на заданное количество миллисекунд //в данном случае 100. Thread.sleep(100); } catch(InterruptedException e){} } } } static class MyThread2 extends Thread { @Override public void run() { int localvar = i; while(i<8){ if(localvar != i){ //Если бы i не был volatile то здесь //была бы копия i и она вечно была //бы 0 и поток бы ничего не вывел. System.out.println("new i = " + i); localvar = i; } } } } }

Вывод:

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

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

//Создадим простенький класс. //Дальше поясним зачем он. class Friend { public static void m1() { System.out.println(“Hello Friend!!!”); } } //Также можно увидеть что мы реализуем не только //Runnable но и расширяем Friend. Это преимущество //реализации Runnable перед расширением Thread //так как в Java можно расширять Лишь Один класс //и если бы мы расширили Thread то расширить //какой либо другой класс (например Friend) //уже не имели бы возможности. class MyThread extends Friend implements Runnable { int i=0; public void run() { i++; m1(); } } class RunnableExample { public static void main(String[] args) { //создаем один объект MyThread. MyThread MyThr = new MyThread(); //Можно у одного объекта //в отдельных потоках запускать его метод run //это преимущество Runnable перед Thread. //Подробнее на следующей странице. Thread t1 = new Thread(MyThr);//передается в поток Thread t2 = new Thread(MyThr);//передается в поток 2 раз //run объекта MyThr запускается //в отдельных потоках t1 и t2. t1.start(); t2.start(); //Еще одно преимущество Runnable перед Thread //что код класса реализующего Runnable можно //использовать вне в отдельном потоке. //То есть m1 просто выполнится здесь в потоке метода main MyThr.m1(); //В конце можно увидеть 2. То есть оба потока работали //с одним и тем же объектом. //Все потоки добавили 1 к полю i объекта MyThr. System.out.println(MyThr.i); } }

Вывод:

В консоли можно увидеть 2. То есть оба потока работали с одним и тем же объектом. Все потоки добавляли 1 к полю i объекта MyThr.

---

Преимущества реализации Runnable

Преимущества реализации Runnable перед расширением Thread:

То есть мы видели в прошлом уроке:

MyThread t1 = new MyThread();

MyThread t2 = new MyThread();

MyThread t3 = new MyThread();

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