Shulou(Shulou.com)06/02 Report--

This article mainly introduces "what is the principle and function of CountDownLatch, Semaphore and CyclicBarrier". In daily operation, I believe many people have doubts about the principle and function of CountDownLatch, Semaphore and CyclicBarrier. The editor consulted all kinds of materials and sorted out simple and easy-to-use operation methods. I hope it will be helpful to answer the doubts about "what is the principle and function of CountDownLatch, Semaphore and CyclicBarrier?" Next, please follow the editor to study!

CountDownLatch

CountDownLatch is a counter lock that allows you to perform a function similar to blocking the current thread, that is, one or more threads wait until the operation performed by other threads is completed. CountDownLatch is initialized with a given counter whose operation is atomic, that is, only one thread can operate on the counter at a time. The thread calling the await method of this class remains blocked until another thread calls the countDown method to change the value of the current counter to zero, minus 1 for each call to the countDown counter. When the counter value decreases to 00:00, all threads that are waiting because of a call to the await () method continue to execute. This occurs only once, because counters cannot be reset, and if the business needs a version that can reset the count number of times, consider using CycliBarrier.

In some business scenarios, program execution needs to wait for a condition to be completed before continuing to perform subsequent operations. For typical applications such as parallel computing, when a processing has a large amount of computation, the computing task can be divided into multiple sub-tasks. After waiting for all the sub-tasks to be completed, the parent task will get the results of all the sub-tasks for summary.

