Java并发编程(12)-AQS及其组件介绍

旧城等待， 2023-06-16 02:27 17阅读 0赞

### 文章目录 ###

*   *   *  一.AQS相关
         *   *  1.什么是AQS？
             *  2.AQS的核心原理是什么？
             *  3.使用AQS实现一个互斥锁
             *  4.AQS定义了哪两种对资源的共享方式？
             *   *  1.==Exclusive(独占)==:只有一个线程能够获取到该资源，如ReentrantLock。对于独占锁，又可分为公平锁和非公平锁：
                 *  2.==Share(共享)==:可以有多个线程获取到该资源，如Semaphore、CountDownLatch、CyclicBarrier和ReadWriteLock。
         *  二.信号量Semaphore相关
         *   *  1，什么是信号量？
             *  2. acquire()和tryAcquire()有何不同？
             *  3.Semaphore的公平模式和非公平模式了解吗？
             *   *  1.公平模式
                 *  2.非公平模式
         *  三.倒计时器(闭锁)CountDownLatch
         *   *  1.什么是CountDownLatch
             *  2.CountDownLatch有什么作用
             *  3.一个计数器使用的例子：主线程等待其他所有线程执行完毕在执行
             *  4.一个并发测试的例子：将所有的子线程全部阻塞在一个入口，等待主线程统一放行
             *  5.CountDownLatch的常用方法
             *   *  1.await() vs wait()
                 *  2.countDown() vs getCount()
         *  四.循环栅栏CyclicBarrier
         *   *  1.什么是循环栅栏？
             *  2，有什么作用
             *  3.具体使用

### 一.AQS相关 ###

https://www.cnblogs.com/waterystone/p/4920797.html

#### 1.什么是AQS？ ####

AQS全称AbstactQueueSynchronizer,即队列同步器。它使用int类型作为线程的状态，通过名为CLH的FIFO双向队列来管理等待资源的线程。是JUC包中locks包下所有锁实现的基础。

#### 2.AQS的核心原理是什么？ ####

AQS的核心原理是当线程尝试获取共享资源时，若共享资源空闲，则将该线程作为工作线程，并且将共享资源设置为锁定状态。当共享资源处于锁定状态时，则将该线程添加到CLH队列末尾（通过CAS的方式）。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9hcm9uZy5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70]

#### 3.使用AQS实现一个互斥锁 ####

AQS使用的是模板方法模式，所以我们在自己实现一个互斥锁的时候，只需要重写指定的方法即可：

*  Mutex.java

package cocurrency;
    
    import org.jetbrains.annotations.NotNull;
    
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.locks.AbstractQueuedSynchronizer;
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.Lock;
    
    public class Mutex implements Lock { 
        private static class MyAQS extends AbstractQueuedSynchronizer { 
            /* * @Author ARong * @Description 是否处于占用状态 * @Date 2019/11/23 8:53 下午 * @Param * @return **/
            @Override
            protected boolean isHeldExclusively() { 
                 return getState() == 1;
            }
    
            /* * @Author ARong * @Description 尝试获取锁 * @Date 2019/11/23 8:52 下午 * @Param [arg] * @return boolean **/
            @Override
            protected boolean tryAcquire(int arg) { 
                // 状态为0才可获取锁
                if (compareAndSetState(0, 1)) { 
                    // 使用CAS设置锁状态为1成功，设置锁的所有者
                    setExclusiveOwnerThread(Thread.currentThread());
                    return true;
                } else { 
                    return false;
                }
            }
            /* * @Author ARong * @Description 尝试释放锁 * @Date 2019/11/23 8:52 下午 * @Param * @return **/
            @Override
            protected boolean tryRelease(int arg) { 
                // 释放锁，将状态设置为0
                if (getState() == 0) { 
                    throw new IllegalMonitorStateException("当前资源没有被锁定");
                } else { 
                    // 将当前线程的锁解除
                    setExclusiveOwnerThread(null);
                    setState(0);
                    return true;
                }
            }
            Condition newCondition() { 
                return new ConditionObject();
            }
        }
    
    
        // 以下操作只需要把AQS的内部类实现类代理操作即可
        private final MyAQS aqs = new MyAQS();
        
        @Override
        public void lock() { 
            aqs.acquire(1);
        }
        @Override
        public void lockInterruptibly() throws InterruptedException { 
            aqs.acquireInterruptibly(1);
        }
        @Override
        public boolean tryLock() { 
            return aqs.tryAcquire(1);
        }
        @Override
        public boolean tryLock(long time, @NotNull TimeUnit unit) throws InterruptedException { 
            return aqs.tryAcquireNanos(1, unit.toNanos(time));
        }
        @Override
        public void unlock() { 
            aqs.release(1);
        }
        @NotNull
        @Override
        public Condition newCondition() { 
            return aqs.newCondition();
        }
    }

#### 4.AQS定义了哪两种对资源的共享方式？ ####

