[TOC] ## J.U.C - AQS AQS 是 AbstractQueuedSynchronizer 的简称，java.util.concurrent（J.U.C）大大提高了并发性能，AQS (AbstractQueuedSynchronizer) 被认为是 J.U.C 的核心。它提供了一个基于 FIFO 队列，这个队列可以用来构建锁或者其他相关的同步装置的基础框架。下图是 AQS 底层的数据结构： [](https://github.com/frank-lam/fullstack-tutorial/blob/master/notes/JavaArchitecture/assets/616953-20160403170136176-573839888.png) 它底层使用的是双向列表，是队列的一种实现 , 因此也可以将它当成一种队列。 * Sync queue 是同步列表，它是双向列表 , 包括 head，tail 节点。其中 head 节点主要用来后续的调度 ; * Condition queue 是单向链表 , 不是必须的 , 只有当程序中需要 Condition 的时候，才会存在这个单向链表 , 并且可能会有多个 Condition queue。 简单的来说： * AQS其实就是一个可以给我们实现锁的**框架** * 内部实现的关键是：**先进先出的队列、state 状态** * 定义了内部类 ConditionObject * 拥有两种线程模式 * * 独占模式 * 共享模式 * 在 LOCK 包中的相关锁（常用的有 ReentrantLock、 ReadWriteLock ）都是基于 AQS 来构建 * 一般我们叫 AQS 为同步器。 ### CountdownLatch CountDownLatch 类位于 java.util.concurrent 包下，利用它可以实现类似计数器的功能。比如有一个任务 A，它要等待其他 4 个任务执行完毕之后才能执行，此时就可以利用 CountDownLatch 来实现这种功能了。 维护了一个计数器 cnt，每次调用 countDown() 方法会让计数器的值减 1，减到 0 的时候，那些因为调用 await() 方法而在等待的线程就会被唤醒。 [](https://github.com/frank-lam/fullstack-tutorial/blob/master/notes/JavaArchitecture/assets/CountdownLatch.png) CountDownLatch 类只提供了一个构造器： ~~~java public CountDownLatch(int count) { }; // 参数count为计数值 ~~~ 然后下面这 3 个方法是 CountDownLatch 类中最重要的方法： ~~~java //调用await()方法的线程会被挂起，它会等待直到count值为0才继续执行 public void await() throws InterruptedException { }; //和await()类似，只不过等待一定的时间后count值还没变为0的话就会继续执行 public boolean await(long timeout, TimeUnit unit) throws InterruptedException { }; //将count值减1 public void countDown() { }; ~~~ 下面看一个例子大家就清楚 CountDownLatch 的用法了： ~~~java public class Test { public static void main(String[] args) { final CountDownLatch latch = new CountDownLatch(2); new Thread() { public void run() { try { System.out.println("子线程" + Thread.currentThread().getName() + "正在执行"); Thread.sleep(3000); System.out.println("子线程" + Thread.currentThread().getName() + "执行完毕"); latch.countDown(); } catch (InterruptedException e) { e.printStackTrace(); } } ; }.start(); new Thread() { public void run() { try { System.out.println("子线程" + Thread.currentThread().getName() + "正在执行"); Thread.sleep(3000); System.out.println("子线程" + Thread.currentThread().getName() + "执行完毕"); latch.countDown(); } catch (InterruptedException e) { e.printStackTrace(); } } ; }.start(); try { System.out.println("等待2个子线程执行完毕..."); latch.await(); System.out.println("2个子线程已经执行完毕"); System.out.println("继续执行主线程"); } catch (InterruptedException e) { e.printStackTrace(); } } } ~~~ 执行结果： ~~~ 线程Thread-0正在执行 线程Thread-1正在执行 等待2个子线程执行完毕... 线程Thread-0执行完毕 线程Thread-1执行完毕 2个子线程已经执行完毕 继续执行主线程 ~~~ ### CyclicBarrier 用来控制多个线程互相等待，只有当多个线程都到达时，这些线程才会继续执行。 和 CountdownLatch 相似，都是通过维护计数器来实现的。但是它的计数器是递增的，每次执行 await() 方法之后，计数器会加 1，直到计数器的值和设置的值相等，等待的所有线程才会继续执行。和 CountdownLatch 的另一个区别是，CyclicBarrier 的计数器可以循环使用，所以它才叫做循环屏障。 下图应该从下往上看才正确。 [](https://github.com/frank-lam/fullstack-tutorial/blob/master/notes/JavaArchitecture/assets/CyclicBarrier.png) ~~~java public class CyclicBarrierExample { public static void main(String[] args) throws InterruptedException { final int totalThread = 10; CyclicBarrier cyclicBarrier = new CyclicBarrier(totalThread); ExecutorService executorService = Executors.newCachedThreadPool(); for (int i = 0; i < totalThread; i++) { executorService.execute(() -> { System.out.print("before.."); try { cyclicBarrier.await(); } catch (InterruptedException e) { e.printStackTrace(); } catch (BrokenBarrierException e) { e.printStackTrace(); } System.out.print("after.."); }); } executorService.shutdown(); } } ~~~ ~~~ before..before..before..before..before..before..before..before..before..before..after..after..after..after..after..after..after..after..after..after.. ~~~ ### Semaphore Semaphore 就是操作系统中的信号量，可以控制对互斥资源的访问线程数。Semaphore 可以控同时访问的线程个数，通过 acquire() 获取一个许可，如果没有就等待，而 release() 释放一个许可。 [](https://github.com/frank-lam/fullstack-tutorial/blob/master/notes/JavaArchitecture/assets/Semaphore.png) Semaphore 类位于 java.util.concurrent 包下，它提供了2个构造器： ~~~java public Semaphore(int permits) { //参数permits表示许可数目，即同时可以允许多少线程进行访问 sync = new NonfairSync(permits); } public Semaphore(int permits, boolean fair) { //这个多了一个参数fair表示是否是公平的，即等待时间越久的越先获取许可 sync = (fair) ? new FairSync(permits) : new NonfairSync(permits); } ~~~ 下面说一下 Semaphore 类中比较重要的几个方法，首先是 acquire()、release() 方法： ~~~java //获取一个许可 public void acquire() throws InterruptedException { } //获取permits个许可 public void acquire(int permits) throws InterruptedException { } //释放一个许可 public void release() { } //释放permits个许可 public void release(int permits) { } ~~~ 　　acquire() 用来获取一个许可，若无许可能够获得，则会一直等待，直到获得许可。 　　release() 用来释放许可。注意，在释放许可之前，必须先获获得许可。 这 4 个方法都会被阻塞，如果想立即得到执行结果，可以使用下面几个方法： ~~~java //尝试获取一个许可，若获取成功，则立即返回true，若获取失败，则立即返回false public boolean tryAcquire() { }; //尝试获取一个许可，若在指定的时间内获取成功，则立即返回true，否则则立即返回false public boolean tryAcquire(long timeout, TimeUnit unit) throws InterruptedException { }; //尝试获取permits个许可，若获取成功，则立即返回true，若获取失败，则立即返回false public boolean tryAcquire(int permits) { }; //尝试获取permits个许可，若在指定的时间内获取成功，则立即返回true，否则则立即返回false public boolean tryAcquire(int permits, long timeout, TimeUnit unit) throws InterruptedException { }; ~~~ 　　另外还可以通过 availablePermits() 方法得到可用的许可数目。 　　下面通过一个例子来看一下 Semaphore 的具体使用： 　　假若一个工厂有 5 台机器，但是有 8 个工人，一台机器同时只能被一个工人使用，只有使用完了，其他工人才能继续使用。那么我们就可以通过 Semaphore 来实现： ~~~java public class Test { public static void main(String[] args) { int N = 8; //工人数 Semaphore semaphore = new Semaphore(5); //机器数目 for (int i = 0; i < N; i++) new Worker(i, semaphore).start(); } static class Worker extends Thread { private int num; private Semaphore semaphore; public Worker(int num, Semaphore semaphore) { this.num = num; this.semaphore = semaphore; } @Override public void run() { try { semaphore.acquire(); System.out.println("工人" + this.num + "占用一个机器在生产..."); Thread.sleep(2000); System.out.println("工人" + this.num + "释放出机器"); semaphore.release(); } catch (InterruptedException e) { e.printStackTrace(); } } } } ~~~ 执行结果： ~~~ 工人0占用一个机器在生产... 工人1占用一个机器在生产... 工人2占用一个机器在生产... 工人4占用一个机器在生产... 工人5占用一个机器在生产... 工人0释放出机器 工人2释放出机器 工人3占用一个机器在生产... 工人7占用一个机器在生产... 工人4释放出机器 工人5释放出机器 工人1释放出机器 工人6占用一个机器在生产... 工人3释放出机器 工人7释放出机器 工人6释放出机器 ~~~ ### 总结 下面对上面说的三个辅助类进行一个总结： * CountDownLatch 和 CyclicBarrier 都能够实现线程之间的等待，只不过它们侧重点不同： * CountDownLatch 一般用于某个线程A等待若干个其他线程执行完任务之后，它才执行； * CyclicBarrier 一般用于一组线程互相等待至某个状态，然后这一组线程再同时执行； * 另外，CountDownLatch 是不能够重用的，而 CyclicBarrier 是可以重用的。 * Semaphore 其实和锁有点类似，它一般用于控制对某组资源的访问权限。 ## J.U.C - 其它组件 ### FutureTask 在介绍 Callable 时我们知道它可以有返回值，返回值通过 Future 进行封装。FutureTask 实现了 RunnableFuture 接口，该接口继承自 Runnable 和 Future 接口，这使得 FutureTask 既可以当做一个任务执行，也可以有返回值。 ~~~java public class FutureTask<V> implements RunnableFuture<V> ~~~ ~~~java public interface RunnableFuture<V> extends Runnable, Future<V> ~~~ FutureTask 可用于异步获取执行结果或取消执行任务的场景。当一个计算任务需要执行很长时间，那么就可以用 FutureTask 来封装这个任务，主线程在完成自己的任务之后再去获取结果。 ~~~java public class FutureTaskExample { public static void main(String[] args) throws ExecutionException, InterruptedException { FutureTask<Integer> futureTask = new FutureTask<Integer>(new Callable<Integer>() { @Override public Integer call() throws Exception { int result = 0; for (int i = 0; i < 100; i++) { Thread.sleep(10); result += i; } return result; } }); Thread computeThread = new Thread(futureTask); computeThread.start(); Thread otherThread = new Thread(() -> { System.out.println("other task is running..."); try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } }); otherThread.start(); System.out.println(futureTask.get()); } } ~~~ ~~~java other task is running... 4950 ~~~ ### BlockingQueue java.util.concurrent.BlockingQueue 接口有以下阻塞队列的实现： * **FIFO 队列**：LinkedBlockingQueue、ArrayBlockingQueue（固定长度） * **优先级队列**：PriorityBlockingQueue 提供了阻塞的 take() 和 put() 方法：如果队列为空 take() 将阻塞，直到队列中有内容；如果队列为满 put() 将阻塞，直到队列有空闲位置。 **使用 BlockingQueue 实现生产者消费者问题** ~~~java public class ProducerConsumer { private static BlockingQueue<String> queue = new ArrayBlockingQueue<>(5); private static class Producer extends Thread { @Override public void run() { try { queue.put("product"); } catch (InterruptedException e) { e.printStackTrace(); } System.out.print("produce.."); } } private static class Consumer extends Thread { @Override public void run() { try { String product = queue.take(); } catch (InterruptedException e) { e.printStackTrace(); } System.out.print("consume.."); } } } ~~~ ~~~java public static void main(String[] args) { for (int i = 0; i < 2; i++) { Producer producer = new Producer(); producer.start(); } for (int i = 0; i < 5; i++) { Consumer consumer = new Consumer(); consumer.start(); } for (int i = 0; i < 3; i++) { Producer producer = new Producer(); producer.start(); } } ~~~ ~~~ produce..produce..consume..consume..produce..consume..produce..consume..produce..consume.. ~~~ ### ForkJoin 主要用于并行计算中，和 MapReduce 原理类似，都是把大的计算任务拆分成多个小任务并行计算。 ~~~java public class ForkJoinExample extends RecursiveTask<Integer> { private final int threshold = 5; private int first; private int last; public ForkJoinExample(int first, int last) { this.first = first; this.last = last; } @Override protected Integer compute() { int result = 0; if (last - first <= threshold) { // 任务足够小则直接计算 for (int i = first; i <= last; i++) { result += i; } } else { // 拆分成小任务 int middle = first + (last - first) / 2; ForkJoinExample leftTask = new ForkJoinExample(first, middle); ForkJoinExample rightTask = new ForkJoinExample(middle + 1, last); leftTask.fork(); rightTask.fork(); result = leftTask.join() + rightTask.join(); } return result; } } ~~~ ~~~js public static void main(String[] args) throws ExecutionException, InterruptedException { ForkJoinExample example = new ForkJoinExample(1, 10000); ForkJoinPool forkJoinPool = new ForkJoinPool(); Future result = forkJoinPool.submit(example); System.out.println(result.get()); } ~~~ ForkJoin 使用 ForkJoinPool 来启动，它是一个特殊的线程池，线程数量取决于 CPU 核数。 ~~~java public class ForkJoinPool extends AbstractExecutorService ~~~ ForkJoinPool 实现了工作窃取算法来提高 CPU 的利用率。每个线程都维护了一个双端队列，用来存储需要执行的任务。工作窃取算法允许空闲的线程从其它线程的双端队列中窃取一个任务来执行。窃取的任务必须是最晚的任务，避免和队列所属线程发生竞争。例如下图中，Thread2 从 Thread1 的队列中拿出最晚的 Task1 任务，Thread1 会拿出 Task2 来执行，这样就避免发生竞争。但是如果队列中只有一个任务时还是会发生竞争。 [](https://github.com/frank-lam/fullstack-tutorial/blob/master/notes/JavaArchitecture/assets/fork-and-join.jpg)