并发编程之Tools&CountDownLatch&Semaphore详解

Semaphore

Semaphore字面意思是信号量，它的作用是控制访问特定资源的线程数目，Semaphore在实例化时传入一个int值，也就是指明信号数量。主要方法有两个：acquire()和release()。acquire()用于请求信号，每调用一次，信号量便少一个。release()用于释放信号，调用一次信号量加一个。信号量用完以后，后续使用acquire()方法请求信号的线程便会加入阻塞队列挂起。

Semaphore对于信号量的控制是基于AQS（AbstractQueuedSynchronizer）来做的。Semaphore有一个内部类Sync继承了AQS。而且Semaphore中还有两个内部类FairSync和NonfairSync继承Sync，也就是说Semaphore有公平锁和非公平锁之分。以下是Semaphore中内部类的结构：

两个构造函数

publicSemaphore(int permits){
        sync=newNonfairSync(permits);}publicSemaphore(int permits,boolean fair){
        sync= fair?newFairSync(permits):newNonfairSync(permits);}

默认是非公平锁。两个构造方法都必须传int permits值。

这个int值在实例化内部类时，被设置为AQS中的state。

Sync(int permits){setState(permits);}

semaphore在aqs里面的原理图
aqs的同步队列和条件队列

semaphore使用的是共享模式

1，acquire()获取信号
内部类Sync调用AQS中的acquireSharedInterruptibly()方法

publicfinalvoidacquireSharedInterruptibly(int arg)throws InterruptedException{if(Thread.interrupted())thrownewInterruptedException();if(tryAcquireShared(arg)<0)doAcquireSharedInterruptibly(arg);}

• 调用tryAcquireShared()方法尝试获取信号。
• 如果没有可用信号，将当前线程加入等待队列并挂起
  tryAcquireShared()方法被Semaphore的内部类NonfairSync和FairSync重写，实现有一些区别。

NonfairSync.tryAcquireShared()