##### 1.Exclusive(独占):只有一个线程能够获取到该资源，如ReentrantLock。对于独占锁，又可分为公平锁和非公平锁： #####

1.  公平锁：以排队队列中的线程先后顺序为为准，先到先得资源的锁。
2.  非公平锁：无数排队顺序，谁抢占到了资源就是谁的。

##### 2.Share(共享):可以有多个线程获取到该资源，如Semaphore、CountDownLatch、CyclicBarrier和ReadWriteLock。 #####

### 二.信号量Semaphore相关 ###

https://blog.csdn.net/qq\_19431333/article/details/70212663

synchronized 和 ReentrantLock 都是一次只允许一个线程访问某个资源，Semaphore(信号量)可以指定多个线程同时访问某个资源。

#### 1，什么是信号量？ ####

信号量规定了一定数量的许可证，在许可证不够用的情况下，获取许可证就会阻塞直到有可用的许可证；再使用完许可证后，可以释放掉许可证。信号量最大的作用就是限制获取资源的线程数量。

*  一个信号量的使用Demo

public class SemaphoreDemo { 
    
        // 请求的数量
        private static final int threadCount = 550;
    
        public static void main(String[] args) throws InterruptedException { 
            // 创建一个具有固定线程数量的线程池对象（如果这里线程池的线程数量给太少的话你会发现执行的很慢）
            ExecutorService threadPool = Executors.newFixedThreadPool(300);
            // 一次只能允许执行的线程数量。
            final Semaphore semaphore = new Semaphore(20);
    
            for (int i = 0; i < threadCount; i++) { 
                final int threadnum = i;
                threadPool.execute(() -> { // Lambda 表达式的运用
                    try { 
                        semaphore.acquire();// 获取一个许可，所以可运行线程数量为20/1=20
                        test(threadnum);
                        semaphore.release();// 释放一个许可
                    } catch (InterruptedException e) { 
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
    
                });
            }
            threadPool.shutdown();
            System.out.println("finish");
        }
    
        public static void test(int threadnum) throws InterruptedException { 
            Thread.sleep(1000);// 模拟请求的耗时操作
            System.out.println("threadnum:" + threadnum);
            Thread.sleep(1000);// 模拟请求的耗时操作
        }
    }

#### 2. acquire()和tryAcquire()有何不同？ ####

这两个方法都可以用于许可证，不同之处在于acquire在许可证不足的时候会阻塞，而tryAcquire会在许可证不足的时候直接返回false。

#### 3.Semaphore的公平模式和非公平模式了解吗？ ####

##### 1.公平模式 #####

根据acquire的顺序来获取许可证，遵循FIFO。

##### 2.非公平模式 #####

即抢占式获取许可证。这是默认情况。

可在构造信号量时通过构造函数去设置这两种模式：

// 默认是非公平的
        public Semaphore(int permits) { 
            sync = new NonfairSync(permits);
        }
    
        public Semaphore(int permits, boolean fair) { 
            sync = fair ? new FairSync(permits) : new NonfairSync(permits);
        }

### 三.倒计时器(闭锁)CountDownLatch ###

#### 1.什么是CountDownLatch ####

CountDownLatch位于java.util.concurrent下，它的作用是使得规定数量的线程在某个位置等待，在数量满足要求后，会在同一时刻并发执行。

#### 2.CountDownLatch有什么作用 ####

1.  可以使得某个线程等待其他线程全部执行完毕再执行
2.  可以用作并发模拟
3.  死锁检测

#### 3.一个计数器使用的例子：主线程等待其他所有线程执行完毕在执行 ####

public class CountDownLatchDemo { 
        private static int counter = 0;
    
        public static void main(String[] args) throws InterruptedException { 
            ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                    0, // corePoolSize
                    Integer.MAX_VALUE, // maxPoolSize
                    60L,
                    TimeUnit.SECONDS,
                    new SynchronousQueue<Runnable>()
            );
            int count = 5;
            CountDownLatch latch = new CountDownLatch(count);
            for (int i = 0; i < count; i++) { 
                threadPoolExecutor.execute(() -> { 
                    try { 
                        run();
                    } catch (Exception e) { 
                    }finally { 
                        latch.countDown();
                    }
                });
            }
            latch.await();
            threadPoolExecutor.shutdown();
            System.out.println("线程执行完毕");
        }
    
        private static void run() throws InterruptedException { 
            Thread.sleep(1000);
            System.out.println(Thread.currentThread());
    
        }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9hcm9uZy5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 1]

#### 4.一个并发测试的例子：将所有的子线程全部阻塞在一个入口，等待主线程统一放行 ####

