标签：java android concurrent 多线程 并发
CountDownLatch
CountDownLatch允许一条或者多条线程等待直至其它线程完成以系列的操作的辅助同步器。 private static final class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = 4982264981922014374L;
Sync(int count) {
setState(count);
}
int getCount() {
return getState();
}
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
protected boolean tryReleaseShared(int releases) {
for (;;) {
int c = getState();
if (c == 0)
return false;
int nextc = c-1;
if (compareAndSetState(c, nextc))
return nextc == 0;
}
}
} 构造函数调用setState把count值设置为当前的状态。内部类Sync由CountDownLatch构造函数时创建，当然保证非负值也是这里判断， public CountDownLatch(int count) {
if (count < 0) throw new IllegalArgumentException("count < 0");
this.sync = new Sync(count);
} 另外，看看CountDownLatch的的await和countDown方法的实现： public void countDown() {
sync.releaseShared(1);
}
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
} 可以看到await调了AQS的acquireSharedInterruptibly尝试获取共享锁，countDown方法调用了releaseShared尝试释放共享锁，两个方法的参数都是1，因此当此时有多条线程同时调用await时，这时候看到内部Sync类的tryAcquireShared方法实现，由于在构造函数里已经调用setState把当前锁状态数设置为count，因此这里在getState()的判断会一直返回count的值，因此tryAcquireShared会一直返回-1，然后这些调用await的线程都会进入等待队列。CyclicBarrier
CyclicBarrier允许一组线程互相等待直到一个公平屏障点（common barrier point）。与CountDownLatch不同的是CyclicBarrier着重与互相等待，并且添加重置原状态的方法。 if (barrier.await() == 0) {
// 执行屏障操作
} CyclicBarrier对于失败的同步尝试，会使用全有或者全无的破坏模型（breakage model）：如果一条线程由于中断、异常或者超时提前离开了屏障点，其它所有在屏障点等待的线程也会通过抛出BrokenBarrierException（或者InterruptedException异常，同时被中断的情况下）离开屏障点。接下来看看具体的实现。 private static class Generation {
boolean broken = false;
} CyclicBarrier声明一个内部类Generation，在每次屏障点的使用就代表着一个Generation实例。当屏障点被破坏或者重置的时候，generation就要改变。 //保护屏障点入口的锁
private final ReentrantLock lock = new ReentrantLock();
//使线程在await中等待直至屏障点被脱落（tripped）的条件对象
private final Condition trip = lock.newCondition();
//需要调用await来脱落屏障点的线程数，为final变量
private final int parties;
//在屏障点被脱落之后需要运行的命令
private final Runnable barrierCommand;
//当前的Generation
private Generation generation = new Generation();
//在每次generation中从parties递减到0，当新的Generation被创建或者屏障点被破坏的时候，count就会被重置
private int count; 成员变量中采用lock和trip进行同步控制，另外parties和count记录线程数，generation则表示屏障点的当前状态，还有barrierCommand记录屏障点破坏后需要执行的命令。 public CyclicBarrier(int parties, Runnable barrierAction) {
if (parties <= 0) throw new IllegalArgumentException();
this.parties = parties;
this.count = parties;
this.barrierCommand = barrierAction;
} 其中parties表示在屏障点被脱落（tripped）之前需要调用await的线程数，需要注意的是parties是final变量，因此不能改变，而成员变量count则在每次await的时候递减，重置的时候把parties赋值即可。barrierAction表示当屏障点被脱落的时候，执行的命令。要注意的时parties数在构造函数里设定以后就不能更改，如果屏障被脱落的时候，可以调用reset重置，我们先来看看await的实现。在具体实现里，await有两个不同的函数版本，包括无超时版本和超时版本，具体如下。 //无超时版本
public int await() throws InterruptedException, BrokenBarrierException {
try {
return dowait(false, 0L);
} catch (TimeoutException toe) {
throw new Error(toe); // cannot happen
}
}
//超时版本
public int await(long timeout, TimeUnit unit)
throws InterruptedException,
BrokenBarrierException,
TimeoutException {
return dowait(true, unit.toNanos(timeout));
}
private int dowait(boolean timed, long nanos)
throws InterruptedException, BrokenBarrierException,
TimeoutException {
final ReentrantLock lock = this.lock;
lock.lock();
try {
final Generation g = generation;
if (g.broken)
throw new BrokenBarrierException();
if (Thread.interrupted()) {
breakBarrier();
throw new InterruptedException();
}
int index = --count;
if (index == 0) { // tripped
boolean ranAction = false;
try {
final Runnable command = barrierCommand;
if (command != null)
command.run();
ranAction = true;
nextGeneration();
return 0;
} finally {
if (!ranAction)
breakBarrier();
}
}
for (;;) {
try {
if (!timed)
trip.await();
else if (nanos > 0L)
nanos = trip.awaitNanos(nanos);
} catch (InterruptedException ie) {
if (g == generation && ! g.broken) {
breakBarrier();
throw ie;
} else {
Thread.currentThread().interrupt();
}
}
if (g.broken)
throw new BrokenBarrierException();
if (g != generation)
return index;
if (timed && nanos <= 0L) {
breakBarrier();
throw new TimeoutException();
}
}
} finally {
lock.unlock();
}
} 可以看到无超时版本和超时版本的await实现只是在调用dowait函数的参数不同。具体来看看dowait的实现。和之前解析的ReentranLock，ReentrantReadWriteLock，以及上面的CountDownLatch不同，CyclicBarrier并没有重载AQS类，而是选择了直接使用ReentrantLock以及其Condition。 private void breakBarrier() {
generation.broken = true;
count = parties;
trip.signalAll();
} breakBarrier作用就是破坏屏障点，函数首先把成员变量的generation.broken变为true，重置count为parties值，然后调用条件对象trip的signalAll唤醒所有在await等待的线程。在await下面我们即将会看到，breakBarrier会造成之前在await的所有线程抛出异常。如果线程没有被中断，则把count自减，并保留至index（count同样会在后面释放锁的时候被其它线程修改），如果index为0，则表示屏障点已经被破坏，然后如果barrierCommand非null，则执行命令，如果执行成功ranAction修改为true，否则在命令里抛出任何异常的话，则会在finally块调用breakBarrier破坏屏障点，唤醒其它线程。命令执行成功后，则会调用nextGeneration（脱落当前的屏障点）： private void nextGeneration() {
trip.signalAll();
count = parties;
generation = new Generation();
} 函数先唤醒所有在等待中的线程，然后重置count，接着创建一个新的Generation类实例，这样就等于把CyclicBarrier内部的状态重置。 public void reset() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
breakBarrier(); // break the current generation
nextGeneration(); // start a new generation
} finally {
lock.unlock();
}
} 函数实现很简单，尝试获取锁，然后调用breakBarrier和nextGeneration方法。这样调用之后，屏障点就会被破坏（breakage），则把之前在await的线程唤醒并让它们抛出异常；然后调用nextGeneration重置当前状态，这样后来的await能够再次重新等待。总结
这样，我们就完整地把CountDownLatch和CyclicBarrier进行了分析。CountDownLatch着重于多组线程等待另外一组线程完成操作，并且是无法重置的；CyclicBarrier则是着重于一组线程互相等待到对方都完成操作为止，但可以重置。CountDownLatch & CyclicBarrier源码Android版实现解析,布布扣,bubuko.com
CountDownLatch & CyclicBarrier源码Android版实现解析
标签：java android concurrent 多线程 并发
原文地址：http://blog.csdn.net/pun_c/article/details/37658841
