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

快来打我* 2022-12-20 03:22 88阅读 0赞

### Semaphore ###

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

Semaphore对于信号量的控制是基于AQS（AbstractQueuedSynchronizer）来做的。Semaphore有一个内部类Sync继承了AQS。而且Semaphore中还有两个内部类FairSync和NonfairSync继承Sync，也就是说Semaphore有公平锁和非公平锁之分。以下是Semaphore中内部类的结构：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5NTEzNDMw_size_16_color_FFFFFF_t_70_pic_center]  
**两个构造函数**

public Semaphore(int permits) {
        
            sync = new NonfairSync(permits);
        }
    public Semaphore(int permits, boolean fair) {
        
            sync = fair ? new FairSync(permits) : new NonfairSync(permits);
        }

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

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

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

**semaphore在aqs里面的原理图**  
aqs的同步队列和条件队列  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5NTEzNDMw_size_16_color_FFFFFF_t_70_pic_center 1]  
semaphore使用的是共享模式  
![\*\*加粗样式\*\*][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5NTEzNDMw_size_16_color_FFFFFF_t_70_pic_center 2]

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

public final void acquireSharedInterruptibly(int arg)
                throws InterruptedException {
        
            if (Thread.interrupted())
                throw new InterruptedException();
            if (tryAcquireShared(arg) < 0)
                doAcquireSharedInterruptibly(arg);
        }

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

NonfairSync.tryAcquireShared()

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

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

FairSync.tryAcquireShared()

protected int tryAcquireShared(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()

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

**2，release()释放信号**

public final boolean releaseShared(int arg) {
        
            if (tryReleaseShared(arg)) {
        
                doReleaseShared();
                return true;
            }
            return false;
        }

tryReleaseShared()

protected boolean tryReleaseShared(int releases) {
        
                // Decrement count; signal when transition to zero
                for (;;) {
        
                    int c = getState();
                    if (c == 0)
                        return false;
                    int nextc = c-1;
                    if (compareAndSetState(c, nextc))
                        return nextc == 0;
                }
            }

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

private void doReleaseShared() {
        
            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）的，将其设置为“传播”状态。
                         * 意味着需要将状态向后一个节点传播
                         */
                    }
                    else if (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简单示例：

public class SemaphoreSample {
        
    
        public static void main(String[] args) {
        
            Semaphore semaphore = new Semaphore(2);
            for (int i=0;i<5;i++){
        
                new Thread(new Task(semaphore,"task+"+i)).start();
            }
        }
    
        static class Task extends Thread{
        
            Semaphore semaphore;
    
            public Task(Semaphore semaphore,String tname){
        
                this.semaphore = semaphore;
                this.setName(tname);
            }
    
            public void run() {
        
                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

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

QueueTask

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

测试：

public class CountDownLaunchSample {
        
    
        public static void main(String[] args) throws InterruptedException {
        
            long now = System.currentTimeMillis();
            CountDownLatch countDownLatch = new CountDownLatch(2);
            new Thread(new SeeDoctorTask(countDownLatch)).start();
            new Thread(new QueueTask(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应用场景例子**  
比如特工一起执行团体任务，少一个人都不行，要等到所有特工准备好才能开始执行。  
代码如下：

public class CyclicBarrierTest implements Runnable {
        
        private CyclicBarrier cyclicBarrier;
        private int index ;
    
        public CyclicBarrierTest(CyclicBarrier cyclicBarrier, int index) {
        
            this.cyclicBarrier = cyclicBarrier;
            this.index = index;
        }
    
        public void run() {
        
            try {
        
                System.out.println("index: " + index);
                index--;
                cyclicBarrier.await();
            } catch (Exception e) {
        
                e.printStackTrace();
            }
        }
    
        public static void main(String[] args) throws Exception {
        
            CyclicBarrier cyclicBarrier = new CyclicBarrier(11, new Runnable() {
        
                public void run() {
        
                    System.out.println("所有特工到达屏障，准备开始执行秘密任务");
                }
            });
            for (int i = 0; i < 10; i++) {
        
                new Thread(new CyclicBarrierTest(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应用场景极少，以下是代码试例：

public class ExchangerTest {
        
    
        public static void main(String []args) {
        
            final Exchanger<Integer> exchanger = new Exchanger<Integer>();
            for(int i = 0 ; i < 10 ; i++) {
        
                final Integer num = i;
                new Thread() {
        
                    public void run() {
        
                        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 后面会详解线程池**

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5NTEzNDMw_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/c9cf8a804f854293afe4e1bb42c1c198.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5NTEzNDMw_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221120/4378713b10c947aeafdf211831e4cc1a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5NTEzNDMw_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221120/f4c761efe6934bfe93d7d403b5ebe916.png