public static void main(String[] args) throws InterruptedException { 
            new CountDownLatchDemo().test2();
        }
    
        private static int counter = 0;
        private static void run() throws InterruptedException { 
            System.out.println(Thread.currentThread() + ":" + (++counter));
        }
    
        private void test2() throws InterruptedException { 
            ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                    0, // corePoolSize
                    Integer.MAX_VALUE, // maxPoolSize
                    60L,
                    TimeUnit.SECONDS,
                    new SynchronousQueue<Runnable>()
            );
            int count = 50, i = 0;
            CountDownLatch latch = new CountDownLatch(1);
            for (i = 0; i <= count; i++) { 
                threadPoolExecutor.execute(() -> { 
                    try { 
                        run();
                    } catch (Exception e) { 
                    }finally { 
                        latch.countDown();
                    }
                });
            }
            // 所有线程启动完成，统一放行
            while (i == count) { 
                latch.countDown();
            }
            threadPoolExecutor.shutdown();
            System.out.println("线程执行完毕");
        }

可以看到执行结果中，counter的值出现了并发冲突：  
![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-RkixXxsE-1574526056647)(http://note.youdao.com/yws/res/3892/600DC263B4A74CB99AAE2DAC706F371F)\]][img-RkixXxsE-1574526056647_http_note.youdao.com_yws_res_3892_600DC263B4A74CB99AAE2DAC706F371F]

这时可以使用原子类AtomicInteger，通过CAS去自增计数器的值（为什么不使用volatile呢？因为volatile虽然能保证多线程读写共享变量的可见性，但修改一个变量并写入毕竟是分为读-改-写这三个过程的，所以仍需要使用CAS来保障）。

private static AtomicInteger counter = new AtomicInteger();
        private static void run() throws InterruptedException { 
            // 使用原子类 CAS更新值
            System.out.println(Thread.currentThread() + ":" + counter.incrementAndGet());
        }

#### 5.CountDownLatch的常用方法 ####

##### 1.await() vs wait() #####

await是CountDownLatch的方法，它的作用是在指定位置阻塞线程，直到闭锁中的计数为0；而wait()是Object类的方法。

##### 2.countDown() vs getCount() #####

countDown()方法用于为计数器减少计数，而getCount()可以获取到当前的计数值。

### 四.循环栅栏CyclicBarrier ###

https://blog.csdn.net/u010185262/article/details/54692886

#### 1.什么是循环栅栏？ ####

CyclicBarrier和CountDownLatch技术差不多，它们都是为了实现线程的等待以及同时执行，但是CyclicBarrier比CountDownLatch更加复杂，它们之间的不同点有：

1.  CyclicBarrier可以循环使用，当计数达到标准时即全部执行，而CountDownLatch只可使用一次。
2.  CyclicBarrier的实现中，计数器是随着等待的线程数量的增加二增加的，当加到一定值时即全部放行；而CountDownLatch中的计数器是通过countDown一个一个地去减少的，减少到0才全部放行。

#### 2，有什么作用 ####

将大任务划分为小任务完成，然后汇集成大任务，并且可以循环这个过程。

#### 3.具体使用 ####

private static CyclicBarrier cyclicBarrier = new CyclicBarrier(5);
    
        public static void main(String[] args) { 
            ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                    0, // corePoolSize
                    Integer.MAX_VALUE, // maxPoolSize
                    60L,
                    TimeUnit.SECONDS,
                    new SynchronousQueue<Runnable>()
            );
            int count = 50, i = 0;
            for (i = 0; i <= count; i++) { 
                threadPoolExecutor.execute(() -> { 
                    try { 
                        runTask();
                    } catch (BrokenBarrierException e) { 
                        e.printStackTrace();
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                });
            }
        }
        private static void runTask() throws BrokenBarrierException, InterruptedException { 
            System.out.println(Thread.currentThread().getName() + "-Ready");
            cyclicBarrier.await();
            System.out.println(Thread.currentThread().getName() + "-Finished");
        }

从打印结果中可以看到，当等待的任务达到5个时，这5个任务就会立刻开始执行，然后开始下一轮的等待：  
![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-fQCWDMka-1574526056647)(http://note.youdao.com/yws/res/3932/55D92A61A79B4BDEB12012D9EB052C48)\]][img-fQCWDMka-1574526056647_http_note.youdao.com_yws_res_3932_55D92A61A79B4BDEB12012D9EB052C48]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9hcm9uZy5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/2019112400212673.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9hcm9uZy5ibG9nLmNzZG4ubmV0,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9hcm9uZy5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20191124002211280.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9hcm9uZy5ibG9nLmNzZG4ubmV0,size_16,color_FFFFFF,t_70
[img-RkixXxsE-1574526056647_http_note.youdao.com_yws_res_3892_600DC263B4A74CB99AAE2DAC706F371F]: https://img-blog.csdnimg.cn/20191124002241849.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9hcm9uZy5ibG9nLmNzZG4ubmV0,size_16,color_FFFFFF,t_70
[img-fQCWDMka-1574526056647_http_note.youdao.com_yws_res_3932_55D92A61A79B4BDEB12012D9EB052C48]: https://img-blog.csdnimg.cn/20191124002305823.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9hcm9uZy5ibG9nLmNzZG4ubmV0,size_16,color_FFFFFF,t_70