CountDownLatch
CountDownLatch是计数器，作用是阻塞主线程，直到若干子线程都执行完毕，再继续执行主线程，比如阻塞餐厅的服务员（不让他开门），等到所有顾客都赶到了（顾客是子线程，赶路是线程的操作），服务员才开门。
CyclicBarrier
CyclicBarrier是循环栅栏，作用是：一个或多个线程都执行完一部分操作后，再一起执行另一部分操作，比如所有顾客都赶到餐厅后，服务员才开门，然后顾客才可以一起吃饭；
CyclicBarrier也可以达到CountDownLatch的效果；
CyclicBarrier可以循环利用，CountDownLatch不行。
Semaphore
Semaphore意为信号量，用于限制访问某个资源的线程数，比如限制同一时间餐厅只能容纳三个顾客。
CountDownLatch使用示例
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.util.concurrent.*;
/**
* CountDownLatch示例类
*
* CountDownLatch作用：
* CountDownLatch是一个计数器，通过构造函数传入count，表示有count个子线程
* 用await方法阻塞主线程
* 每个子线程执行完毕后，通过countDown函数把计数器-1，countDown为0表示子线程执行完毕，主线程继续执行
*
* ThreadPoolExecutor的awaitTermination()能达到同样的效果
*
*/
public class CountDownLatchDemo {
public static void main(String[] args) throws Exception{
System.out.println("主线程开始执行");
int limit = 5;
CountDownLatch countDownLatch = new CountDownLatch(limit);
ThreadFactory factory = new ThreadFactoryBuilder().setNameFormat("CountDownLatchDemo-pool-%d").build();
ThreadPoolExecutor pool = new ThreadPoolExecutor(limit, limit, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingDeque<>(1024), factory, new ThreadPoolExecutor.AbortPolicy());
for (int i = 0; i < limit; i++) {
pool.execute(new Runnable() {
@Override
public void run() {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
synchronized (countDownLatch) {
countDownLatch.countDown();
System.out.println("子线程开始执行，name: " + Thread.currentThread().getName() + ", countDownLatch.getCount(): " + countDownLatch.getCount());
}
}
});
}
pool.shutdown();
countDownLatch.await();
System.out.println("子线程执行完毕，主线程继续执行");
}
}
输出结果
主线程开始执行
子线程开始执行，name: CountDownLatchDemo-pool-2, countDownLatch.getCount(): 4
子线程开始执行，name: CountDownLatchDemo-pool-1, countDownLatch.getCount(): 3
子线程开始执行，name: CountDownLatchDemo-pool-0, countDownLatch.getCount(): 2
子线程开始执行，name: CountDownLatchDemo-pool-3, countDownLatch.getCount(): 1
子线程开始执行，name: CountDownLatchDemo-pool-4, countDownLatch.getCount(): 0
子线程执行完毕，主线程继续执行
CyclicBarrier示例代码
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.util.concurrent.*;
/**
* CyclicBarrier示例类
*
* CyclicBarrier作用：
* CyclicBarrier意为循环栅栏，栅栏相当于一个门，一开始栅栏关闭着，所有子线程先执行一部分操作，都到达栅栏后，栅栏才开启，然后所有子线程再一起执行下一部分操作。
* 可以把CyclicBarrier比喻成餐厅的门，所有顾客都到了，餐厅才开门，然后顾客才可以吃饭。
* 之所以叫循环栅栏，是因为CyclicBarrier的reset方法可以重置栅栏。
*
*/
public class CyclicBarrierDemo {
public static void main(String[] args) {
int limit = 5;
CyclicBarrier cyclicBarrier = new CyclicBarrier(5, new Runnable() {
@Override
public void run() {
// 栅栏开启时，主线程的操作
System.out.println("开启栅栏");
}
});
ThreadFactory factory = new ThreadFactoryBuilder().setNameFormat("CyclicBarrierDemo-pool-%d").build();
ThreadPoolExecutor pool = new ThreadPoolExecutor(limit, limit, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingDeque<>(1024), factory, new ThreadPoolExecutor.AbortPolicy());
for (int i = 0; i < limit; i++) {
pool.execute(new Runnable() {
@Override
public void run() {
try {
// 栅栏开启前子线程的操作
Thread.sleep(1000);
System.out.println("子线程" + Thread.currentThread().getName() + "开始等待");
// await方法会阻塞子线程至栅栏开启
cyclicBarrier.await();
// 栅栏开启后子线程的操作
System.out.println("子线程" + Thread.currentThread().getName() + "开始执行");
} catch (Exception e) {
e.printStackTrace();
}
}
});
}
pool.shutdown();
}
}
输出结果
子线程CyclicBarrierDemo-pool-1开始等待
子线程CyclicBarrierDemo-pool-0开始等待
子线程CyclicBarrierDemo-pool-4开始等待
子线程CyclicBarrierDemo-pool-2开始等待
子线程CyclicBarrierDemo-pool-3开始等待
开启栅栏
子线程CyclicBarrierDemo-pool-3开始执行
子线程CyclicBarrierDemo-pool-1开始执行
子线程CyclicBarrierDemo-pool-2开始执行
子线程CyclicBarrierDemo-pool-4开始执行
子线程CyclicBarrierDemo-pool-0开始执行
Semaphore示例代码
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.util.concurrent.*;
/**
* Semaphore示例类
*
* Semaphore作用：
* Semaphore意为信号量，用于限制访问某个资源的线程数
*
*/
public class SemaphoreDemo {
public static void main(String[] args) {
Semaphore semaphore = new Semaphore(3);
int threadCount = 10;
ThreadFactory factory = new ThreadFactoryBuilder().setNameFormat("SemaphoreDemo-pool-%d").build();
ThreadPoolExecutor pool = new ThreadPoolExecutor(threadCount, threadCount, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingDeque<>(1024), factory, new ThreadPoolExecutor.AbortPolicy());
for (int i = 0; i < threadCount; i++) {
pool.execute(new Runnable() {
@Override
public void run() {
try {
// 获取信号量
semaphore.acquire();
System.out.println("子线程" + Thread.currentThread().getName() + "开始执行");
Thread.sleep(1000);
System.out.println("子线程" + Thread.currentThread().getName() + "执行完毕");
} catch (Exception e) {
e.printStackTrace();
} finally {
// release方法放到finally，确保可以释放信号量
semaphore.release();
}
}
});
}
pool.shutdown();
}
}
输出结果
子线程SemaphoreDemo-pool-0开始执行
子线程SemaphoreDemo-pool-1开始执行
子线程SemaphoreDemo-pool-2开始执行
子线程SemaphoreDemo-pool-0执行完毕
子线程SemaphoreDemo-pool-3开始执行
子线程SemaphoreDemo-pool-1执行完毕
子线程SemaphoreDemo-pool-2执行完毕
子线程SemaphoreDemo-pool-4开始执行
子线程SemaphoreDemo-pool-5开始执行
子线程SemaphoreDemo-pool-3执行完毕
子线程SemaphoreDemo-pool-4执行完毕
子线程SemaphoreDemo-pool-5执行完毕
子线程SemaphoreDemo-pool-7开始执行
子线程SemaphoreDemo-pool-6开始执行
子线程SemaphoreDemo-pool-8开始执行
子线程SemaphoreDemo-pool-8执行完毕
子线程SemaphoreDemo-pool-7执行完毕
子线程SemaphoreDemo-pool-6执行完毕
子线程SemaphoreDemo-pool-9开始执行
子线程SemaphoreDemo-pool-9执行完毕
网友评论