Analysis of the main concepts of concurrency

All coffee!



Tomorrow, we have a smoothly launched almost anniversary stream Java Developer course - already the sixth in a row since April last year. And this means that we have again picked up, translated the most interesting material that we share with you.



Go!

')

This memo will help Java developers working with multithreaded programs understand the basic concepts of concurrency and how to use them. You will learn about key aspects of the Java language with links to the standard library.



SECTION 1



Introduction



Since its inception, Java has supported key concepts of concurrency, such as threads and locks. This memo will help Java developers working with multithreaded programs understand the basic concepts of concurrency and how to use them.



SECTION 2



Concepts

Concept	Description
Atomicity	An atomic operation is an operation that is performed completely or not at all, partial execution is impossible.
Visibility	Conditions under which one thread sees changes made by another thread

Table 1: Concurrency Concepts







Race condition



A race condition occurs when the same resource is used by several threads simultaneously, and depending on the order of actions of each thread there can be several possible results. The code below is not thread-safe, and the value variable can be initialized more than once, since a check-then-act (check for null and then initialization) that lazily initializes the field is not atomic :

 class Lazy <T> { private volatile T value; T get() { if (value == null) value = initialize(); return value; } } 

Data Race



Data race occurs when two or more threads try to access the same non-final variable without synchronization. Lack of synchronization can lead to changes that will not be visible to other threads, because of this, it is possible to read obsolete data, which, in turn, leads to infinite loops, corrupted data structures or inaccurate calculations. This code can lead to an infinite loop, because the read stream may never notice the changes made by the rewriting streams:

 class Waiter implements Runnable { private boolean shouldFinish; void finish() { shouldFinish = true; } public void run() { long iteration = 0; while (!shouldFinish) { iteration++; } System.out.println("Finished after: " + iteration); } } class DataRace { public static void main(String[] args) throws InterruptedException { Waiter waiter = new Waiter(); Thread waiterThread = new Thread(waiter); waiterThread.start(); waiter.finish(); waiterThread.join(); } } 

SECTION 3



Java Memory Model: the happens-before relationship



The Java memory model is defined in terms of actions such as reading / writing fields and synchronization in the monitor. Actions are ordered using an happens-before relationship (performed before), which can be used to explain when a thread sees the result of actions from another thread, and what a correctly synchronized program is.



RELATIONSHIPS HAPPENS-BEFORE HAVE THE FOLLOWING PROPERTIES:

• Calling Thread # start occurs before any action in this thread.
• The return of the monitor occurs before any subsequent capture of the same monitor.
• Writing to a volatile variable occurs before any subsequent reading of the volatile variable.
• Writing to the final variable occurs before the object link is published.
• All actions in the thread are performed before returning from Thread # join in this thread.

In Image 1, Action X occurs before Action Y , so in Thread 2 all operations to the right of Action Y will see all operations to the left of Action X in Thread 1 .





Image 1: The example happens-before





SECTION 4



Standard sync features



Keyword synchronized



The synchronized used to prevent simultaneous execution by different threads of the same block of code. It ensures that if you get a lock (by entering a synchronized block), the data on which this lock is imposed is processed in exclusive mode, so the operation can be considered atomic. In addition, it ensures that other threads see the result of the operation after they get the same lock.

 class AtomicOperation { private int counter0; private int counter1; void increment() { synchronized (this) { counter0++; counter1++; } } } 

The synchronized keyword can also be expanded at the method level.

METHOD TYPE	REFERENCE, USED AS A MONITOR
static	reference to the Class object <?>
non-static	this-link

Table 2: Monitors that are used when the entire method is synchronized



The reentrant lock, so if the thread already contains a lock, it can successfully get it again.

 class Reentrantcy { synchronized void doAll() { doFirst(); doSecond(); } synchronized void doFirst() { System.out.println("First operation is successful."); } synchronized void doSecond() { System.out.println("Second operation is successful."); } } 

The level of rivalry affects how the monitor is captured:

