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，有一个成员身份变量。基于AQS，有一个同步组件可重入锁。在这个重入锁中，状态指示获取锁的线程数量。例如，status=0表示还没有线程获得锁，status=1表示线程已经获得锁，status1表示重新锁的数量。

用法：继承。

通过实现状态的方法继承和管理状态{acquire和release}来操纵状态。

可以同时实现排他锁和共享锁模式(排他和共享)。

AQS同步组件

countdowlatch，锁定，通过一个计数来保证线程是否需要一直被阻塞。

信号量，一次控制并发线程的数量。

Cyclicbarrier和countdownlatch一样，可以阻塞进程。

可重入锁

情况

未来任务

00-1010是一个同步辅助类，通过它可以实现类似阻塞当前线程的功能。一个线程或多个线程等待，直到其他线程的操作完成。countdownlatch用给定的计数器初始化，这是一个原子操作，也就是说，只有一个线程可以同时操作计数器。调用此类的await()方法将一直被阻止，直到其他线程调用倒计时()方法。每次调用倒计时()方法都会将计数器值减少1。当计数器值减少到0时，所有等待调用await方法的线程将继续执行。这种状态只会出现一次，因为这里的计数器不能复位。如果企业需要一个可以重置计数时间的版本，可以考虑使用cyclicbarrier。

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

下面是countdownlatch的基本用法示例：

@Slf4j

public classcounttlotchexample 1 {

privatefinitiationthreadcount=200；

publicationstatinvitmain(String[]args)throwsInterruptedException {

ExecutorServiceexecutorService=executors . new cachedthreadpool()；

financicountdowlatchcountdowlatch=new countdowlatch(thread count)；

for(inti=0；ithreadCountI){ 0

最终

readNum = 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可以处理更复杂的业务场景，例如计数器发生错误可以重置计数器，让线程重新执行一次。

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