Import lombok.extern.slf4j.Slf4j;import java.util.concurrent.CountDownLatch;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;@Slf4jpublic class CountDownLatchExample1 {private final static int threadCount = 200; public static void main (String [] args) throws Exception {ExecutorService exec = Executors.newCachedThreadPool (); final CountDownLatch countDownLatch = new CountDownLatch (threadCount); for (int I = 0; I

< threadCount; i++) { final int threadNum = i; exec.execute(() ->

{try {test (threadNum);} catch (Exception e) {log.error ("exception", e);} finally {countDownLatch.countDown ();}} countDownLatch.await () Log.info ("finish"); exec.shutdown ();} private static void test (int threadNum) throws Exception {Thread.sleep; log.info ("{}", threadNum); Thread.sleep (100);}}

Results:

20 pool-1-thread-6 18 INFO com.mmall.concurrency.example.aqs.CountDownLatchExample1 32.917 [pool-1-thread-6] INFO com.mmall.concurrency.example.aqs.CountDownLatchExample1-520 18 pool-1-thread-7 32.919 [pool-1-thread-5] INFO com.mmall.concurrency.example.aqs.CountDownLatchExample1-420 18 pool-1-thread-7 32.918 [pool-1-thread-1] INFO com.mmall.concurrency.example.aqs.CountDownLatchExample1-020 18: 32.918 [pool-1-thread-3] INFO com.mmall.concurrency.example.aqs.CountDownLatchExample1-220 pool-1-thread-9 18 INFO com.mmall.concurrency.example.aqs.CountDownLatchExample1-820 INFO com.mmall.concurrency.example.aqs.CountDownLatchExample1-820 18 INFO com.mmall.concurrency.example.aqs.CountDownLatchExample1 32.918 [pool-1-thread-4] INFO com.mmall.concurrency.example.aqs.CountDownLatchExample1-320 1832. 916 [pool-1-thread-10] INFO com.mmall.concurrency.example.aqs.CountDownLatchExample1-920 18 32. 916 [pool-1-thread-4] Pool-1-thread-8] INFO com.mmall.concurrency.example.aqs.CountDownLatchExample1-720 main 18 INFO com.mmall.concurrency.example.aqs.CountDownLatchExample1 32.917 [pool-1-thread-2] INFO com.mmall.concurrency.example.aqs.CountDownLatchExample1-120 18V 33.032 [main] INFO com.mmall.concurrency.example.aqs.CountDownLatchExample1-finishimport java.util.concurrent.CountDownLatch Import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.TimeUnit;@Slf4jpublic class CountDownLatchExample2 {private final static int threadCount = 200; public static void main (String [] args) throws Exception {ExecutorService exec = Executors.newCachedThreadPool (); final CountDownLatch countDownLatch = new CountDownLatch (threadCount); for (int I = 0; I

< threadCount; i++) { final int threadNum = i; exec.execute(() ->

{try {test (threadNum);} catch (Exception e) {log.error ("exception", e);} finally {countDownLatch.countDown ();}});} countDownLatch.await (10, TimeUnit.MILLISECONDS) Log.info ("finish"); exec.shutdown ();} private static void test (int threadNum) throws Exception {Thread.sleep; log.info ("{}", threadNum);}}

Result: skip waiting beyond the specified time

20 main 19 pool-1-thread-3 34.878 [main] INFO com.mmall.concurrency.example.aqs.CountDownLatchExample2-finish20:19:34.964 [pool-1-thread-3] INFO com.mmall.concurrency.example.aqs.CountDownLatchExample2-220 pool-1-thread-3 19V 34.965 [pool-1-thread-10] INFO com.mmall.concurrency.example.aqs.CountDownLatchExample2-920 19mer 34.964 [pool-1-thread-1] INFO com.mmall.concurrency.example.aqs.CountDownLatchExample2-020 19V 34.965 [pool- 1-thread-8] INFO com.mmall.concurrency.example.aqs.CountDownLatchExample2-720 INFO com.mmall.concurrency.example.aqs.CountDownLatchExample2-120 pool-1-thread-2 INFO com.mmall.concurrency.example.aqs.CountDownLatchExample2-120 INFO com.mmall.concurrency.example.aqs.CountDownLatchExample2-120 INFO com.mmall.concurrency.example.aqs.CountDownLatchExample2-120 INFO com.mmall.concurrency.example.aqs.CountDownLatchExample2-120 INFO com.mmall.concurrency.example.aqs.CountDownLatchExample2-120 INFO com.mmall.concurrency.example.aqs.CountDownLatchExample2-120 INFO com.mmall.concurrency.example.aqs.CountDownLatchExample2-120 INFO com.mmall.concurrency.example.aqs.CountDownLatchExample2-120 INFO com.mmall.concurrency.example.aqs.CountDownLatchExample2-120 INFO com.mmall.concurrency.example.aqs.CountDownLatchExample2-120 INFO com.mmall.concurrency.example.aqs.CountDownLatchExample2-120 INFO com.mmall.concurrency.example.aqs.CountDownLatchExample2-120 INFO com.mmall.concurrency.example.aqs.CountDownLatchExample2-120 INFO com.mmall.concurrency.example.aqs.CountDownLatchExample2-120 INFO com.mmall.concurrency.example.aqs.CountDownLatchExample2-120 INFO com.mmall.concurrency.example.aqs.CountDownLatchExample2-120 INFO com.mmall.concurrency.example.aqs.CountDownLatchExample2-120 INFO com.mmall.concurrency.example.aqs.CountDownLatchExample2-120 INFO com.mmall.concurrency.example.aqs.CountDownLatchExample2-120 INFO com.mmall.concurrency.example.aqs.CountDownLatchExample2-120 INFO com.mmall.concurrency.example.aqs.CountDownLatchExample2-120 INFO com.mmall.concurrency.example.aqs.CountDownLatchExample2-120 INFO com.mmall.concurrency.example.aqs.CountDownLatchExample2-120 INFO com.mmall.concurrency.example.aqs.CountDownLatchExample2-120 INFO com.mmall.concurrency.example.aqs.CountDownLatchExample2-120 INFO com.mmall.concurrency.example.aqs.CountDownLatchExample2-120 INFO com.mmall.concurrency.example.aqs.CountDownLatchExample2-120 INFO com.mmall.concurrency.example.aqs.CountDownLatchExample2-120 4] INFO com.mmall.concurrency.example.aqs.CountDownLatchExample2-320 INFO com.mmall.concurrency.example.aqs.CountDownLatchExample2 19 INFO com.mmall.concurrency.example.aqs.CountDownLatchExample2 34.965 [pool-1-thread-6] INFO com.mmall.concurrency.example.aqs.CountDownLatchExample2-5Semaphore

Semaphore is similar to CountDownLatch, except that the value of Semaphore can be released after it is obtained, and it is not reduced to the end like CountDownLatch. It is also used more to limit flow, similar to valve functions. If some resources are limited to a maximum of N threads, then more than N masters are not allowed to access any more threads, and when the existing thread ends, it will be released and new threads will be allowed to come in. It is somewhat similar to the lock and unlock processes of locks. Relatively speaking, he also has two main methods:

Acquire (), which is used to obtain permissions, whose underlying implementation is similar to CountDownLatch.countdown ()

The underlying implementation of release (), which is used to release permissions, is an inverse process with acquire ().