condition	Description
init	Just created, while no one was captured.
biased	There is no struggle, and the code protected by blocking is executed by only one thread. Cheapest to capture.
thin	The monitor is captured by multiple threads without a fight. For blocking, a relatively cheap CAS is used.
fat	There is a struggle. The JVM queries the OS mutexes and allows the OS scheduler to handle parking threads and wake ups.

Table 3: Monitor Status



wait/notify



The wait/notify/notifyAll are declared in the Object class. wait used to force the thread to go to the WAITING or TIMED_WAITING (if the time-out value is transmitted). To wake the thread, you can do any of these actions:

• Another thread calls notify, which wakes up an arbitrary thread waiting on the monitor.
• Another thread calls notifyAll, which wakes up all threads waiting on the monitor.
• Called Thread # interrupt. In this case, an InterruptedException is thrown.

The most common example is the conditional loop:

 class ConditionLoop { private boolean condition; synchronized void waitForCondition() throws InterruptedException { while (!condition) { wait(); } } synchronized void satisfyCondition() { condition = true; notifyAll(); } } 

• Keep in mind that in order to use wait/notify/notifyAll for an object, you must first impose a lock on this object.
• Always wait inside a loop that checks the condition you are expecting. This concerns a synchronization problem if another thread satisfies the condition before waiting. In addition, it protects your code from side wakeups, which can (and will) occur.
• Always check that you meet the wait condition before calling notify / notifyAll. Failure to comply with this requirement will result in a notification, but the thread cannot avoid the wait loop.

Volatile keyword



