java并发容器J.U.C AQS怎么用

java并发容器J.U.C AQS怎么用，很多新手对此不是很清楚，为了帮助大家解决这个难题，下面小编将为大家详细讲解，有这方面需求的人可以来学习下，希望你能有所收获。

AbstractQueueSynchronizer (AQS)

J.U.C 大大提高了java并发的性能，而AQS则是J.U.C的核心。

AQS底层使用双向列表（队列的一种实现）。

• 使用Node实现FIFO队列，可以用于构建锁或者其他同步装置的基础框架

• 利用了一个int类型表示状态。 在AQS中有一个status的成员变量，基于AQS有一个同步组件ReentrantLock，在这个ReentrantLock中status表示获取锁的线程数，例如status=0表示还没有线程获取锁，status=1表示已经有线程获取了锁，status>1表示重入锁的数量。

• 使用方法：继承

• 子类通过继承并通过实现它的方法管理其状态{acquire和release}的方法操纵状态

• 可以同时实现排它锁和共享锁模式（独占，共享）

AQS同步组件

• countdownLatch，闭锁，通过一个计数来保证线程是否需要一直阻塞。

• semaphore，控制同一时间并发线程的数量。

• cyclicbarrier，与countdownlatch很像，都能阻塞进程。

• reentrantlock

• condition

• futuretask

countdownlatch 

是一个同步辅助类，通过他可以实现类似于阻塞当前线程的功能。一个线程或多个线程一直等待，直到其他线程操作完成，countdownlatch用了一个给定的计数器来进行初始化，该计数器的操作是原子操作，也就是同时只能有一个线程操作该计数器。调用该类的await()方法则会一直处于阻塞状态，直到其他线程调用countdown()方法，每次调用countdown()方法会使得计数器的值减1，当计数器的值减为0时，所有因调用await方法处于等待状态的线程就会继续往下执行。这种状态只会出现一次，因为这里的计数器是不能被重置的，如果业务上需要一个可以重置计数次数的版本，可以考虑使用cyclicbarrier。

使用场景

在某些业务场景中，程序执行需要等到某个条件完成后才能继续执行后续的操作，典型的应用例如并行计算：当某个处理的运算量很大时，可以将该运算任务拆分多个子任务，等待所有的子任务都完成之后，父任务拿到所有的子任务的运行结果进行汇总。

下面举例countdownlatch的基本用法：

@Slf4j
public class CountDownLatchExample1 {

    private final static int threadCount = 200;

    public static void main(String[] args) throws InterruptedException {
        ExecutorService executorService = Executors.newCachedThreadPool();
        final CountDownLatch countDownLatch = new CountDownLatch(threadCount);
        for(int i = 0; i< threadCount; i++) {
            final int threadNum = i;
            executorService.execute(() ->{
                try {
                    test(threadNum);
                } catch (InterruptedException e) {
                    log.error("exception", e);
                }finally {
                    countDownLatch.countDown();
                }
            });
        }
        //可以保证之前的线程都执行完成
        countDownLatch.await();
        log.info("finish");
        executorService.shutdown();
    }
    private static void test(int threadNum) throws InterruptedException {
        Thread.sleep(100);
        log.info("{}", threadNum);
        Thread.sleep(100);
    }
}

一个复杂的场景：我们开了很多个线程去完成一个任务，但是这个任务需要在指定的时间内完成，如果超过一定的时间没有完成则放弃该任务。

@Slf4j
public class CountDownLatchExample2 {

    private final static int threadCount = 200;

    public static void main(String[] args) throws InterruptedException {
        ExecutorService executorService = Executors.newCachedThreadPool();
        final CountDownLatch countDownLatch = new CountDownLatch(threadCount);
        for(int i = 0; i< threadCount; i++) {
            final int threadNum = i;
            executorService.execute(() ->{
                try {
                    test(threadNum);
                } catch (InterruptedException e) {
                    log.error("exception", e);
                }finally {
                    countDownLatch.countDown();
                }
            });
        }
        //可以保证之前的线程都执行完成
        countDownLatch.await(10, TimeUnit.MILLISECONDS);
        log.info("finish");
        // 第一时间内并不会把所有线程都销毁，而是让当前已有线程执行完之后在把线程池销毁。
        executorService.shutdown();
    }
    private static void test(int threadNum) throws InterruptedException {
        Thread.sleep(100);
        log.info("{}", threadNum);
    }
}

semaphore 信号量

可以控制某个资源可被同时访问的个数，与countdownlatch有些类似，提供了两个核心方法：aquire和release。aquire表示获取一个许可，如果没有则等待，release表示操作完成后释放一个许可。semaphore维护了当前访问的个数，提供同步机制控制访问的个数。

使用场景 

常用于仅能提供有限访问的资源例如数据库连接数是有限的，而上层应用的并发数会远远大于连接数，如果同时对数据库进行操作可能出现因为无法获取数据库连接而导致异常。这时可以通过信号量semaphore来并发访问控制。当semaphore把并发数控制到1时就跟单线程运行很相似了。

举例如下：

@Slf4j
public class SemaphoreExample1 {