Import lombok.extern.slf4j.Slf4j;import java.util.concurrent.CountDownLatch;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.Semaphore;@Slf4jpublic class SemaphoreExample1 {private final static int threadCount = 20; public static void main (String [] args) throws Exception {ExecutorService exec = Executors.newCachedThreadPool (); / / up to three threads at a time get permission final Semaphore semaphore = new Semaphore (3); for (int I = 0) I

< threadCount; i++) { final int threadNum = i; exec.execute(() ->

{try {semaphore.acquire (); / / obtain a license test (threadNum); semaphore.release (); / / release a license} catch (Exception e) {log.error ("exception", e) }});} exec.shutdown ();} private static void test (int threadNum) throws Exception {log.info ("{}", threadNum); Thread.sleep (1000);}} import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.Semaphore;@Slf4jpublic class SemaphoreExample2 {private final static int threadCount = 20 Public static void main (String [] args) throws Exception {ExecutorService exec = Executors.newCachedThreadPool (); final Semaphore semaphore = new Semaphore (3); for (int I = 0; I

< threadCount; i++) { final int threadNum = i; exec.execute(() ->

{try {semaphore.acquire (3); / / obtain multiple licenses test (threadNum); semaphore.release (3); / / release multiple licenses} catch (Exception e) {log.error ("exception", e) }});} exec.shutdown ();} private static void test (int threadNum) throws Exception {log.info ("{}", threadNum); Thread.sleep (1000);}} import lombok.extern.slf4j.Slf4j;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.Semaphore;import java.util.concurrent.TimeUnit @ Slf4jpublic class SemaphoreExample3 {private final static int threadCount = 20; public static void main (String [] args) throws Exception {ExecutorService exec = Executors.newCachedThreadPool (); final Semaphore semaphore = new Semaphore (3); for (int I = 0; I

< threadCount; i++) { final int threadNum = i; exec.execute(() ->

{try {if (semaphore.tryAcquire ()) {/ / attempt to obtain a license test (threadNum); semaphore.release () / / release a license}} catch (Exception e) {log.error ("exception", e);}});} exec.shutdown ();} private static void test (int threadNum) throws Exception {log.info ("{}", threadNum) Thread.sleep (1000);}} import lombok.extern.slf4j.Slf4j;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.Semaphore;import java.util.concurrent.TimeUnit;@Slf4jpublic class SemaphoreExample4 {private final static int threadCount = 20; public static void main (String [] args) throws Exception {ExecutorService exec = Executors.newCachedThreadPool (); final Semaphore semaphore = new Semaphore (3); for (int I = 0; I

< threadCount; i++) { final int threadNum = i; exec.execute(() ->

{try {if (semaphore.tryAcquire (5000, TimeUnit.MILLISECONDS)) {/ / attempt to obtain a license test (threadNum); semaphore.release () / / release a license}} catch (Exception e) {log.error ("exception", e);}});} exec.shutdown ();} private static void test (int threadNum) throws Exception {log.info ("{}", threadNum) Thread.sleep (1000);}} CyclicBarrier

CyclicBarrier is also a synchronization helper class that allows a group of threads to wait for each other until a common barrier point (common barrier point) is reached. Through it, multiple threads can wait for each other, and only when each thread is ready can each continue to perform the following operations. Similar to CountDownLatch, it is also implemented through counters. When a thread calls the await method, the thread enters the waiting state, and the counter is added by 1. When the value of the counter reaches the set initial value, all threads that enter the waiting state due to calling await are awakened to continue to perform subsequent operations. Because CycliBarrier can be reused after releasing waiting threads, it is called circular barrier. CycliBarrier supports an optional Runnable that runs once after the value of the counter reaches the set value (but before releasing all threads). Note that Runnable runs only one at each barrier point.

The usage scenario is similar to the difference between CountDownLatch and CountDownLatch

CountDownLatch mainly realizes that 1 or N threads need to wait for other threads to complete an operation before they can continue to execute the operation, which describes the relationship between 1 thread or N threads waiting for other threads. CyclicBarrier mainly realizes that multiple threads wait for each other until all threads meet the conditions before they can continue to perform subsequent operations, describing the relationship between multiple threads waiting for each other.

CountDownLatch is disposable, while CyclicBarrier can be reset and reused.

@ Slf4jpublic class CyclicBarrierExample1 {private static CyclicBarrier barrier = new CyclicBarrier (5); public static void main (String [] args) throws Exception {ExecutorService executor = Executors.newCachedThreadPool (); for (int I = 0; I

< 10; i++) { final int threadNum = i; Thread.sleep(1000); executor.execute(() ->

{try {race (threadNum);} catch (Exception e) {log.error ("exception", e);}});} executor.shutdown ();} private static void race (int threadNum) throws Exception {Thread.sleep (1000) Log.info ("{} is ready", threadNum); barrier.await (); log.info ("{} continue", threadNum);}}