volatile solves the visibility problem and makes the value change atomic , because here the relationship happens-before: writing to a volatile variable happens before any subsequent reading of the volatile variable. Thus, it guarantees that the next reading of the field will show the value that was specified by the most recent entry.

 class VolatileFlag implements Runnable { private volatile boolean shouldStop; public void run() { while (!shouldStop) { //do smth } System.out.println("Stopped."); } void stop() { shouldStop = true; } public static void main(String[] args) throws InterruptedException { VolatileFlag flag = new VolatileFlag(); Thread thread = new Thread(flag); thread.start(); flag.stop(); thread.join(); } } 

Atomicity



The java.util.concurrent.atomic package contains a set of classes that support composite atomic actions on a single value without locks, like volatile .



Using the classes AtomicXXX, you can implement the atomic check-then-act operation:

 class CheckThenAct { private final AtomicReference<String> value = new AtomicReference<>(); void initialize() { if (value.compareAndSet(null, "Initialized value")) { System.out.println("Initialized only once."); } } } 

Both AtomicInteger and AtomicLong have an atomic increment / decrement operation:

 class Increment { private final AtomicInteger state = new AtomicInteger(); void advance() { int oldState = state.getAndIncrement(); System.out.println("Advanced: '" + oldState + "' -> '" + (oldState + 1) + "'."); } } 

If you need a counter and you do not need to get its value atomically, consider using LongAdder instead of AtomicLong/AtomicInteger . LongAdder processes the value in several cells and increases their number, if needed, and, therefore, it works better with high competition.



ThreadLocal



One way to store data in the stream and make the lock optional is to use the ThreadLocal store. Conceptually, ThreadLocal acts as if each thread has its own version of the variable. ThreadLocal commonly used to capture the values ​​of each thread, such as the “current transaction,” or other resources. In addition, they are used to maintain stream counters, statistics, or identifier generators.

 class TransactionManager { private final ThreadLocal<Transaction> currentTransaction = ThreadLocal.withInitial(NullTransaction::new); Transaction currentTransaction() { Transaction current = currentTransaction.get(); if (current.isNull()) { current = new TransactionImpl(); currentTransaction.set(current); } return current; } } 

SECTION 5



Secure publication



The publication of the object makes its link available outside the current scope (for example, returning a link from the getter). Ensuring the secure publication of an object (only when it is fully created) may require synchronization. Publication security can be achieved using:

• Static initializers Only one thread can initialize static variables, since the class is initialized under exclusive locking.

 class StaticInitializer { //       public static final Year year = Year.of(2017); public static final Set<String> keywords; //        static { //    Set<String> keywordsSet = new HashSet<>(); //   keywordsSet.add("java"); keywordsSet.add("concurrency"); //    keywords = Collections.unmodifiableSet(keywordsSet); } } 

• Volatile-field. A read thread will always read the last value, because writing to a volatile variable occurs before any subsequent read.

 class Volatile { private volatile String state; void setState(String state) { this.state = state; } String getState() { return state; } } 

• Atomicity. For example, AtomicInteger stores the value in a volatile-field, so the rule for volatile-variables is also applicable here.

 class Atomics { private final AtomicInteger state = new AtomicInteger(); void initializeState(int state) { this.state.compareAndSet(0, state); } int getState() { return state.get(); } } 

• Final fields.

 class Final { private final String state; Final(String state) { this.state = state; } String getState() { return state; } } 

Ensure that this link does not evaporate during creation.

 class ThisEscapes { private final String name; ThisEscapes(String name) { Cache.putIntoCache(this); this.name = name; } String getName() { return name; } } class Cache { private static final Map<String, ThisEscapes> CACHE = new ConcurrentHashMap<>(); static void putIntoCache(ThisEscapes thisEscapes) { // 'this'   ,    . CACHE.putIfAbsent(thisEscapes.getName(), thisEscapes); } } 

• Correctly synchronized fields.

 class Synchronization { private String state; synchronized String getState() { if (state == null) state = "Initial"; return state; } } 

SECTION 6



Immutable objects



One of the most remarkable properties of immutable objects is thread safety , so synchronization is not needed for them. Requirements for a fixed object:

• All fields are final fields.
• All fields must be either changeable or immutable objects, but not go beyond the object, therefore the state of the object can not be changed after creation.
• This link does not disappear during creation.
• A class is a final-class; therefore, overriding its behavior in subclasses is impossible.

An example of an immutable object:

 //   final -   public final class Artist { //  ,  final private final String name; //   , final  private final List<Track> tracks; public Artist(String name, List<Track> tracks) { this.name = name; //   List<Track> copy = new ArrayList<>(tracks); //      this.tracks = Collections.unmodifiableList(copy); // 'this'       } // Getters, equals, hashCode, toString } //  final -   public final class Track { // ,  final private final String title; public Track(String title) { this.title = title; } // Getters, equals, hashCode, toString } 

SECTION 7



Streams



The java.lang.Thread class is used to represent an application or a JVM stream. The code is always executed in the context of some Thread class (you can use Thread#currentThread()). to get the current thread Thread#currentThread()).

condition	Description
NEW	It did not start.
RUNNABLE	Started and running.
BLOCKED	Waiting on the monitor - he is trying to get a lock and enter the critical section.
WAITING	Waiting to perform a specific action by another thread (notify / notifyAll, LockSupport # unpark).
TIMED_WAITING	Same as WAITING, but with a timeout.
TERMINATED	Stopped

Table 4: Thread States

Stream method	Description
start	Runs an instance of the Thread class and executes the run () method.
join	Blocks until the end of the stream.
interrupt	Stops the thread. If a thread is blocked in a method that responds to interrupts, InterruptedException will be thrown in another thread, otherwise the interrupt status will be set.
stop, suspend, resume, destroy	All of these methods are outdated. They perform dangerous operations depending on the state of the stream in question. Instead, use Thread # interrupt () or the volatile flag to tell the thread what to do.

Table 5: Thread coordination methods Thread coordination methods



How to handle InterruptedException?

• Clear all resources and terminate the thread if possible at the current level.
• Declare the current method to throw an InterruptedException.
• If the method does not throw an InterruptedException exception, the interrupted flag must be restored to true by calling Thread.currentThread (). Interrupt () and an exception should be thrown that is more appropriate at this level. It is very important to return a true flag in order to allow interrupts to be processed at a higher level.

Handling unexpected exceptions



UncaughtExceptionHandler may be reported in UncaughtExceptionHandler , which will receive notification of any uncaught exception for which the stream is interrupted.

 Thread thread = new Thread(runnable); thread.setUncaughtExceptionHandler((failedThread, exception) -> { logger.error("Caught unexpected exception in thread '{}'.", failedThread.getName(), exception); }); thread.start(); 

SECTION 8



Vitality (Liveness)



Deadlock



Deadlock , or deadlock, occurs when there are several threads and each is waiting for a resource belonging to another thread, so that a loop is formed from the resources and the threads that are waiting for them. The most obvious type of resource is an object monitor, but any resource that causes a lock (for example, wait/notify ) is also appropriate.



An example of a potential deadlock:

 class Account { private long amount; void plus(long amount) { this.amount += amount; } void minus(long amount) { if (this.amount < amount) throw new IllegalArgumentException(); else this.amount -= amount; } static void transferWithDeadlock(long amount, Account first, Account second){ synchronized (first) { synchronized (second) { first.minus(amount); second.plus(amount); } } } } 

Mutual locking occurs if at the same time:

• One thread is trying to transfer data from one account to another and has already imposed a lock on the first account.
• Another thread is trying to transfer data from the second account to the first one, and has already imposed a lock on the second account.

Ways to prevent deadlock:

• Locking Order — Always apply locks in the same order.

 class Account { private long id; private long amount; //    static void transferWithLockOrdering(long amount, Account first, Account second){ boolean lockOnFirstAccountFirst = first.id < second.id; Account firstLock = lockOnFirstAccountFirst ? first : second; Account secondLock = lockOnFirstAccountFirst ? second : first; synchronized (firstLock) { synchronized (secondLock) { first.minus(amount); second.plus(amount); } } } } 

• Locking with time-out - do not block indefinitely when blocking, it is better to remove all locks as soon as possible and try again.

 class Account { private long amount; //    static void transferWithTimeout( long amount, Account first, Account second, int retries, long timeoutMillis ) throws InterruptedException { for (int attempt = 0; attempt < retries; attempt++) { if (first.lock.tryLock(timeoutMillis, TimeUnit.MILLISECONDS)) { try { if (second.lock.tryLock(timeoutMillis, TimeUnit.MILLISECONDS)) { try { first.minus(amount); second.plus(amount); } finally { second.lock.unlock(); } } } finally { first.lock.unlock(); } } } } } 

The JVM is able to detect the interlocking of monitors and output information about them in the dumps of the streams.



Livelock and stream starvation



Livelock occurs when threads spend all their time negotiating access to a resource, or they discover and avoid a deadlock so that the thread does not actually move forward. Fasting occurs when the streams keep blocking for long periods, so some streams “starve” without progress.



SECTION 9



java.util.concurrent



Thread pools



The main interface for thread pools is ExecutorService.java.util.concurrent also provides a static factory Executors, which contains factory methods for creating a thread pool with the most common configurations.

Method	Description
newSingleThreadExecutor	Returns an ExecutorService with only one thread.
newFixedThreadPool	Returns an ExecutorService with a fixed number of threads.
newCachedThreadPool	Returns an ExecutorService with a pool of threads of various sizes.
newSingleThreadScheduledExecutor	Returns a single thread ScheduledExecutorService.
newScheduledThreadPool	Returns a ScheduledExecutorService with the main set of threads.
newWorkStealingPool	Returns the task stealing ExecutorService.

Table 6: Static Factory Methods



When determining the size of thread pools, it is often useful to determine the size of the number of logical cores in the machine on which the application is running. You can get this value in Java by calling Runtime.getRuntime().AvailableProcessors() .

Implementation	Description
ThreadPoolExecutor	Default implementation with resizable thread pool, one working queue, and custom policies for rejected tasks (via RejectedExecutionHandler) and thread creation (via ThreadFactory).
ScheduledThreadPoolExecutor	The ThreadPoolExecutor extension that provides the ability to schedule periodic tasks.
ForkJoinPool	Task pool stealing tasks: all threads in the pool are trying to find and start either the assigned tasks, or tasks created by other active tasks.

Table 7: Thread Pool Implementations



Tasks are sent using ExecutorService#submit , ExecutorService#invokeAll or ExecutorService#invokeAny , which have several overloads for different types of tasks.

Interface	Description
Runnable	Represents a task with no return value.
Callable	Represents a calculation with a return value. It also throws the original Exeption, so no wrapper is required for the checked exception.

Table 8: Task Functional Interfaces



Future



Future is an abstraction for asynchronous computing. It represents the result of the calculation, which may be available at some point: either the calculated value or the exception. Most ExecutorService methods use Future as the return type. It provides methods for examining the current state of the future or blocks until the result is available.

 ExecutorService executorService = Executors.newSingleThreadExecutor(); Future<String> future = executorService.submit(() -> "result"); try { String result = future.get(1L, TimeUnit.SECONDS); System.out.println("Result is '" + result + "'."); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new RuntimeException(e); } catch (ExecutionException e) { throw new RuntimeException(e.getCause()); } catch (TimeoutException e) { throw new RuntimeException(e); } assert future.isDone(); 

Locks



Lock



The java.util.concurrent.locks package has a standard Lock interface. The implementation of ReentrantLock duplicates the synchronized keyword functionality, but also provides additional functions, such as getting information about the state of the lock, non-blocking tryLock() and interruptable locking. An example of using an explicit ReentrantLock instance:

 class Counter { private final Lock lock = new ReentrantLock(); private int value; int increment() { lock.lock(); try { return ++value; } finally { lock.unlock(); } } } 

ReadWriteLock



java.util.concurrent.locks ReadWriteLock ( ReentrantReadWriteLock), , , .

 class Statistic { private final ReadWriteLock lock = new ReentrantReadWriteLock(); private int value; void increment() { lock.writeLock().lock(); try { value++; } finally { lock.writeLock().unlock(); } } int current() { lock.readLock().lock(); try { return value; } finally { lock.readLock().unlock(); } } } 

CountDownLatch



CountDownLatch . await() , , 0. ( ) countDown() , . , 0. , .



CompletableFuture



CompletableFuture . Future, — , , , . ( CompletableFuture#supplyAsync/runAsync ), ( *async ) , ( ForkJoinPool#commonPool ).



, CompletableFuture , , *async , .



future , CompletableFuture#allOf , future , , future , CompletableFuture#anyOf , , - future .

 ExecutorService executor0 = Executors.newWorkStealingPool(); ExecutorService executor1 = Executors.newWorkStealingPool(); //,   future  CompletableFuture<String> waitingForAll = CompletableFuture .allOf( CompletableFuture.supplyAsync(() -> "first"), CompletableFuture.supplyAsync(() -> "second", executor1) ) .thenApply(ignored -> " is completed."); CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> "Concurrency Refcard", executor0) //    .thenApply(result -> "Java " + result) //   .thenApplyAsync(result -> "Dzone " + result, executor1) //,     future  .thenCombine(waitingForAll, (first, second) -> first + second) //  ForkJoinPool#commonPool   .thenAcceptAsync(result -> { System.out.println("Result is '" + result + "'."); }) //  .whenComplete((ignored, exception) -> { if (exception != null) exception.printStackTrace(); }); //   - ,     . future.join(); future //    (  ). .thenRun(() -> System.out.println("Current thread is '" + Thread.currentThread().getName() + "'.")) //  ForkJoinPool#commonPool   .thenRunAsync(() -> System.out.println("Current thread is '" + Thread.currentThread().getName() + "'.")); 

— Collections#synchronized* . , java.util.concurrent , .



Lists

	Description
CopyOnWriteArrayList	, ( , ). .

9: java.util.concurrent

	Description
ConcurrentHashMap	-. , , . CAS- ( ), ( ).
ConcurrentSkipListMap	Map, TreeMap. TreeMap, , .

10: java.util.concurrent



The sets

	Description
CopyOnWriteArraySet	CopyOnWriteArrayList, copy-on-write Set.
ConcurrentSkipListSet	ConcurrentSkipListMap, Set.

11: java.util.concurrent



Map:

 Set<T> concurrentSet = Collections.newSetFromMap(new ConcurrentHashMap<T, Boolean>()); 

«» «». « , » (FIFO). BlockingQueue Queue , , , ( ) ( ). BlockingQueue , , , - .

	Description
ConcurrentLinkedQueue	, .
LinkedBlockingQueue	, .
PriorityBlockingQueue	, . , Comparator, ( FIFO).
DelayQueue	, . , .
SynchronousQueue	-, , . , . .

12: java.util.concurrent



THE END



.



Thank.