You Can't Spoil Java with Threads: Part VI - Toward the Barrier!

Introduction

Streams are an interesting thing. In past reviews, we looked at some of the available means for implementing multithreading. Let's see what else we can do. At this point, we know a lot. For example, from " Java Threads Don't Mess Up: Part I - Threads " we know that a thread is a Thread. We know that a thread is performing some task. If we want our tasks to be started ( run), then we must tell the thread some Runnable. To remember, we can use the 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 thing as lok. We read about it in " You Can't Mess with Java Threads: Part II - Synchronization ". A thread can acquire a lock, and then another thread that tries to acquire the lock will have to wait for the lock to be released:

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 interesting.

semaphores

The simplest means of controlling how many threads can run at the same time is a semaphore. Like on the railroad. Lights up green - you can. The red light is on - we are waiting. What do we expect from a semaphore? Permissions. Permission in English - permit. To get permission, you need to get 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 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 you can see, having memorized English words, we understand how the semaphore works. Interestingly, the main condition is that the semaphore "account" must have a positive number of permits. Therefore, it can be initiated with a minus. And you can request (acquire) more than 1.

CountDownLatch

The next mechanism CountDownLatchis CountDown in English is a countdown, and Latch is a valve or latch. That is, if you translate, then this is a latch with a countdown. Here we need the corresponding class import java.util.concurrent.CountDownLatch. It's like running or racing, when everyone gathers at the starting line and when everyone is ready, they give permission, 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 - expect. That is, we speak first countDown. As Google translate 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 value of the counter becomes zero. It is interesting that such a counter is disposable. As stated in the JavaDoc - "When threads must repeatedly count down in this way, instead use a CyclicBarrier", that is, if you need a reusable count, you need to use another option called CyclicBarrier.

CyclicBarrier

As the name implies, CyclicBarrierit is a cyclical barrier. We 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. This reduces the value of the barrier. The barrier is considered broken ( berrier.isBroken()) when the countdown has reached zero. To reset the barrier, you need to call berrier.reset(), which was missing in CountDownLatch.

Exchanger

The next tool is Exchanger. Exchange is translated from English as an exchange or exchange. A Exchanger- an exchanger, that is, through which they exchange. Let's look at the simplest 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 start two threads. Each of them executes the exchange method and waits for the other thread to also execute the exchange method. Thus, the threads will exchange the passed arguments with each other. An interesting thing. Does she remind you of anything? And he reminds SynchronousQueue, which underlies cachedThreadPool'a. For clarity, 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, because the queue will be empty. And then mainthe thread will queue the text "Message". At the same time, he himself will stop for the time that is needed until this text element is received from the queue. You can also read SynchronousQueue Vs Exchanger on this topic .

Phaser

And finally, the sweetest - Phaser. We 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();
    }

It can be seen from the example that the barrier, when using Phaser'a, breaks when the number of registrations matches the number of arrivals at the barrier. More details can be found Phaserin the article from Habr " New Phaser Synchronizer ".

Results

As you can see from the examples, there are various ways to synchronize threads. Earlier, I tried to remember something from 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 pretty fundamental and still holds its own. 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 you should refer to " Chapter 4. Modern concurrency ". There is another review on the same subject:". There are also tips on what else to read to really understand this 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, i.e. OCP And there are tests in "∫". In this book there are both questions and answers with explanation. For example: Many may start to say that this is another memorization of methods. On the one hand, yes. On the other hand, this question can be give an answer, remembering that ExecutorService- this is a kind of "upgrade" Executor'a. And Executorit is intended simply to hide the method of creating threads, but not the main way to execute them, that is, running in a new thread Runnable... Therefore execute(Callable), no, because in ExecutorServicek Executor'u just added methodssubmit, which can return Future. As you can see, we can memorize the list of methods, but it's much easier to guess, knowing the nature of the classes themselves. Well, a few additional materials on the topic:

• " Java.util.concurrent.* synchronizers reference "
• " Concurrent Package Synchronizers "
• Yuriy Tkach " Synchronizers - Concurrency #4 - Advanced Java "

#Viacheslav