    private final static int threadCount = 20;

    public static void main(String[] args) throws InterruptedException {
        ExecutorService executorService = Executors.newCachedThreadPool();
        //允许的并发数
        final Semaphore semaphore = new Semaphore(3);
        for(int i = 0; i< threadCount; i++) {
            final int threadNum = i;
            executorService.execute(() ->{
                try {
                    // 获取一个许可
                    semaphore.acquire();
                    test(threadNum);
                    // 释放一个许可
                    semaphore.release();
                } catch (InterruptedException e) {
                    log.error("exception", e);
                }
            });
        }
        log.info("finish");
        executorService.shutdown();
    }
    private static void test(int threadNum) throws InterruptedException {
        log.info("{}", threadNum);
        Thread.sleep(1000);
    }
}

运行结果可以看到同时3个线程在执行。

也可以获得多个许可：

@Slf4j
public class SemaphoreExample2 {

    private final static int threadCount = 20;

    public static void main(String[] args) throws InterruptedException {
        ExecutorService executorService = Executors.newCachedThreadPool();
        //允许的并发数
        final Semaphore semaphore = new Semaphore(3);
        for(int i = 0; i< threadCount; i++) {
            final int threadNum = i;
            executorService.execute(() ->{
                try {
                    // 获取多个许可
                    semaphore.acquire(3);
                    test(threadNum);
                    // 释放多个许可
                    semaphore.release(3);
                } catch (InterruptedException e) {
                    log.error("exception", e);
                }
            });
        }
        log.info("finish");
        executorService.shutdown();
    }
    private static void test(int threadNum) throws InterruptedException {
        log.info("{}", threadNum);
        Thread.sleep(1000);
    }
}

每一次获取三个许可，而同时只允许3个并发数，相当于单线程在运行。

@Slf4j
public class SemaphoreExample3 {

    private final static int threadCount = 20;

    public static void main(String[] args) throws InterruptedException {
        ExecutorService executorService = Executors.newCachedThreadPool();
        //允许的并发数
        final Semaphore semaphore = new Semaphore(3);
        for (int i = 0; i < threadCount; i++) {
            final int threadNum = i;
            executorService.execute(() -> {
                try {
                    // 尝试获取一个许可
                    if (semaphore.tryAcquire()) {
                        test(threadNum);
                        // 释放一个许可
                        semaphore.release();
                    }
                } catch (InterruptedException e) {
                    log.error("exception", e);
                }
            });
        }
        log.info("finish");
        executorService.shutdown();
    }

    private static void test(int threadNum) throws InterruptedException {
        log.info("{}", threadNum);
        Thread.sleep(1000);
    }
}

输出结果：

15:24:21.098 [pool-1-thread-1] INFO com.vincent.example.aqs.SemaphoreExample3 - 0
15:24:21.098 [pool-1-thread-2] INFO com.vincent.example.aqs.SemaphoreExample3 - 1
15:24:21.098 [main] INFO com.vincent.example.aqs.SemaphoreExample3 - finish
15:24:21.098 [pool-1-thread-3] INFO com.vincent.example.aqs.SemaphoreExample3 - 2

因为我们往线程池中放了二十个请求，二十个请求在同一时间内都会尝试去执行，semaphore会尝试让每个线程去获取许可，而同一时刻内我们的并发数是3，也就是只有三个线程获取到了许可，而test方法内有Thread.sleep(1000)，因此其余17个线程都不能拿到许可，直接结束。

semaphore.tryAcquire(3, TimeUnit.SECONDS)

表示可以等3秒，如果3秒内没拿到许可就结束。

CyclicBarrier

也是一个同步辅助类，允许一组线程相互等待，直到到达某个公共的屏障点。可以完成多个线程之间相互等待，只有当每个线程都准备就绪后，才能各自继续往下执行谋面的操作。它和countdownlatch有相似的地方，都是通过计数器来实现的，当一个线程调用await()方法后，该线程就进入了等待状态。当循环计数器的值达到设置的初始值之后，进入等待状态的线程会被唤醒，继续执行后续操作。因为CyclicBarrier在释放等待线程后可以重用，所以称他为循环屏障。

CyclicBarrier的使用场景与countdownlatch类似，CyclicBarrier可以用于多线程计算数据，最后合并计算结果的应用场景。

CyclicBarrier与Countdownlatch的区别：

• countdownlatch的计数器只能使用一次，CyclicBarrier可以使用reset方法重复使用

• countdownlatch主要是实现一个或n个线程需要等待其他线程完成某项操作之后，才能继续往下执行，他描述的是1个或n个线程等待其他线程的关系。而CyclicBarrier主要实现了多个线程之间相互等待，直到所有线程都满足了条件之后才能继续执行后续的操作，它描述的是各个线程内部相互等待的关系。所以CyclicBarrier可以处理更复杂的业务场景，例如计数器发生错误可以重置计数器，让线程重新执行一次。

看完上述内容是否对您有帮助呢？如果还想对相关知识有进一步的了解或阅读更多相关文章，请关注亿速云行业资讯频道，感谢您对亿速云的支持。