You can't spoil Java with a Thread: Part VI - To the barrier!

Introduction

Streams are an interesting thing. In previous reviews, we looked at some of the available tools for implementing multithreading. Let's see what else interesting things we can do. At this point we know a lot. For example, from “ You Can't Spoil Java with a Thread: Part I - Threads, ” we know that a thread is a Thread. We know that a thread is performing some task. If we want our task to be able to run ( run), then we must specify the thread to be a certain Runnable. To remember, we can use Tutorialspoint Java Online Compiler :

public static void main(String []args){
	Runnable task = () -> {
 		Thread thread = Thread.currentThread();
		System.out.println("Hello from " + thread.getName());
	};
	Thread thread = new Thread(task);
	thread.start();
}

We also know that we have such a concept as lock. We read about it in “ You Can’t Spoil Java with a Thread: Part II - Synchronization .” A thread can occupy a lock and then another thread that tries to occupy the lock will be forced to wait for the lock to become free:

import java.util.concurrent.locks.*;

public class HelloWorld{
	public static void main(String []args){
		Lock lock = new ReentrantLock();
		Runnable task = () -> {
			lock.lock();
			Thread thread = Thread.currentThread();
			System.out.println("Hello from " + thread.getName());
			lock.unlock();
		};
		Thread thread = new Thread(task);
		thread.start();
	}
}

I think it’s time to talk about what else we can do that’s interesting.

Semaphores

The simplest means of controlling how many threads can work simultaneously is a semaphore. Like on the railroad. The green light is on - you can. The red light is on - we are waiting. What do we expect from a semaphore? Permissions. Permission in English - permit. To obtain permission, you need to obtain it, which in English will be acquire. And when the permission is no longer needed, we must give it away, that is, release it or get rid of it, which in English will be release. Let's see how it works. We will need to import the class java.util.concurrent.Semaphore. Example:

public static void main(String[] args) throws InterruptedException {
	Semaphore semaphore = new Semaphore(0);
	Runnable task = () -> {
		try {
			semaphore.acquire();
			System.out.println("Finished");
			semaphore.release();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	};
	new Thread(task).start();
	Thread.sleep(5000);
	semaphore.release(1);
}

As we can see, having memorized English words, we understand how semaphore works. Interestingly, the main condition is that the semaphore “account” must have a positive number of permits. Therefore, you can initiate it with a minus. And you can request (acquire) more than 1.

CountDownLatch

The next mechanism is CountDownLatch. CountDown in English is a countdown, and Latch is a bolt or latch. That is, if we translate it, then this is a latch with a countdown. Here we need the appropriate import of the class java.util.concurrent.CountDownLatch. It's like a race or a race where everyone gathers at the starting line and when everyone is ready, permission is given and everyone starts at the same time. Example:

public static void main(String[] args) {
	CountDownLatch countDownLatch = new CountDownLatch(3);
	Runnable task = () -> {
		try {
			countDownLatch.countDown();
			System.out.println("Countdown: " + countDownLatch.getCount());
			countDownLatch.await();
			System.out.println("Finished");
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	};
	for (int i = 0; i < 3; i++) {
		new Thread(task).start();
 	}
}

await in English - to expect. That is, we speak first countDown. As Google Translator says, count down is “an act of counting numerals in reverse order to zero,” that is, perform a countdown action, the purpose of which is to count to zero. And then we say await- that is, wait until the counter value becomes zero. It is interesting that such a counter is disposable. As it is said in the JavaDoc - "When threads must repeatedly count down in this way, instead use a CyclicBarrier", that is, if you need reusable counting, you need to use another option, which is called CyclicBarrier.

CyclicBarrier

As the name suggests, CyclicBarrierit is a cyclical barrier. We will need to import the class java.util.concurrent.CyclicBarrier. Let's look at an example:

public static void main(String[] args) throws InterruptedException {
	Runnable action = () -> System.out.println("На старт!");
	CyclicBarrier berrier = new CyclicBarrier(3, action);
	Runnable task = () -> {
		try {
			berrier.await();
			System.out.println("Finished");
		} catch (BrokenBarrierException | InterruptedException e) {
			e.printStackTrace();
		}
	};
	System.out.println("Limit: " + berrier.getParties());
	for (int i = 0; i < 3; i++) {
		new Thread(task).start();
	}
}

As you can see, the thread is executing await, that is, waiting. In this case, the value of the barrier decreases. The barrier is considered broken ( berrier.isBroken()) when the countdown reaches zero. To reset the barrier, you need to call berrier.reset(), which was missing in CountDownLatch.

Exchanger

The next remedy is Exchanger. Exchange from English is translated as exchange or exchange. A Exchangeris an exchanger, that is, something through which they exchange. Let's look at a simple example:

public static void main(String[] args) {
	Exchanger<String> exchanger = new Exchanger<>();
	Runnable task = () -> {
		try {
			Thread thread = Thread.currentThread();
			String withThreadName = exchanger.exchange(thread.getName());
			System.out.println(thread.getName() + " обменялся с " + withThreadName);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	};
	new Thread(task).start();
	new Thread(task).start();
}

Here we launch two threads. Each of them executes the exchange method and waits for another thread to also execute the exchange method. Thus, the threads will exchange the passed arguments among themselves. Interesting thing. Doesn't she remind you of anything? And he reminds SynchronousQueue, which lies at the heart of cachedThreadPool'a. For clarity, here is an example:

public static void main(String[] args) throws InterruptedException {
	SynchronousQueue<String> queue = new SynchronousQueue<>();
	Runnable task = () -> {
		try {
			System.out.println(queue.take());
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	};
	new Thread(task).start();
	queue.put("Message");
}

The example shows that by launching a new thread, this thread will go into waiting mode, because the queue will be empty. And then mainthe thread will queue the text “Message”. At the same time, it will stop for the required time until it receives this text element from the queue. On this topic you can also read " SynchronousQueue Vs Exchanger ".

Phaser

And finally, the sweetest thing - Phaser. We will need to import the class java.util.concurrent.Phaser. Let's look at a simple example:

public static void main(String[] args) throws InterruptedException {
        Phaser phaser = new Phaser();
        // Вызывая метод register, мы регистрируем текущий поток (main) How участника
        phaser.register();
        System.out.println("Phasecount is " + phaser.getPhase());
        testPhaser(phaser);
        testPhaser(phaser);
        testPhaser(phaser);
        // Через 3 секунды прибываем к барьеру и снимаемся регистрацию. Кол-во прибывших = кол-во регистраций = пуск
        Thread.sleep(3000);
        phaser.arriveAndDeregister();
        System.out.println("Phasecount is " + phaser.getPhase());
    }

    private static void testPhaser(final Phaser phaser) {
        // Говорим, что будет +1 участник на Phaser
        phaser.register();
        // Запускаем новый поток
        new Thread(() -> {
            String name = Thread.currentThread().getName();
            System.out.println(name + " arrived");
            phaser.arriveAndAwaitAdvance(); //threads register arrival to the phaser.
            System.out.println(name + " after passing barrier");
        }).start();
    }

The example shows that the barrier, when using Phaser'a, is broken when the number of registrations coincides with the number of arrivals at the barrier. You can find out more Phaserin the article from the hub " New Phaser synchronizer ".

Results

As you can see from the examples, there are different ways to synchronize threads. Earlier I tried to remember something about multithreading, I hope the previous parts were useful. They say that the path to multithreading begins with the book "Java Concurrency in Practice". Although it came out in 2006, people respond that the book is quite fundamental and still packs a punch. For example, you can read the discussions here: " Is Java Concurrency In Practice still valid? ". It's also helpful to read the links from the discussion. For example, there is a link to the book " The Well-Grounded Java Developer ", in which it is worth paying attention to " Chapter 4. Modern concurrency ". There is another whole review on the same topic: “ Is Java cocurrency in pracitce still relevant in the era of java 8 ”. It also has tips on what else you should read to really understand the topic. After that, you can take a closer look at such a wonderful book as " OCA OCP JavaSE 8 Programmer Practice Tests ". We are interested in the second part, that is, OCP. And there are tests in "∫". This book contains both questions and answers with explanations. For example: Many may start to say that this is just another memorization of methods. On the one hand, yes. On the other hand, this question can be answered by remembering that ExecutorServicethis is a kind of “upgrade” Executor. And Executorit is intended simply to hide the method of creating threads, but not the main way of executing them, that is, running in a new thread Runnable. Therefore, execute(Callable)no, because they simply added methods ExecutorServicethat can return . As you can see, we can memorize a list of methods, but it’s much easier to guess if we know the nature of the classes themselves. Well, some additional materials on the topic: ExecutorsubmitFuture

• " Synchronizer Reference java.util.concurrent.* "
• " Concurrent package synchronizers "
• Yuri Tkach " Synchronizers - Concurrency #4 - Advanced Java "

#Viacheslav