CountDownLatch
CountDownLatch的典型使用环境是:确保某个操作在其需要的一个多线程操作完成后执行。比如说赛跑者必须等发令枪响。await是阻塞,countDown是倒计,构造函数中指定初始值。只有初始值被countDown到0时,阻塞才会被打开。
//所有线程阻塞,然后统一开始
CountDownLatch begin = new CountDownLatch(1);
//主线程阻塞,直到所有分线程执行完毕
CountDownLatch end = new CountDownLatch(5);
for(int i = 0; i < 5; i++){
Thread thread = new Thread(new Runnable() {
@Override
public void run() {
try {
begin.await();
System.out.println(Thread.currentThread().getName() + " 起跑");
Thread.sleep(1000);
System.out.println(Thread.currentThread().getName() + " 到达终点");
end.countDown();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
thread.start();
}
try {
System.out.println("1秒后统一开始");
Thread.sleep(1000);
begin.countDown(); //发令枪响
end.await();
System.out.println("停止比赛");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
CountDownLatch可以当成一个带数量的join。允许1或者多个线程,等待另外N个线程完成某件事情之后,这1个或者多个线程才能执行。
CyclicBarrier
与CountDownLatch不同的是该barrier在释放等待线程后可以重用,所以称它为循环(Cyclic)的屏障(Barrier)。
设想这个场景,一群工人要在AB两地往返工作,负责接送他们的是一个大巴,这个大巴不等坐满就不发车。
public class JUCMain {
public static Object object = new Object();
public static void main(String[] args) {
//System.out.println("begin");
JUCMain main = new JUCMain();
main.testCB();
//System.out.println("end");
}
public static class WorkerThread implements Runnable{
CyclicBarrier barrier;
public WorkerThread(CyclicBarrier b){
this.barrier = b;
}
@Override
public void run() {
int i = 10;
// TODO Auto-generated method stub
try{
while (i>0){
System.out.println("ID:"+Thread.currentThread().getId()+" Waiting");
//线程在这里等待,直到所有线程都到达barrier。
barrier.await();
System.out.println("ID:"+Thread.currentThread().getId()+" Working");
i--;
barrier.await();
}
}catch(Exception e){
e.printStackTrace();
}
}
}
public void testCB() {
// TODO Auto-generated method stub
CyclicBarrier cb = new CyclicBarrier(5, new Runnable() {
//当所有线程到达barrier时执行
@Override
public void run() {
// TODO Auto-generated method stub
System.out.println("Inside Barrier");
}
});
for(int i=0;i<5;i++){
new Thread(new WorkerThread(cb)).start();
}
}
上述代码输出:
ID:10 Waiting
ID:11 Waiting
ID:12 Waiting
ID:13 Waiting
ID:14 Waiting
Inside Barrier
ID:14 Working
ID:10 Working
ID:12 Working
ID:13 Working
ID:11 Working
Inside Barrier
ReadWriteLock
ReadWriteLock是一个接口,常见的实现是ReentrantReadWriteLock。
public static void main(String[] args) {
final JUCMain main = new JUCMain();
// 建N个线程,同时读
ExecutorService service = Executors.newCachedThreadPool();
service.execute(new Runnable() {
@Override
public void run() {
main.readFile(Thread.currentThread());
}
});
// 建N个线程,同时写
ExecutorService service1 = Executors.newCachedThreadPool();
service1.execute(new Runnable() {
@Override
public void run() {
main.writeFile(Thread.currentThread());
}
});
System.out.println("begin");
}
ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
public void readFile(Thread thread){
lock.readLock().lock();
boolean readLock = lock.isWriteLocked();
if(!readLock){
System.out.println("当前为读锁!");
}
try{
for(int i=0; i<5; i++){
try {
Thread.sleep(20);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(thread.getName() + ":正在进行读操作……");
}
System.out.println(thread.getName() + ":读操作完毕!");
}finally{
System.out.println("释放读锁!");
lock.readLock().unlock();
}
}
// 写操作
public void writeFile(Thread thread){
lock.writeLock().lock();
boolean writeLock = lock.isWriteLocked();
if(writeLock){
System.out.println("当前为写锁!");
}
try{
for(int i=0; i<5; i++){
try {
Thread.sleep(20);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(thread.getName() + ":正在进行写操作……");
}
System.out.println(thread.getName() + ":写操作完毕!");
}finally{
System.out.println("释放写锁!");
lock.writeLock().unlock();
}
}
Semaphore
Semaphore的原意是旗语,技术文档中常翻译成信号量。
private static Semaphore semaphore = new Semaphore(2);
public static void main(String[] args) {
ExecutorService service = Executors.newFixedThreadPool(5);
for (int i = 0; i < 4; i++) {
service.execute(new Runnable() {
@Override
public void run() {
try{
System.out.println("require_permit"+Thread.currentThread().getId());
semaphore.acquire();
System.out.println("begin"+Thread.currentThread().getId());
Thread.sleep(1000);
System.out.println("end"+Thread.currentThread().getId());
}catch (Exception e){
e.printStackTrace();
}finally {
semaphore.release();
System.out.println("release_permit"+Thread.currentThread().getId());
}
}
});
}
}
上述程序的输出是
require_permit10
require_permit12
begin10
require_permit11
begin12
require_permit13
end10
end12
release_permit10
begin11
release_permit12
begin13
end11
end13
release_permit11
release_permit13