CountDownLatch
允许一个或多个线程等待其他线程完成操作。
public static void main(String[] args) throws InterruptedException {
final CountDownLatch c = new CountDownLatch(2);
new Thread(new Runnable() {
@Override
public void run() {
c.countDown();
System.out.println("线程1执行完成");
}
}).start();
new Thread(new Runnable() {
@Override
public void run() {
c.countDown();
System.out.println("线程2执行完成");
}
}).start();
c.await();
System.out.println("退出");
}
结果:
线程1执行完成
线程2执行完成
退出
CountDownLatch的构造函数接收一个,其中构造参数n表示等待n个完成,即执行countDown n次。
CyclicBarrier
让一组线程到达一个屏障(也可以叫同步点)时被阻塞,直到最后一个线程到达屏障
public static void main(String[] args) {
new Thread(new Runnable() {
@Override
public void run() {
try {
c.await();
} catch (Exception e) {
}
System.out.println(1);
}
}).start();
try {
c.await();
} catch (Exception e) {
}
System.out.println(2);
}
当两个线程都执行c.await()时,表示当前屏障拦截到两个线程,比较CyclicBarrier拦截参数,如果已经达到同步标准,两个线程继续执行,例如:new CyclicBarrier(2);如果count!=0,两个线程继续等待。
源码分析:
public CyclicBarrier(int parties, Runnable barrierAction) {
if (parties <= 0) throw new IllegalArgumentException();
//在barrier被触发之前必须调用的线程数
this.parties = parties;
this.count = parties;
//在barrier被触发时调用action
this.barrierCommand = barrierAction;
}
public CyclicBarrier(int parties) {
this(parties, null);
}
public int await() throws InterruptedException, BrokenBarrierException {
try {
return dowait(false, 0L);
} catch (TimeoutException toe) {
throw new Error(toe); // cannot happen
}
}
}
private int dowait(boolean timed, long nanos)
throws InterruptedException, BrokenBarrierException,
TimeoutException {
//.....
int index = --count;
if (index == 0) { // tripped
//....
nextGeneration();
return 0;
//....
}
trip.await();
//.....
}
private void nextGeneration() {
// signal completion of last generation
trip.signalAll();
// set up next generation
count = parties;
generation = new Generation();
}
这里只做简要分析。
Semaphore
Semaphore(信号量)是用来控制同时访问特定资源的线程数量,它通过协调各个线程,以保证合理的使用公共资源。
Semaphore的构造方法Semaphore(int permits)接受一个整型的数字,表示可用的许可证数量。Semaphore(10)表示允许10个线程获取许可证,也就是最大并发数是10。
Semaphore的acquire()方法获取一个许可证,使用完之后调用release()方法归还许可证。还可以用tryAcquire()方法尝试获取许可证。
public static void main(String[] args) {
ExecutorService threadPool = Executors.newFixedThreadPool(10);
final Semaphore s = new Semaphore(5);
for (int i = 0; i < 30; i++) {
threadPool.execute(new Runnable() {
@Override
public void run() {
try {
s.acquire();
System.out.println("save data");
s.release();
} catch (InterruptedException e) {
}
}
});
}
threadPool.shutdown();
}
分析:当前线程池线程数量最大线程数为10,需要执行的任务为30,同一时间被执行 的任务许可大小为5。也就是说任务执行的并发受线程数和许可影响,虽然线程数时10,但是同一时间也最多只能执行5个任务。
Semaphore其他方法:
Int availablePermits():返回此信号量中当前可用的许可证数。
Int getQueueLength():返回正在等待获取许可证的线程数。
Boolean hasQueuedThreads():是否有线程正在等待获取许可证。
void reducePermits(int reduction):减少reduction个许可证,是个protected方法。
Collection getQueuedThreads():返回所有等待获取许可证的线程集合,是个protected方法。
Exchanger
Exchanger(交换者)是一个用于线程间协作的工具类。Exchanger用于进行线程间的数据交换。它提供一个同步点,在这个同步点,两个线程可以交换彼此的数据。这两个线程通过exchange方法交换数据,如果第一个线程先执行exchange()方法,它会一直等待第二个线程也执行exchange方法,当两个线程都到达同步点时,这两个线程就可以交换数据,将本线程生产出来的数据传递给对方。
public static void main(String[] args) {
final Exchanger<String> exgr = new Exchanger<String>();
ExecutorService threadPool = Executors.newFixedThreadPool(2);
threadPool.execute(new Runnable() {
@Override
public void run() {
try {
String A = "银行流水A";// A录入银行流水数据
exgr.exchange(A);
} catch (InterruptedException e) {
}
}
});
threadPool.execute(new Runnable() {
@Override
public void run() {
try {
String B = "银行流水B";// B录入银行流水数据
String A = exgr.exchange("B");
System.out.println("A和B数据是否一致:" + A.equals(B) + ",A录入的是:" + A + ",B录入是:" + B);
} catch (InterruptedException e) {
}
}
});
threadPool.shutdown();
}