finalintnonfairTryAcquireShared(int acquires){for(;;){// 自旋int available=getState();int remaining= available- acquires;if(remaining<0||// 判断资源是否小于0compareAndSetState(available, remaining))return remaining;}}

可以看到，非公平锁对于信号的获取是直接使用CAS进行尝试的。

FairSync.tryAcquireShared()

protectedinttryAcquireShared(int acquires){for(;;){if(hasQueuedPredecessors())return-1;int available=getState();int remaining= available- acquires;if(remaining<0||compareAndSetState(available, remaining))return remaining;}}

• 先调用hasQueuedPredecessors()方法，判断队列中是否有等待线程。如果有，直接返回-1，表示没有可用信号
• 队列中没有等待线程，再使用CAS尝试更新state，获取信号

再看看acquireSharedInterruptibly()方法中，如果没有可用信号加入队列的方法doAcquireSharedInterruptibly()

privatevoiddoAcquireSharedInterruptibly(int arg)throws InterruptedException{final Node node=addWaiter(Node.SHARED);// 1boolean failed=true;try{for(;;){final Node p= node.predecessor();if(p== head){// 2int r=tryAcquireShared(arg);if(r>=0){setHeadAndPropagate(node, r);
                        p.next= null;// help GC
                        failed=false;return;}}if(shouldParkAfterFailedAcquire(p, node)&&// 3parkAndCheckInterrupt())thrownewInterruptedException();}}finally{if(failed)cancelAcquire(node);}}

2，release()释放信号

publicfinalbooleanreleaseShared(int arg){if(tryReleaseShared(arg)){doReleaseShared();returntrue;}returnfalse;}

tryReleaseShared()

protectedbooleantryReleaseShared(int releases){// Decrement count; signal when transition to zerofor(;;){int c=getState();if(c==0)returnfalse;int nextc= c-1;if(compareAndSetState(c, nextc))return nextc==0;}}

这里也就是直接使用CAS算法，将state也就是可用信号加1
doReleaseShared()

privatevoiddoReleaseShared(){for(;;){
            Node h= head;if(h!= null&& h!= tail){int ws= h.waitStatus;if(ws== Node.SIGNAL){//head是SIGNAL状态/* head状态是SIGNAL，重置head节点waitStatus为0，这里不直接设为Node.PROPAGATE,
                     * 是因为unparkSuccessor(h)中，如果ws < 0会设置为0，所以ws先设置为0，再设置为PROPAGATE
                     * 这里需要控制并发，因为入口有setHeadAndPropagate跟release两个，避免两次unpark
                     */if(!compareAndSetWaitStatus(h, Node.SIGNAL,0))continue;//设置失败，重新循环/* head状态为SIGNAL，且成功设置为0之后,唤醒head.next节点线程
                     * 此时head、head.next的线程都唤醒了，head.next会去竞争锁，成功后head会指向获取锁的节点，
                     * 也就是head发生了变化。看最底下一行代码可知，head发生变化后会重新循环，继续唤醒head的下一个节点
                     */unparkSuccessor(h);/*
                     * 如果本身头节点的waitStatus是出于重置状态（waitStatus==0）的，将其设置为“传播”状态。
                     * 意味着需要将状态向后一个节点传播
                     */}elseif(ws==0&&!compareAndSetWaitStatus(h,0, Node.PROPAGATE))continue;// loop on failed CAS}if(h== head)//如果head变了，重新循环break;}}

• 把当前结点设置为SIGNAL或者PROPAGATE
• 唤醒head.next(B节点)，B节点唤醒后可以竞争锁，成功后head->B，然后又会唤醒B.next，一直重复直到共享节点都唤醒
• head节点状态为SIGNAL，重置head.waitStatus->0，唤醒head节点线程，唤醒后线程去竞争共享锁
• head节点状态为0，将head.waitStatus->Node.PROPAGATE传播状态，表示需要将状态向后继节点传播

Semaphore简单示例：

publicclassSemaphoreSample{publicstaticvoidmain(String[] args){
        Semaphore semaphore=newSemaphore(2);for(int i=0;i<5;i++){newThread(newTask(semaphore,"task+"+i)).start();}}staticclassTaskextendsThread{
        Semaphore semaphore;publicTask(Semaphore semaphore,String tname){this.semaphore= semaphore;this.setName(tname);}publicvoidrun(){try{
                semaphore.acquire();
                System.out.println(Thread.currentThread().getName()+":aquire() at time:"+System.currentTimeMillis());

                Thread.sleep(1000);

                semaphore.release();
                System.out.println(Thread.currentThread().getName()+":aquire() at time:"+System.currentTimeMillis());}catch(InterruptedException e){
                e.printStackTrace();}}}}

打印结果：

Thread-3:aquire() at time:1604998004183
Thread-1:aquire() at time:1604998004183
Thread-1:aquire() at time:1604998005184
Thread-5:aquire() at time:1604998005184
Thread-3:aquire() at time:1604998005184
Thread-7:aquire() at time:1604998005184
Thread-7:aquire() at time:1604998006185
Thread-9:aquire() at time:1604998006185
Thread-5:aquire() at time:1604998006185
Thread-9:aquire() at time:1604998007185

Process finished with exit code 0

从打印结果可以看出，虽然开启了5个线程，一次只有2个线程执行 acquire()，只有线程进行 release() 方法后 才会有别的线程执行 acquire()。

CountDownLatch

CountDownLatch这个类能够使一个线程等待其他线程完成各自的工作后再执行。例 如，应用程序的主线程希望在负责启动框架服务的线程已经启动所有的框架服务之后再执行。

CountDownLatch如何工作?
CountDownLatch是通过一个计数器来实现的，计数器的初始值为线程的数量。每当 一个线程完成了自己的任务后，计数器的值就会减1。当计数器值到达0时，它表示所有的 线程已经完成了任务，然后在闭锁上等待的线程就可以恢复执行任务。

其实CountDownlatch原理和Semaphore原理都差不多这里就不多赘述了，讲讲用法吧

API
CountDownLatch.countDown()
CountDownLatch.await();

CountDownLatch应用场景例子
比如陪媳妇去看病。 医院里边排队的人很多，如果一个人的话，要先看大夫，看完大夫再去排队交钱取药。 现在我们是双核，可以同时做这两个事(多线程)。 假设看大夫花3秒钟，排队交费取药花5秒钟。我们同时搞的话，5秒钟我们就能完成，然后 一起回家(回到主线程)。
代码如下:
SeeDoctorTask

publicclassSeeDoctorTaskimplementsRunnable{private CountDownLatch countDownLatch;publicSeeDoctorTask(CountDownLatch countDownLatch){this.countDownLatch= countDownLatch;}publicvoidrun(){try{
            System.out.println("开始看医生");
            Thread.sleep(3000);
            System.out.println("看医生结束，准备离开病房");}catch(InterruptedException e){
            e.printStackTrace();}finally{if(countDownLatch!= null)
                countDownLatch.countDown();}}}

QueueTask

publicclassQueueTaskimplementsRunnable{private CountDownLatch countDownLatch;publicQueueTask(CountDownLatch countDownLatch){this.countDownLatch= countDownLatch;}publicvoidrun(){try{
            System.out.println("开始在医院药房排队买药....");
            Thread.sleep(5000);
            System.out.println("排队成功，可以开始缴费买药");}catch(InterruptedException e){
            e.printStackTrace();}finally{if(countDownLatch!= null)
                countDownLatch.countDown();}}}

测试：

publicclassCountDownLaunchSample{publicstaticvoidmain(String[] args)throws InterruptedException{long now= System.currentTimeMillis();
        CountDownLatch countDownLatch=newCountDownLatch(2);newThread(newSeeDoctorTask(countDownLatch)).start();newThread(newQueueTask(countDownLatch)).start();//等待线程池中的2个任务执行完毕，否则一直
        countDownLatch.await();
        System.out.println("over，回家 cost:"+(System.currentTimeMillis()-now));}}

打印结果：

开始看医生
开始在医院药房排队买药....
看医生结束，准备离开病房
排队成功，可以开始缴费买药
over，回家 cost:5005

Process finished with exit code 0

CyclicBarrier

栅栏屏障，让一组线程到达一个屏障(也可以叫同步点)时被阻塞，直到最后一个线程 到达屏障时，屏障才会开门，所有被屏障拦截的线程才会继续运行。
CyclicBarrier默认的构造方法是CyclicBarrier(int parties)，其参数表示屏障拦截的线 程数量，每个线程调用await方法告CyclicBarrier我已经到达了屏障，然后当前线程被阻塞。

API
cyclicBarrier.await();

CyclicBarrier应用场景例子
比如特工一起执行团体任务，少一个人都不行，要等到所有特工准备好才能开始执行。
代码如下：

publicclassCyclicBarrierTestimplementsRunnable{private CyclicBarrier cyclicBarrier;privateint index;publicCyclicBarrierTest(CyclicBarrier cyclicBarrier,int index){this.cyclicBarrier= cyclicBarrier;this.index= index;}publicvoidrun(){try{
            System.out.println("index: "+ index);
            index--;
            cyclicBarrier.await();}catch(Exception e){
            e.printStackTrace();}}publicstaticvoidmain(String[] args)throws Exception{
        CyclicBarrier cyclicBarrier=newCyclicBarrier(11,newRunnable(){publicvoidrun(){
                System.out.println("所有特工到达屏障，准备开始执行秘密任务");}});for(int i=0; i<10; i++){newThread(newCyclicBarrierTest(cyclicBarrier, i)).start();}
        cyclicBarrier.await();
        System.out.println("全部到达屏障....");}}

打印结果：

index: 1
index: 6
index: 5
index: 0
index: 2
index: 4
index: 3
index: 9
index: 8
index: 7
所有特工到达屏障，准备开始执行秘密任务
全部到达屏障....

Process finished with exit code 0

Exchanger

Exchanger（交换者）是一个用于线程间协作的工具类。Exchanger用于进行线程间的数据交换。它提供一个同步点，在这个同步点两个线程可以交换彼此的数据。这两个线程通过exchange方法交换数据， 如果第一个线程先执行exchange方法，它会一直等待第二个线程也执行exchange，当两个线程都到达同步点时，这两个线程就可以交换数据，将本线程生产出来的数据传递给对方。因此使用Exchanger的重点是成对的线程使用exchange()方法，当有一对线程达到了同步点，就会进行交换数据。因此该工具类的线程对象是成对的。

Exchanger应用场景极少，以下是代码试例：

publicclassExchangerTest{publicstaticvoidmain(String[]args){final Exchanger<Integer> exchanger=newExchanger<Integer>();for(int i=0; i<10; i++){final Integer num= i;newThread(){publicvoidrun(){
                    System.out.println("我是线程：Thread_"+this.getName()+"我的数据是："+ num);try{
                        Integer exchangeNum= exchanger.exchange(num);
                        Thread.sleep(1000);
                        System.out.println("我是线程：Thread_"+this.getName()+"我原先的数据为："+ num+" , 交换后的数据为："+ exchangeNum);}catch(InterruptedException e){
                        e.printStackTrace();}}}.start();}}}

打印结果：

我是线程：Thread_Thread-1我的数据是：1
我是线程：Thread_Thread-6我的数据是：6
我是线程：Thread_Thread-7我的数据是：7
我是线程：Thread_Thread-5我的数据是：5
我是线程：Thread_Thread-8我的数据是：8
我是线程：Thread_Thread-4我的数据是：4
我是线程：Thread_Thread-3我的数据是：3
我是线程：Thread_Thread-0我的数据是：0
我是线程：Thread_Thread-2我的数据是：2
我是线程：Thread_Thread-9我的数据是：9
我是线程：Thread_Thread-4我原先的数据为：4 , 交换后的数据为：8
我是线程：Thread_Thread-5我原先的数据为：5 , 交换后的数据为：7
我是线程：Thread_Thread-8我原先的数据为：8 , 交换后的数据为：4
我是线程：Thread_Thread-6我原先的数据为：6 , 交换后的数据为：1
我是线程：Thread_Thread-0我原先的数据为：0 , 交换后的数据为：3
我是线程：Thread_Thread-1我原先的数据为：1 , 交换后的数据为：6
我是线程：Thread_Thread-3我原先的数据为：3 , 交换后的数据为：0
我是线程：Thread_Thread-7我原先的数据为：7 , 交换后的数据为：5
我是线程：Thread_Thread-2我原先的数据为：2 , 交换后的数据为：9
我是线程：Thread_Thread-9我原先的数据为：9 , 交换后的数据为：2

Process finished with exit code 0

Executors

主要用来创建线程池，代理了线程池的创建，使得你的创建入口参数变得简单
重要方法

• newCachedThreadPool创建一个可缓存线程池，如果线程池长度超过处理需
  要，可灵活回收空闲线程，若无可回收，则新建线程。
• newFixedThreadPool 创建一个定长线程池，可控制线程最大并发数，超出的
  线程会在队列中等待。
• newScheduledThreadPool 创建一个定长线程池，支持定时及周期性任务执
  行。
• newSingleThreadExecutor 创建一个单线程化的线程池，它只会用唯一的工作
  线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。

// TODO 后面会详解线程池