Android 并发之CountDownLatch、Cyclic
2019-03-04 本文已影响0人
有没有口罩给我一个
Android 线程简单分析（一）
Android 并发之synchronized锁住的是代码还是对象（二）
Android 并发之CountDownLatch、CyclicBarrier的简单应用（三）
Android 并发HashMap和ConcurrentHashMap的简单应用（四）(待发布)
Android 并发之Lock、ReadWriteLock和Condition的简单应用(五)
Android 并发之CAS（原子操作）简单介绍（六）
Android 并发Kotlin协程的重要性（七）(待发布)
Android 并发之AsyncTask原理分析（八）(待发布)
Android 并发之Handler、Looper、MessageQueue和ThreadLocal消息机制原理分析（九）
Android 并发之HandlerThread和IntentService原理分析（十）
CountDownLatch：
它可以实现类似计数器的功能。比如有一个任务A，它要等待其他4个任务执行完毕之后才能执行，此时就可以利用CountDownLatch来实现这种功能了。
CountDownLatch.png
构造函数：
public CountDownLatch(int count) {
}
参数count为计数值
然后下面这3个方法是CountDownLatch类中最重要的方法：
public void await() throws InterruptedException { }; //调用await()方法的线程，如果count>0会被挂起，它会等待直到 count值为0才继续执行
public boolean await(long timeout, TimeUnit unit) throws InterruptedException { }; //和await()类似，只不过等待一定的时间后count值还没变为0的话就会继续执行
public void countDown() { }; //将count值减1
场景
类似的场景，项目开发流程，将各个功能模块分发至每个人手里，每个人做完及时汇报，当所有人都开发完成，那么接下来就是发布上线，CountDownLatch就是等待一组任务完成之后，继续执行。
举个栗子：
创建Project线程类CountDownLatchProject，并声明一个CountDownLatch属性mCountDownLatch来控制Project项目线程等待所有开发者开发完成。项目线程启动后会调用await()方法并进入等待状态，每一个开发者开发完成后调用complete()方法，并把mCountDownLatch中的计数器减1，当计数器等于0的时候项目线程继续执行；
项目线程类：
public class CountDownLatchProject implements Runnable {
private CountDownLatch mCountDownLatch;
public CountDownLatchProject(int number) {
this.mCountDownLatch = new CountDownLatch(number);
}
public void complete(String name) {
Log.e("tag", name + "完成...............");
mCountDownLatch.countDown();
}
@Override
public void run() {
Log.e("tag", "项目开始..................");
try {
mCountDownLatch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
Log.e("tag", "项目上线.....................");
}
}
开发者类：
public class CountDownLatchdevalper implements Runnable {
private CountDownLatchProject mCountDownLatchProject;
public CountDownLatchdevalper(CountDownLatchProject count) {
this.mCountDownLatchProject = count;
}
@Override
public void run() {
Log.e("tag", "" + Thread.currentThread().getName() + "开始开发.............");
try {
long duration = (long) (Math.random() * 20);
TimeUnit.SECONDS.sleep(duration);
} catch (InterruptedException e) {
e.printStackTrace();
}
mCountDownLatchProject.complete(Thread.currentThread().getName());
}
}
创建一个项目开发，并初始化10个人参加项目开发，当所有开发人员开发完成，项目进入下一阶段：
public class MainActivity extends AppCompatActivity {
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
CountDownLatchProject project = new CountDownLatchProject(10);
new Thread(project).start();
for (int i = 0; i < 10; i++) {
CountDownLatchdevalper count = new CountDownLatchdevalper(project);
new Thread(count).start();
}
}
}
查看日志：
CountDownLatchProject： 项目开始..................
CountDownLatchdevalper： Thread-232开始了............
CountDownLatchdevalper： Thread-228开始了............
CountDownLatchdevalper： Thread-229开始了............
CountDownLatchdevalper： Thread-231开始了............
CountDownLatchdevalper： Thread-230开始了............
CountDownLatchdevalper： Thread-233开始了............
CountDownLatchdevalper： Thread-234开始了............
CountDownLatchdevalper： Thread-236开始了............
CountDownLatchdevalper： Thread-235开始了............
CountDownLatchdevalper： Thread-237开始了............
CountDownLatchProject: Thread-233完成...............
Thread-233CountDownLatchProject： 还有: 9 人未完成
CountDownLatchProject: Thread-232完成...............
Thread-232CountDownLatchProject： 还有: 8 人未完成
CountDownLatchProject: Thread-228完成...............
Thread-228CountDownLatchProject： 还有: 7 人未完成
CountDownLatchProject: Thread-237完成...............
Thread-237CountDownLatchProject： 还有: 6 人未完成
CountDownLatchProject: Thread-230完成...............
Thread-230CountDownLatchProject： 还有: 5 人未完成
CountDownLatchProject: Thread-235完成...............
Thread-235CountDownLatchProject： 还有: 4 人未完成
CountDownLatchProject: Thread-229完成...............
Thread-229CountDownLatchProject： 还有: 3 人未完成
CountDownLatchProject: Thread-231完成...............
Thread-231CountDownLatchProject： 还有: 2 人未完成
CountDownLatchProject: Thread-234完成...............
Thread-234CountDownLatchProject： 还有: 1 人未完成
CountDownLatchProject: Thread-236完成...............
Thread-236CountDownLatchProject： 还有: 0 人未完成
CountDownLatchProject： 项目上线....................
- CountDownLatch是当前线程等待其他一组线程任务完成之后在继续往下执行；
- CountDownLatch并不是用来保护共享资源同步访问的，而是用来控制并发线程等待的；
- CountDownLatch只允许使用一次，一旦内部计数器等于0，再调用这个方法将不起作用，如果还有第二次并发等待，你还得创建一个新的CountDownLatch。
- 有一个任务A，它要等待其他4个任务执行完毕之后才能执行，此时就可以利用CountDownLatch来实现这种功能了
CyclicBarrier：
字面意思循环屏障，通过它可以实现让一组线程等待至某个状态之后再全部同时执行。叫做循环是因为当所有等待线程都执行完成以后，CyclicBarrier可以被重用。我们暂且把这个状态就叫做屏障，当调用await()方法之后，线程就处在屏障点了，屏障就是要阻碍的意思，它要做的事情是让一组线程到达一个屏障（同步点）时被阻塞，直到最后一个线程到达屏障时候，屏障才会开门。所有被屏障拦截的线程才会运行。
CyclicBarrier.png
构造函数：
1：
public CyclicBarrier(int parties, Runnable barrierAction) {
}
2：
public CyclicBarrier(int parties) {
}
参数parties指让多少个线程或者任务等待至barrier状态；参数barrierAction为当这些线程都达到barrier状态时会执行的内容。
然后CyclicBarrier中最重要的方法就是await方法：
public int await() throws InterruptedException, BrokenBarrierException { };//来挂起当前线程，直至所有线程都到达barrier状态再同时执行后续任务；
public int await(long timeout, TimeUnit unit)throws InterruptedException,BrokenBarrierException,TimeoutException { };//让这些线程等待至一定的时间，如果还有线程没有到达barrier状态就直接让到达barrier的线程执行后续任务
举个栗子：
public class Statistic implements Runnable {
private CyclicBarrier mCyclicBarrier;
public Statistic(CyclicBarrier parties) {
this.mCyclicBarrier = parties;
}
@Override
public void run() {
Log.e("tag", "mCyclicBarrier: " + Thread.currentThread().getName() + "开始" + mCyclicBarrier.getNumberWaiting());
try {
long duration = (long) (Math.random() * 20);
TimeUnit.SECONDS.sleep(duration);
mCyclicBarrier.await();
} catch (BrokenBarrierException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
Log.e("tag", "mCyclicBarrier: " + Thread.currentThread().getName() + "结束");
}
}
创建3个Statistic线程类：
public class MainActivity extends AppCompatActivity {
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
CyclicBarrier cyclicBarrier = new CyclicBarrier(3);
for (int i = 0; i < 3; i++) {
new Thread(new Statistic(cyclicBarrier)).start();
}
}
}
查看日志：
mCyclicBarrier: Thread-143开始0
mCyclicBarrier: Thread-144开始0
mCyclicBarrier: Thread-142开始0
mCyclicBarrier: Thread-143结束
mCyclicBarrier: Thread-144结束
mCyclicBarrier: Thread-142结束
从上面输出结果可以看出，每个线程执行完操作之后，就在等待其他线程的操作完毕，然后一起执行，即，当所有线程线程写入操作完毕之后，所有线程就继续进行后续的操作了。
如果说想要在所有线程执行完之后，进行额外的其他操作可以为使用CyclicBarrier构造函数提供Runnable参数；
CyclicBarrier cyclicBarrier = new CyclicBarrier(3, new Runnable() {
@Override
public void run() {
Log.e("tag", "执行完，做额外的操作,然后接着后续的操作。");
}
});
for (int i = 0; i < 3; i++) {
new Thread(new Statistic(cyclicBarrier)).start();
}
查看日志：
mCyclicBarrier: Thread-154开始0
mCyclicBarrier: Thread-152开始0
mCyclicBarrier: Thread-153开始0
执行完，做额外的操作,然后接着后续的操作，Thread-154。
mCyclicBarrier: Thread-152结束
mCyclicBarrier: Thread-153结束
mCyclicBarrier: Thread-154结束
当3个线程都到达barrier状态后，会从3个线程中选择最后一个到达同步点的线程去执行Runnable，然后继续后续操作，注意：CyclicBarrier可以复用，而CountDownLatch无法进行重复使用；
看看CycliBarrier中等待其他线程的源码：
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) { // 最后一个线程到达屏障点
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();//断裂当前代，如果要复用，必须调用reset方法
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();
}
}
private void nextGeneration() {
trip.signalAll(); //唤醒所有等待的线程
count = parties;
generation = new Generation();
}
private void breakBarrier() {
generation.broken = true;//标志当前代断裂或被破坏
count = parties;
trip.signalAll();//唤醒所有等待的线程
}
- 其实CyclicBarrier中使用了ReentrantLock 独享锁或者可重入锁；
- 当前线程执行await方法是其实是调用ReentrantLock 的await方法，进入条件等待队里，并释放锁;
- 当所有的线程都到达屏障点，会选择最后一个到达屏障点的线程执行barrierCommand，然后调用ReentrantLock .signalAll唤醒所有线程，线程将进入同步队列，竞争锁，ReentrantLock （非公平锁），并调用nextGeneration()重置Generation，所以正常情况下重用CyclicBarrier可以不用调用reset方法，出现异常情况就需要调用reset方法重置状态，CyclicBarrier是可以复用的，而CountDownLatch无法进行重复使用；