Results: ready ready.. Go

20 is ready20:24:35.610 24 pool-1-thread-1 34.616 [pool-1-thread-2] INFO com.mmall.concurrency.example.aqs.CyclicBarrierExample1-1 is ready20:24:36.610 [pool-1-thread-3] INFO com.mmall.concurrency.example.aqs.CyclicBarrierExample1-2 is ready20:24:37.611 [pool-1-thread-4] INFO com.mmall.concurrency.example.aqs. CyclicBarrierExample1-3 is ready20:24:38.612 [pool-1-thread-5] INFO com.mmall.concurrency.example.aqs.CyclicBarrierExample1-4 is ready20:24:38.612 [pool-1-thread-1] INFO com.mmall.concurrency.example.aqs.CyclicBarrierExample1-0 continue20:24:38.612 [pool-1-thread-2] INFO com.mmall.concurrency.example.aqs.CyclicBarrierExample1-1 continue20:24:38.612 [pool-1-thread-5] INFO com.mmall.concurrency.example. Aqs.CyclicBarrierExample1-4 continue20:24:38.612 [pool-1-thread-4] INFO com.mmall.concurrency.example.aqs.CyclicBarrierExample1-3 continue20:24:38.612 [pool-1-thread-3] INFO com.mmall.concurrency.example.aqs.CyclicBarrierExample1-2 continue20:24:39.614 [pool-1-thread-6] INFO com.mmall.concurrency.example.aqs.CyclicBarrierExample1-5 is ready20:24:40.613 [pool-1-thread-5] INFO com.mmall.concurrency.example .aqs.CyclicBarrier INFO com.mmall 1-6 is ready20:24:41.614 [pool-1-thread-2] INFO com.mmall.concurrency.example.aqs.CyclicBarrierExample1-7 is ready20:24:42.615 [pool-1-thread-4] INFO com.mmall.concurrency.example.aqs.CyclicBarrierExample1-8 is ready20:24:43.615 [pool-1-thread-3] INFO com.mmall.concurrency.example.aqs.CyclicBarrierExample1-9 is ready20:24:43.615 [pool-1-thread-3] INFO com.mmall .concurrency.example.aqs.CyclicBarrier continue20:24:43.615 1-9 continue20:24:43.615 [pool-1-thread-6] INFO com.mmall.concurrency.example.aqs.CyclicBarrierExample1-5 continue20:24:43.615 [pool-1-thread-5] INFO com.mmall.concurrency.example.aqs.CyclicBarrierExample1-6 continue20:24:43.615 [pool-1-thread-2] INFO com.mmall.concurrency.example.aqs.CyclicBarrierExample1-7 continue20:24:43.615 [pool-1-thread-4] INFO com.mmall .concurrency.example.aqs.CyclicBarrierExample1-8 continueimport lombok.extern.slf4j.Slf4j Import java.util.concurrent.CyclicBarrier;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.TimeUnit;@Slf4jpublic class CyclicBarrierExample2 {private static CyclicBarrier barrier = new CyclicBarrier (5); public static void main (String [] args) throws Exception {ExecutorService executor = Executors.newCachedThreadPool (); for (int I = 0; I

< 10; i++) { final int threadNum = i; Thread.sleep(1000); executor.execute(() ->

{try {race (threadNum);} catch (Exception e) {log.error ("exception", e);}});} executor.shutdown ();} private static void race (int threadNum) throws Exception {Thread.sleep (1000) Log.info ("{} is ready", threadNum); try {barrier.await (2000, TimeUnit.MILLISECONDS);} catch (Exception e) {log.warn ("BarrierException", e);} log.info ("{} continue", threadNum);}} import lombok.extern.slf4j.Slf4j;import java.util.concurrent.CyclicBarrier;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors @ Slf4jpublic class CyclicBarrierExample3 {private static CyclicBarrier barrier = new CyclicBarrier (5, ()-> {log.info ("callback is running");}); public static void main (String [] args) throws Exception {ExecutorService executor = Executors.newCachedThreadPool (); for (int I = 0; I

< 10; i++) { final int threadNum = i; Thread.sleep(1000); executor.execute(() ->

{try {race (threadNum);} catch (Exception e) {log.error ("exception", e);}});} executor.shutdown ();} private static void race (int threadNum) throws Exception {Thread.sleep (1000) Log.info ("{} is ready", threadNum); barrier.await (); log.info ("{} continue", threadNum);}}

