In Java 1.5, there are several helper classes that are very useful for concurrent programming, such as CountDownLatch, CyclicBarrier, and Semaphore. Today we will learn how to use these helper classes.
The following is an outline of the table of contents:
- CountDownLatch usage
- CyclicBarrier usage
- Semaphore usage
CountDownLatch usage
The CountDownLatch class, located under the java.util.Concurrent package, enables count-like functionality. For example, if task A is waiting for the other four tasks to complete, CountDownLatch can be used to implement this feature.
The CountDownLatch class provides only one constructor:
public CountDownLatch(int count) {};// The argument count is the count value
Copy the codeThe following three methods are the most important in the CountDownLatch class:
public void await(a) throws InterruptedException {};// The thread calling the await() method is suspended and waits until count is 0 before continuing
public boolean await(long timeout, TimeUnit unit) throws InterruptedException {};// Similar to await(), except to wait a certain amount of time before count has changed to 0
public void countDown(a) {};// Subtract count by 1
Copy the codeHere’s an example of how CountDownLatch can be used:
public class Test {
public static void main(String[] args) {
final CountDownLatch latch = new CountDownLatch(2);
new Thread(){
public void run(a) {
try {
System.out.println(Child thread+Thread.currentThread().getName()+"In process");
Thread.sleep(3000);
System.out.println(Child thread+Thread.currentThread().getName()+"Executed");
latch.countDown();
} catch(InterruptedException e) { e.printStackTrace(); }}; }.start();new Thread(){
public void run(a) {
try {
System.out.println(Child thread+Thread.currentThread().getName()+"In process");
Thread.sleep(3000);
System.out.println(Child thread+Thread.currentThread().getName()+"Executed");
latch.countDown();
} catch(InterruptedException e) { e.printStackTrace(); }}; }.start();try {
System.out.println("Wait for 2 child threads to complete...");
latch.await();
System.out.println("2 child threads have completed execution");
System.out.println("Continue executing the main thread");
} catch(InterruptedException e) { e.printStackTrace(); }}}Copy the codeExecution Result:
Thread thread0 Executing Thread thread1 executing Wait for two child threads to finish executing... Thread thread-0 completion Thread Thread-1 completion Thread Two child threads have completed and continue to execute the main ThreadCopy the codeCyclicBarrier usage
Literally loopback fence, which allows a group of threads to wait for a certain state before all execute simultaneously. Cyclicbarriers are called loops because they can be reused once all waiting threads have been released. We’ll call this state a barrier for now, and the thread is in a barrier after calling the await() method.
The CyclicBarrier class is located under the java.util.concurrent package and CyclicBarrier provides two constructors:
public CyclicBarrier(int parties, Runnable barrierAction) {}public CyclicBarrier(int parties) {}Copy the codeThe parties parameter specifies how many threads or tasks to wait until the barrier state is reached. The barrierAction argument is what will be executed when these threads all reach the Barrier state.
Then the most important method in CyclicBarrier is the await method, which has two overloaded versions:
public int await(a) throws InterruptedException, BrokenBarrierException {};public int await(long timeout, TimeUnit unit)throws InterruptedException,BrokenBarrierException,TimeoutException {};Copy the codeThe first version is commonly used to suspend the current thread until all threads reach the barrier state and then execute subsequent tasks simultaneously.
The second version makes these threads wait for a certain amount of time, and if there are any threads that haven’t reached the Barrier state, the threads that have reached the barrier do the subsequent tasks.
Here are a few examples:
A CyclicBarrier can be used if several threads need to write data and all threads need to write data before they can proceed:
public class Test {
public static void main(String[] args) {
int N = 4;
CyclicBarrier barrier = new CyclicBarrier(N);
for(int i=0; i<N; i++)new Writer(barrier).start();
}
static class Writer extends Thread{
private CyclicBarrier cyclicBarrier;
public Writer(CyclicBarrier cyclicBarrier) {
this.cyclicBarrier = cyclicBarrier;
}
@Override
public void run(a) {
System.out.println("Thread"+Thread.currentThread().getName()+"Writing data...");
try {
Thread.sleep(5000); // Use sleep to simulate writing data
System.out.println("Thread"+Thread.currentThread().getName()+"Write data complete, wait for other threads to write data complete.");
cyclicBarrier.await();
} catch (InterruptedException e) {
e.printStackTrace();
}catch(BrokenBarrierException e){
e.printStackTrace();
}
System.out.println("All threads write, continue processing other tasks..."); }}}Copy the codeExecution Result:
Thread thread-0 is writing data... Thread thread-3 is writing data... Thread thread-2 is writing data... Thread thread-1 is writing data... Thread thread-2 completes data writing and waits for other threads to complete data writing Thread thread0 completes data writing and waits for other threads to complete data writing Thread thread-3 completes data writing and waits for other threads to complete data writing Thread thread-1 completes data writing, Wait for other threads to finish writing all threads finish writing and continue processing other tasks... All threads write, continue processing other tasks... All threads write, continue processing other tasks... All threads write, continue processing other tasks...Copy the codeAs you can see from the output above, after each writing thread has finished writing data, it is waiting for other threads to finish writing data.
When all threads have finished writing, all threads continue to perform subsequent operations.
If you want to perform additional operations on a CyclicBarrier after all thread writes are complete, you can provide the Runnable parameter to CyclicBarrier:
public class Test {
public static void main(String[] args) {
int N = 4;
CyclicBarrier barrier = new CyclicBarrier(N,new Runnable() {
@Override
public void run(a) {
System.out.println("Current thread"+Thread.currentThread().getName()); }});for(int i=0; i<N; i++)new Writer(barrier).start();
}
static class Writer extends Thread{
private CyclicBarrier cyclicBarrier;
public Writer(CyclicBarrier cyclicBarrier) {
this.cyclicBarrier = cyclicBarrier;
}
@Override
public void run(a) {
System.out.println("Thread"+Thread.currentThread().getName()+"Writing data...");
try {
Thread.sleep(5000); // Use sleep to simulate writing data
System.out.println("Thread"+Thread.currentThread().getName()+"Write data complete, wait for other threads to write data complete.");
cyclicBarrier.await();
} catch (InterruptedException e) {
e.printStackTrace();
}catch(BrokenBarrierException e){
e.printStackTrace();
}
System.out.println("All threads write, continue processing other tasks..."); }}}Copy the codeRunning results:
Thread thread-0 is writing data... Thread thread-1 is writing data... Thread thread-2 is writing data... Thread thread-3 is writing data... Thread thread-1 completes data writing and waits for other threads to complete data writing Thread thread-2 completes data writing and waits for other threads to complete data writing Thread thread-3 completes data writing Wait until other threads finish writing. Current Thread THREAD-3 All threads finish writing and continue processing other tasks. All threads write, continue processing other tasks... All threads write, continue processing other tasks... All threads write, continue processing other tasks...Copy the codeAs you can see from the result, when all four threads reach the barrier state, one of the four threads is selected to execute the Runnable.
Let’s look at the effect of specifying time for await:
public class Test {
public static void main(String[] args) {
int N = 4;
CyclicBarrier barrier = new CyclicBarrier(N);
for(int i=0; i<N; i++) {if(i<N-1)
new Writer(barrier).start();
else {
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
newWriter(barrier).start(); }}}static class Writer extends Thread{
private CyclicBarrier cyclicBarrier;
public Writer(CyclicBarrier cyclicBarrier) {
this.cyclicBarrier = cyclicBarrier;
}
@Override
public void run(a) {
System.out.println("Thread"+Thread.currentThread().getName()+"Writing data...");
try {
Thread.sleep(5000); // Use sleep to simulate writing data
System.out.println("Thread"+Thread.currentThread().getName()+"Write data complete, wait for other threads to write data complete.");
try {
cyclicBarrier.await(2000, TimeUnit.MILLISECONDS);
} catch (TimeoutException e) {
// TODO Auto-generated catch blocke.printStackTrace(); }}catch (InterruptedException e) {
e.printStackTrace();
}catch(BrokenBarrierException e){
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName()+"All threads write, continue processing other tasks..."); }}}Copy the codeExecution Result:
Thread thread-0 is writing data... Thread thread-2 is writing data... Thread thread-1 is writing data... Thread THREAD-2 Finished writing data and waiting for other threads to finish writing data Thread THREAD0 finished writing data and waiting for other threads to finish writing data Thread THREAD-1 finished writing data and waiting for other threads to finish writing data Thread THREAD-3 is writing data... Java. Util. Concurrent. TimeoutException all threads in the Thread - 1, continue to deal with other tasks... Thread-0 After all threads are written, continue processing other tasks... at java.util.concurrent.CyclicBarrier.dowait(Unknown Source) at java.util.concurrent.CyclicBarrier.await(Unknown Source) at com.cxh.test1.Test$Writer.run(Test.java:58)
java.util.concurrent.BrokenBarrierException
at java.util.concurrent.CyclicBarrier.dowait(Unknown Source)
at java.util.concurrent.CyclicBarrier.await(Unknown Source)
at com.cxh.test1.Test$Writer.run(Test.java:58)
java.util.concurrent.BrokenBarrierException
at java.util.concurrent.CyclicBarrier.dowait(Unknown Source)
at java.util.concurrent.CyclicBarrier.await(Unknown Source)
at com.cxh.test1.Test$Writer.run(test.java :58) thread-2 After all threads are written, continue to process other tasks... Java. Util. Concurrent. BrokenBarrierException Thread Thread - 3 finished writing data, After waiting for another thread to write at Java. Util. Concurrent. CyclicBarrier. Dowait (Unknown Source) at java.util.concurrent.CyclicBarrier.await(Unknown Source) at com.cxh.test1.Test$Writer.run(test.java :58) thread-3Copy the codeThe code above purposely delays the start of the last thread in the for loop of the main method, because after the first three threads have reached the barrier, the fourth thread has not reached the barrier at the specified time, and then throws an exception and continues.
Cyclicbarriers can also be reused, as shown in the following example:
public class Test {
public static void main(String[] args) {
int N = 4;
CyclicBarrier barrier = new CyclicBarrier(N);
for(int i=0; i<N; i++) {new Writer(barrier).start();
}
try {
Thread.sleep(25000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("CyclicBarrier reuse");
for(int i=0; i<N; i++) {newWriter(barrier).start(); }}static class Writer extends Thread{
private CyclicBarrier cyclicBarrier;
public Writer(CyclicBarrier cyclicBarrier) {
this.cyclicBarrier = cyclicBarrier;
}
@Override
public void run(a) {
System.out.println("Thread"+Thread.currentThread().getName()+"Writing data...");
try {
Thread.sleep(5000); // Use sleep to simulate writing data
System.out.println("Thread"+Thread.currentThread().getName()+"Write data complete, wait for other threads to write data complete.");
cyclicBarrier.await();
} catch (InterruptedException e) {
e.printStackTrace();
}catch(BrokenBarrierException e){
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName()+"All threads write, continue processing other tasks..."); }}}Copy the codeExecution Result:
Thread thread-0 is writing data... Thread thread-1 is writing data... Thread thread-3 is writing data... Thread thread-2 is writing data... Thread thread-1 completes data writing and waits for other threads to complete data writing Thread Thread-3 completes data writing and waits for other threads to complete data writing Thread thread-2 completes data writing and waits for other threads to complete data writing Thread thread0 completes data writing, Wait until other threads finish writing thread-0 All threads finish writing and continue processing other tasks... Thread-3 After all threads are written, continue processing other tasks... Thread-1 After all threads are written, continue processing other tasks... Thread-2 After all threads are written, continue processing other tasks... CyclicBarrier reuse Thread thread-4 is writing data... Thread thread-5 is writing data... Thread thread-6 is writing data... Thread thread-7 is writing data... Thread THREAD-5 data writing is complete. Thread thread6 data writing is complete. Thread thread-4 data writing is complete. Wait until other threads finish writing thread4 After all threads finish writing, continue processing other tasks... Thread-5 After all threads are written, continue processing other tasks... Thread-6 After all threads are written, continue to process other tasks... Thread-7 After all threads are written, continue processing other tasks...Copy the codeAs can be seen from the execution result, after the first four threads have crossed the barrier state, they can be used for another round of use. CountDownLatch cannot be reused.
Semaphore usage
Semaphore can control the number of threads accessed at the same time, acquire a license through acquire(), wait if not, and release() releases a license.
The Semaphore class, located under the java.util.concurrent package, provides two constructors:
public Semaphore(int permits) { // Permitting indicates the number of permits, that is, how many threads are allowed to access at the same time
sync = new NonfairSync(permits);
}
public Semaphore(int permits, boolean fair) { // If it is fair, the longer you wait, the sooner you get the license
sync = (fair)? new FairSync(permits) : new NonfairSync(permits);
}
Copy the codeAcquire (); release(); Semaphore ();
public void acquire(a) throws InterruptedException {}// Get a license
public void acquire(int permits) throws InterruptedException {}// Obtain permits
public void release(a) {}// Release a license
public void release(int permits) {}// Release permitting
Copy the code**acquire()** Is used to acquire a license, and if no license can be obtained, wait until the license is obtained.
**release()** to release permissions. Note that permission must be obtained before release.
All four methods block. If you want immediate results, you can use the following methods:
public boolean tryAcquire(a) {};// Try to obtain a license, return true immediately on success, false immediately on failure
public boolean tryAcquire(long timeout, TimeUnit unit) throws InterruptedException {};// Try to obtain a license, return true immediately if successful within the specified time, false immediately otherwise
public boolean tryAcquire(int permits) {};// Attempts to obtain permits, true immediately if successful, false immediately if failed
public boolean tryAcquire(int permits, long timeout, TimeUnit unit) throws InterruptedException {};// Attempt to obtain permits immediately true if successful within the specified time, otherwise immediately false
Copy the codeThe number of availablePermits can also be obtained via the **availablePermits()** method.
Here’s an example of Semaphore in action:
If a factory has five machines but eight workers, each machine can only be used by one worker at a time, until it is used up, other workers can continue to use it. We can do this with Semaphore:
public class Test {
public static void main(String[] args) {
int N = 8; / / the number of workers
Semaphore semaphore = new Semaphore(5); // Number of machines
for(int i=0; i<N; i++)new Worker(i,semaphore).start();
}
static class Worker extends Thread{
private int num;
private Semaphore semaphore;
public Worker(int num,Semaphore semaphore){
this.num = num;
this.semaphore = semaphore;
}
@Override
public void run(a) {
try {
semaphore.acquire();
System.out.println("Workers"+this.num+"Occupy a machine in production...");
Thread.sleep(2000);
System.out.println("Workers"+this.num+"Unleash the machine.");
semaphore.release();
} catch(InterruptedException e) { e.printStackTrace(); }}}}Copy the codeExecution Result:
Worker 0 occupies a machine in production... Worker 1 occupies a machine in production... Worker 2 occupies a machine in production... Worker 4 occupies a machine in production... Worker 5 occupies a machine in production... Worker 0 releases robo 2 releases Robo 3 occupies a machine in production... Worker 7 occupies a machine during production... Worker 4 unleashing robo 5 Unleashing Robo 1 Unleashing Robo 6 occupying a machine in production... Worker 3 releases the robot worker 7 releases the robot worker 6 releases the machineCopy the codeconclusion
Here’s a summary of the three helper classes:
- Both CountDownLatch and CyclicBarrier implement waits between threads, but they have different priorities:
CountDownLatch is typically used when thread A waits for several other threads to complete their tasks before it executes.
A CyclicBarrier is typically used when a group of threads wait for each other to reach a state and then execute simultaneously.
In addition, countdownlatches are not reusable, while cyclicBarriers are.
- Semaphore is similar to a lock in that it is used to control access to a set of resources.