Results:

20 INFO com.mmall.concurrency.example.aqs.CyclicBarrierExample3 28 pool-1-thread-1 32.790 [pool-1-thread-1] INFO com.mmall.concurrency.example.aqs.CyclicBarrierExample3-0 is ready20:28:33.785 [pool-1-thread-2] INFO com.mmall.concurrency.example.aqs.CyclicBarrierExample3-1 is ready20:28:34.786 [pool-1-thread-3] INFO com.mmall.concurrency.example.aqs.CyclicBarrierExample3-2 is ready20:28:35.787 [pool-1-thread-4] INFO com.mmall.concurrency.example.aqs. CyclicBarrierExample3-3 is ready20:28:36.787 [pool-1-thread-5] INFO com.mmall.concurrency.example.aqs.CyclicBarrierExample3-4 is ready20:28:36.787 [pool-1-thread-5] INFO com.mmall.concurrency.example.aqs.CyclicBarrierExample3-callback is running20:28:36.787 [pool-1-thread-5] INFO com.mmall.concurrency.example.aqs.CyclicBarrierExample3-4 continue20:28:36.788 [pool-1-thread-1] INFO com.mmall.concurrency.example .aqs.CyclicBarrier Example3-0 continue20:28:36.788 [pool-1-thread-2] INFO com.mmall.concurrency.example.aqs.CyclicBarrierExample3-1 continue20:28:36.788 [pool-1-thread-3] INFO com.mmall.concurrency.example.aqs.CyclicBarrierExample3-2 continue20:28:36.788 [pool-1-thread-4] INFO com.mmall.concurrency.example.aqs.CyclicBarrierExample3-3 continue20:28:37.788 [pool-1-thread-6] INFO com.mmall.concurrency.example .aqs.CyclicBarrier is ready20:28:38.789 3-5 is ready20:28:38.789 [pool-1-thread-4] INFO com.mmall.concurrency.example.aqs.CyclicBarrierExample3-6 is ready20:28:39.789 [pool-1-thread-5] INFO com.mmall.concurrency.example.aqs.CyclicBarrierExample3-7 is ready20:28:40.790 [pool-1-thread-2] INFO com.mmall.concurrency.example.aqs.CyclicBarrierExample3-8 is ready20:28:41.791 [pool-1-thread-3] INFO com.mmall .concurrency.example.aqs.CyclicBarrierExample3-9 is ready20:28:41.791 [pool-1-thread-3] INFO com.mmall.concurrency.example.aqs.CyclicBarrierExample3-callback is running20:28:41.791 [pool-1-thread-3] INFO com.mmall.concurrency.example.aqs.CyclicBarrierExample3-9 continue20:28:41.791 [pool-1-thread-6] INFO com.mmall.concurrency.example.aqs.CyclicBarrierExample3-5 continue20:28:41.791 [pool-1-thread-4] INFO com .mmall.concurrency.example.aqs.CyclicBarrierExample3-6 continue20:28:41.818 [pool-1-thread-2] INFO com.mmall.concurrency.example.aqs.CyclicBarrierExample3-8 continue20:28:41.818 [pool-1-thread-5] INFO com.mmall.concurrency.example.aqs.CyclicBarrierExample3-7 c underlying implementation

1.CountDownLatch uses shared locks at the bottom, and it has an inner class Sync, which inherits from AQS and implements shared locks.

Refer to the review of the JUC series-CountDownLatch underlying principles and examples

Briefly draw the implementation of the shared lock.

For example, there are four threads waiting in the queue, and the node types are all shared locks. The thread Thread1 in the next node of the head node is awakened. The head node becomes the next node of the previous head node, and then repeats the operation. This process is a propagation process that wakes up threads in each shared node in turn.

two。 Semaphore, another utility class under the concurrent package, is also implemented using shared locks. But the only difference between it and CountDownLatch is that it does not wake up all threads in the shared node, but the maximum number of threads it can wake up (determined by the available size of the semaphore).

The underlying 3.CyclicBarrier uses ReentrantLock and the conditional object Condition of this lock.

At this point, the study of "what is the principle and function of CountDownLatch, Semaphore and CyclicBarrier" is over. I hope to be able to solve your doubts. The collocation of theory and practice can better help you learn, go and try it! If you want to continue to learn more related knowledge, please continue to follow the website, the editor will continue to work hard to bring you more practical articles!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.