对于Java的面试，突然想到并发的内容不全。本文将通篇大致讲解多线程和并发，至少达到了解原理和使用的目的。内容会更多，从最基础的线程到我们常用的类，我们统一再说一遍。慢慢来。
并发编程
每个人都应该熟悉线程的基本概念，进程是资源分配的单位，线程是CPU调度的单位.线程是进程中的一个实体。
我们的Java程序天生就是多线程的。当我们通过main方法启动它时，我们启动一个JVM进程并创建一个名为main的线程，它是JVM进程中的一个实体线程。
过程线程
线程生命周期的几种基本状态：
New，初始状态，即已经创建了一个新线程，但是还没有调用start方法Runnable，Ready或Running state。线程的就绪和运行状态统称为可运行运行状态blocked/watching/timed _ watching，这些状态的统一称为休眠状态终止。我们将分别讨论终止状态的几种状态之间的转换。
New:我们创建了一个线程，但是这个线程没有调用start方法，也就是说，它被初始化了。
Runnable:调用start()启动线程进入就绪可运行状态，等待CPU调度后再进入运行状态。
Blocked:阻塞状态，当一个线程等待进入同步锁时，它就进入阻塞状态。
Waiting:等待状态需要被显示器唤醒。进入这种状态有三种情况：调用Object.wait()，调用Thread.join()和在synchonized中调用LockSupport.park()。
Timed_Waiting:等待和等待的区别在于有更多的暂停时间，所以没有必要显示唤醒。到了超时时间会自动唤醒，调用图中一些带有超时参数的方法会进入这种状态。
Terminated:终止状态，线程执行完毕。
Java中的线程分为守护线程和用户线程，上面提到的主线程其实就是一个用户线程。
两者的主要区别在于，只要有非守护线程没有结束，JVM就不会正常退出，而守护线程不会影响JVM的退出。
您可以用一种简单的方法将一个线程设置为守护进程。
Thread t=新线程()；t . set daemon(true)；锁是控制多线程并发访问共享资源的一种方式。为了更简单快捷地理解Java中的锁，我们可以对显式锁和隐式锁做一个大致的区分。
锁
在没有锁接口之前，隐式锁是通过同步锁实现的。这个我在之前的Java基础系列中已经说过了，这里就不做过多阐述了。更多详情请参考《我想进大厂》的Java基础杀链第16题。
Synchronized是java提供的原子内置锁。这种内置且不可见的锁在使用synchronized后也被称为监视器锁.它会在编译后的synchronized代码块前后添加monitorenter和monitorexit字节码指令。它依靠操作系统的底层互斥锁实现原子操作，解决共享变量的内存可见性问题。
当monitorenter命令被执行时，它将尝试获取对象的锁。如果对象未被锁定或锁已被获取，则锁的计数器为1。此时，竞争锁的其他线程将进入等待队列。
当执行monitorexit指令时，计数器将被设置为-1。当计数器值为0时，锁将被释放，等待队列中的线程将继续竞争锁。
如果深入到源代码中，synchronized实际上有两个队列waitSet和entryList。
当多个线程进入同步代码块时，它们首先进入entryList。在一个线程获得监视器锁之后，它被分配给当前线程，如果该线程调用wait方法，计数器1将被释放。
锁，当前线程置为null，计数器-1，同时进入waitSet等待被唤醒，调用notify或者notifyAll之后又会进入entryList竞争锁如果线程执行完毕，同样释放锁，计数器-1，当前线程置为null
显示锁虽然synchronized使用简单，但是也使得加锁的流程固化了，显示锁在Java1.5版本之后加入了Lock接口，可以通过声明式显示的加锁和解锁。
Lock lock = new ReentrantLock();lock.lock(); //加锁lock.unlock(); //解锁独占锁在上述的伪代码中，我们使用到了ReentrantLock，它其实就是独占锁，独占锁保证任何时候都只有一个线程能获得锁，当然了，synchronized也是独占锁。
这里我们看ReentrantLock的几个加锁接口。
void lock(); //阻塞加锁void lockInterruptibly() throws InterruptedException; //可中断boolean tryLock(); //非阻塞boolean tryLock(long time, TimeUnit unit) throws InterruptedException; //超时加锁这几个加锁接口，向我们明白地展示了他和synchronized的区别。
可中断加锁lockInterruptibly，synchronized可能会有死锁的问题，那么解决方案就是能响应中断。当前线程加锁时，如果其他线程调用当前线程的中断方法，则会抛出异常。非阻塞加锁tryLock，调用后立刻返回，获取锁则返回true，否则返回false支持超时加锁tryLock(long time, TimeUnit unit)，超时时间内获取锁返回true，否则返回false支持公平和非公平锁，公平指的是获取锁按照请求锁的时间顺序决定，先到先得，非公平则是直接竞争锁，先到不一定先得支持Condition如果你看过阻塞队列的源码，那么你对 Condition 应该挺了解了，我们举个栗子来看看，我们需要实现：
如果队列满了，那么写入阻塞如果队列空了，那么删除（取元素）阻塞我们给阻塞队列提供一个 put 写入元素和 take 删除元素的方法。
put 时候加锁且响应中断，如果队列满了，notFull.await 释放锁，进入阻塞状态，反之，则把元素添加到队列中，notEmpty.signal 唤醒阻塞在删除元素的线程。
take 的时候一样加锁且响应中断，如果队列空了，notEmpty.await 进入释放锁，进入阻塞状态，反之，则删除元素，notFull.signal 唤醒阻塞在添加元素的线程。
public class ConditionTest { public static void main(String<> args) throws Exception { ArrayBlockingQueue arrayBlockingQueue = new ArrayBlockingQueue(10); } static class ArrayBlockingQueue<E> { private Object<> items; int takeIndex; int putIndex; int count; private ReentrantLock lock; private Condition notEmpty; private Condition notFull; public ArrayBlockingQueue(int capacity) { this.items = new Object
ReentrantReadWriteLock
通过阅读源码发现它内部维护了两个锁：读锁和写锁。
private final ReentrantReadWriteLock.ReadLock readerLock;private final ReentrantReadWriteLock.WriteLock writerLock;本质上，不管是ReentrantLock还是ReentrantReadWriteLock都是基于AQS，AQS只有一个状态位state，对于ReentrantReadWriteLock实现读锁和写锁则是对state做出了区分，高16位表示的是读锁的状态，低16表示的是写锁的状态。
我们可以看一个源码中给出的使用例子。
class CacheData { Object data; volatile boolean cacheValid; final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock(); void processCachedData() { rwl.readLock().lock(); if (!cacheValid) { // 必须先释放读锁，再加写锁 rwl.readLock().unlock(); rwl.writeLock().lock(); try { // 重新校验状态，防止并发问题 if (!cacheValid) { data = ... cacheValid = true; } // 写锁降级为读锁 rwl.readLock().lock(); } finally { rwl.writeLock().unlock(); // 写锁释放，仍然持有读锁 } } try { use(data); } finally { rwl.readLock().unlock(); } }}这个例子嵌套写的其实不太好理解，因为他包含了一个写锁降级的概念，实际上我们自己写最简单的例子就是这样，例子中给到的示例其实是一个意思，只是在写锁释放前先降级为读锁，明白意思就好。
rwl.readLock().lock();doSomething();rwl.readLock().unlock();rwl.writeLock().lock();doSomething();rwl.writeLock().unlock();额外需要注意的是，写锁可以降级为读锁，但是读锁不能升级为写锁，比如下面这种写法是不支持的。
rwl.readLock().lock();doSomething();rwl.writeLock().lock();doSomething();rwl.writeLock().unlock();rwl.readLock().unlock();StampedLock
这是JDK1.8之后新增的一个锁，相比ReentrantReadWriteLock他的性能更好，在读锁和写锁的基础上增加了一个乐观读锁。
写锁：他的写锁基本上和ReentrantReadWriteLock一样，但是不可重入。
读锁：也和ReentrantReadWriteLock一样，但是不可重入。
乐观读锁：普通的读锁通过CAS去修改当前state状态，乐观锁实现原理则是加锁的时候返回一个stamp（锁状态），然后还需要调用一次validate(stamp)判断当前是否有其他线程持有了写锁，通过的话则可以直接操作数据，反之升级到普通的读锁，之前我们说到读写锁也是互斥的，那么乐观读和写就不是这样的了，他能支持一个线程去写。所以，他性能更高的原因就来自于没有CAS的操作，只是简单的位运算拿到当前的锁状态stamp，并且能支持另外的一个线程去写。
总结下来可以理解为：读读不互斥，读写不互斥，写写互斥，另外通过tryConvertToReadLock()和tryConvertToWriteLock()等方法支持锁的升降级。
还是按照官方的文档举个栗子，方便理解，两个方法分别表示乐观锁的使用和锁升级的使用。
public class StampedLockTest { private double x, y; private final StampedLock sl = new StampedLock(); double distanceFromOrigin() { // 乐观锁 long stamp = sl.tryOptimisticRead(); double currentX = x, currentY = y; if (!sl.validate(stamp)) { //状态已经改变，升级到读锁，重新读取一次最新的数据 stamp = sl.readLock(); try { currentX = x; currentY = y; } finally { sl.unlockRead(stamp); } } return Math.sqrt(currentX * currentX + currentY * currentY); } void moveIfAtOrigin(double newX, double newY) { // 可以使用乐观锁替代 long stamp = sl.readLock(); try { while (x == 0.0 && y == 0.0) { // 尝试升级到写锁 long ws = sl.tryConvertToWriteLock(stamp); if (ws != 0L) { //升级成功，替换当前stamp标记 stamp = ws; x = newX; y = newY; break; } else { //升级失败，再次获取写锁 sl.unlockRead(stamp); stamp = sl.writeLock(); } } } finally { sl.unlock(stamp); } }}LockSupportLockSupport是一个比较基础的工具类，基于Unsafe实现，主要就是提供线程阻塞和唤醒的能力，上面我们提到对线程生命周期状态的时候也说过了，LockSupport的几个park功能将会把线程阻塞，直到被唤醒。
看看他的几个核心方法：
public static void park(); //阻塞当前线程public static void parkNanos(long nanos); //阻塞当前线程加上了超时时间，达到超时时间之后返回public static void parkUntil(long deadline); //和上面类似，参数deadline代表的是从1970到现在时间的毫秒数public static void unpark(Thread thread);// 唤醒线程举个栗子：
public class Test { public static void main(String<> args) throws Exception { int sleepTime = 3000; Thread t = new Thread(() -> { System.out.println(Thread.currentThread().getName() + "挂起"); LockSupport.park(); System.out.println(Thread.currentThread().getName() + "继续工作"); }); t.start(); System.out.println("主线程sleep" + sleepTime); Thread.sleep(sleepTime); System.out.println("主线程唤醒阻塞线程"); LockSupport.unpark(t); }}//输出如下主线程sleep3000Thread-0挂起主线程唤醒阻塞线程Thread-0继续工作原子类多线程环境下操作变量，除了可以用我们上面一直说的加锁的方式，还有其他更简单快捷的办法吗？
JDK1.5之后引入的原子操作包下面的一些类提供给了我们一种无锁操作变量的方式，这种通过CAS操作的方式更高效并且线程安全。
原子类
基本数据类型我们先说针对基本数据类型提供的AtomicInteger、AtomicLong、AtomicBoolean，看名字都知道是干嘛的，由于基本上没什么区别，以AtomicInteger的方法举例来说明。
public final int getAndIncrement(); //旧值+1，返回旧值public final int getAndDecrement(); //旧值-1，返回旧值public final int getAndAdd(int delta); //旧值+delta，返回旧值public final int getAndSet(int newValue); //旧值设置为newValue，返回旧值public final int getAndAccumulate(int x,IntBinaryOperator accumulatorFunction); //旧值根据传入方法进行计算，返回旧值public final int getAndUpdate(IntUnaryOperator updateFunction)//旧值根据传入进行计算，返回旧值与之相对应的还有一套方法比如incrementAndGet()等等，规则完全一样，只是返回的是新值。
我们看看下面的例子，针对自定义规则传参，比如我们可以把计算规则改成乘法。
public class AtomicIntegerTest { public static void main(String<> args) { AtomicInteger atomic = new AtomicInteger(10); System.out.println(atomic.getAndIncrement()); //10 System.out.println(atomic.getAndDecrement()); //11 System.out.println(atomic.getAndAdd(2));//10 System.out.println(atomic.getAndSet(10)); //12 System.out.println(atomic.get()); //10 System.out.println("====================="); System.out.println(atomic.getAndAccumulate(3, (left, right) -> left * right)); // 10 System.out.println(atomic.get()); //30 System.out.println(atomic.getAndSet(10)); //30 System.out.println("====================="); System.out.println(atomic.getAndUpdate(operand -> operand * 20)); // 10 System.out.println(atomic.get()); //200 }}另外提到一嘴，基本数据类型只给了Integer、Long、Boolean，那其他的基本数据类型呢？其实看下AtomicBoolean的源码我们发现其实他本质上是转成了Integer处理的，那么针对其他的类型也可以参考这个思路来实现。
数组针对数组类型的原子操作提供了3个，可以方便的更新数组中的某个元素。
AtomicIntegerArray：针对Integer数组的原子操作。
AtomicLongArray：针对Long数组的原子操作。
AtomicReferenceArray：针对引用类型数组的原子操作。
和上面说的Atomic其实也没有太大的区别，还是以AtomicIntegerArray举例说明，主要方法也基本一样。
public final int getAndIncrement(int i);public final int getAndDecrement(int i);public final int getAndAdd(int i, int delta);public final int getAndSet(int i, int newValue);public final int getAndAccumulate(int i, int x,IntBinaryOperator accumulatorFunction);public final int getAndUpdate(int i, IntUnaryOperator updateFunction);操作一模一样，只是多了一个参数表示当前索引的位置，同样有incrementAndGet等一套方法，返回最新值，没有区别，对于引用类型AtomicReferenceArray来说只是没有了increment和decrement这些方法，其他的也都大同小异，不再赘述。
说实话，这个都没有举栗子的必要。
public class AtomicIntegerArrayTest { public static void main(String<> args) { int<> array = {10}; AtomicIntegerArray atomic = new AtomicIntegerArray(array); System.out.println(atomic.getAndIncrement(0)); //10 System.out.println(atomic.get(0));//11 System.out.println(atomic.getAndDecrement(0)); //11 System.out.println(atomic.getAndAdd(0, 2));//10 System.out.println(atomic.getAndSet(0, 10)); //12 System.out.println(atomic.get(0)); //10 System.out.println("====================="); System.out.println(atomic.getAndAccumulate(0, 3, (left, right) -> left * right)); // 10 System.out.println(atomic.get(0)); //30 System.out.println(atomic.getAndSet(0, 10)); //30 System.out.println("====================="); System.out.println(atomic.getAndUpdate(0, operand -> operand * 20)); // 10 System.out.println(atomic.get(0)); //200 }}引用类型像AtomicInteger那种，只能原子更新一个变量，如果需要同时更新多个变量，就需要使用我们的引用类型的原子类，针对引用类型的原子操作提供了3个。
AtomicReference：针对引用类型的原子操作。
AtomicMarkableReference：针对带有标记位的引用类型的原子操作。
AtomicStampedReference：针对带有标记位的引用类型的原子操作。
AtomicMarkableReference和AtomicStampedReference非常类似，他们是为了解决CAS中的ABA的问题（别说你不知道啥是ABA问题），只不过这个标记的类型不同，我们看下源码。
AtomicMarkableReference标记类型是布尔类型，所以其实他版本就俩，true和false。
AtomicMarkableReference标记类型是整型，那可不就是正常的版本号嘛。
public class AtomicMarkableReference<V> { private static class Pair<T> { final T reference; final boolean mark; //标记 }}public class AtomicStampedReference<V> { private static class Pair<T> { final T reference; final int stamp; // 标记 }}方法还是那几个，老样子。
public final V getAndSet(V newValue);public final V getAndUpdate(UnaryOperator<V> updateFunction);public final V getAndAccumulate(V x, BinaryOperator<V> accumulatorFunction);public final boolean compareAndSet(V expect, V update);简单举个栗子：
public class AtomicReferenceTest { public static void main(String<> args) { User user = new User(1L, "test", "test"); AtomicReference<User> atomic = new AtomicReference<>(user); User pwdUpdateUser = new User(1L,"test","newPwd"); System.out.println(atomic.getAndSet(pwdUpdateUser)); System.out.println(atomic.get()); } @NoArgsConstructor @AllArgsConstructor @Data @ToString static class User { private Long id; private String username; private String password; }}//输出AtomicReferenceTest.User(id=1, username=test, password=test)AtomicReferenceTest.User(id=1, username=test, password=newPwd)对象属性针对对象属性的原子操作也还是提供了3个。
AtomicIntegerFieldUpdater：针对引用类型里的整型属性的原子操作。
AtomicLongFieldUpdater：针对引用类型里的长整型属性的原子操作。
AtomicReferenceFieldUpdater：针对引用类型里的属性的原子操作。
需要注意的是，需要更新的属性字段不能是private，并且必须用volatile修饰，否则会报错。
举个栗子：
public class AtomicReferenceFieldTest { public static void main(String<> args) { AtomicReferenceFieldUpdater<User, String> atomic = AtomicReferenceFieldUpdater.newUpdater(User.class, String.class, "password"); User user = new User(1L, "test", "test"); System.out.println(atomic.getAndSet(user, "newPwd")); System.out.println(atomic.get(user)); } @NoArgsConstructor @AllArgsConstructor @Data @ToString static class User { private Long id; private String username; volatile String password; }}//输出testnewPwd累加器累加器有4个，都来自JDK1.8新增的，为啥新增呢？因为Doug大佬觉得AtomicLong还不够快，虽然说通过CAS操作已经很快了，但是众所知周，高并发同时操作一个共享变量只有一个成功，那其他的线程都在无限自旋，大量的浪费了CPU的资源，所以累加器Accumulator的思路就是把一个变量拆成多个变量，这样多线程去操作竞争多个变量资源，性能不就提升了嘛。
也就是说，在高并发的场景下，可以尽量的使用下面这些类来替换基础类型操作的那些AtomicLong之类的，可以提高性能。
LongAdder：Long类型的累加，LongAccumulator的特例。
LongAccumulator：Long类型的累加。
DoubleAdder：Double类型的累加，DoubleAccumulator的特例。
DoubleAccumulator:Double类型的累加。
由于LongAdder和DoubleAdder都是一样的，我们以LongAdder和LongAccumulator举例来说明它的一些简单的原理。
LongAdder
它继承自Striped64，内部维护了一个Cell数组，核心思想就是把单个变量的竞争拆分，多线程下如果一个Cell竞争失败，转而去其他Cell再次CAS重试。
transient volatile Cell<> cells;transient volatile long base;在计算当前值的时候，则是累加所有cell的value再加上base。
public long sum() { Cell<> as = cells; Cell a; long sum = base; if (as != null) { for (int i = 0; i < as.length; ++i) { if ((a = as) != null) sum += a.value; } } return sum;}这里还涉及到一个伪共享的概念，至于啥是伪共享，看看之前我写的真实字节二面：什么是伪共享？。
解决伪共享的真正的核心就在Cell数组，可以看到，Cell数组使用了Contented注解。
@sun.misc.Contended static final class Cell { volatile long value; Cell(long x) { value = x; }}在上面我们提到数组的内存地址都是连续的，所以数组内的元素经常会被放入一个缓存行，这样的话就会带来伪共享的问题，影响性能，这里使用Contented进行填充，就避免了伪共享的问题，使得数组中的元素不再共享一个缓存行。
LongAccumulator
上面说到，LongAdder其实就是LongAccumulator的一个特例，相比LongAdder他的功能会更加强大，可以自定义累加的规则，在上面演示AtomicInteger功能的时候其实我们也使用过了。
*** ***，实际上就是实现了一个LongAdder的功能，初始值我们传入0，而LongAdder的初始值就是0并且只能是0。
public class LongAdderTest { public static void main(String<> args) { LongAdder longAdder = new LongAdder(); LongAccumulator accumulator = new LongAccumulator((left, right) -> 0, 0); }}工具类&容器类这里要说到一些我们在平时开发中经常使用到的一些类以及他们的实现原理。
工具类&容器类
CountDownLatchCountDownLatch适用于在多线程的场景需要等待所有子线程全部执行完毕之后再做操作的场景。
假设现在我们有一个业务场景，我们需要调用多个RPC接口去查询数据并且写入excel，最后把所有excel打包压缩发送邮件出去。
public class CountDownLatchTest { public static void main(String<> args) throws Exception{ ExecutorService executorService = Executors.newFixedThreadPool(10); CountDownLatch countDownLatch = new CountDownLatch(2); executorService.submit(()->{ try { Thread.sleep(1000); System.out.println("写excelA完成"); countDownLatch.countDown(); } catch (InterruptedException e) { throw new RuntimeException(e); } }); executorService.submit(()->{ try { Thread.sleep(3000); System.out.println("写excelB完成"); countDownLatch.countDown(); } catch (InterruptedException e) { throw new RuntimeException(e); } }); System.out.println("等待excel写入完成"); countDownLatch.await(); System.out.println("开始打包发送数据.."); executorService.shutdown(); }}//输出等待excel写入完成写excelA完成写excelB完成开始打包发送数据..整个过程如下：
初始化一个CountDownLatch实例传参2，因为我们有2个子线程，每次子线程执行完毕之后调用countDown()方法给计数器-1，主线程调用await()方法后会被阻塞，直到最后计数器变为0，await()方法返回，执行完毕。
他和join有个区别，像我们这里用的是ExecutorService创建线程池，是没法使用join的，相比起来，CountDownLatch的使用会显得更加灵活。
CountDownLatch基于AQS实现，用volatile修饰state变量维持倒数状态，多线程共享变量可见。
CountDownLatch通过构造函数初始化传入参数实际为AQS的state变量赋值，维持计数器倒数状态当主线程调用await()方法时，当前线程会被阻塞，当state不为0时进入AQS阻塞队列等待。其他线程调用countDown()时，通过CAS修改state值-1，当state值为0的时候，唤醒所有调用await()方法阻塞的线程CyclicBarrierCyclicBarrier叫做回环屏障，它的作用是让一组线程全部达到一个状态之后再全部同时执行，他和CountDownLatch主要区别在于，CountDownLatch的计数器只能用一次，而CyclicBarrier的计数器状态则是可以一直重用的。
我们可以使用CyclicBarrier一样实现上面的需求。
public class CyclicBarrierTest { public static void main(String<> args) throws Exception{ ExecutorService executorService = Executors.newFixedThreadPool(10); CyclicBarrier cyclicBarrier = new CyclicBarrier(2, () -> { System.out.println("开始打包发送数据.."); }); executorService.submit(()->{ try { Thread.sleep(1000); System.out.println("写excelA完成"); cyclicBarrier.await(); } catch (Exception e) { throw new RuntimeException(e); } }); executorService.submit(()->{ try { Thread.sleep(3000); System.out.println("写excelB完成"); cyclicBarrier.await(); } catch (Exception e) { throw new RuntimeException(e); } }); System.out.println("等待excel写入完成"); executorService.shutdown(); }}//输出等待excel写入完成写excelA完成写excelB完成开始打包发送数据..初始化的时候我们传入2个线程和一个回调方法，线程调用await()之后进入阻塞状态并且计数器-1，这个阻塞点被称作为屏障点或者同步点，只有最后一个线程到达屏障点的时候，所有被屏障拦截的线程才能继续运行，这也是叫做回环屏障的名称原因。
而当计数器为0时，就去执行CyclicBarrier构造函数中的回调方法，回调方法执行完成之后，就会退出屏障点，唤醒其他阻塞中的线程。
CyclicBarrier基于ReentrantLock实现，本质上还是基于AQS实现的，内部维护parties记录总线程数，count用于计数，最开始count=parties，调用await()之后count原子递减，当count为0之后，再次将parties赋值给count，这就是复用的原理。
当子线程调用await()方法时，获取独占锁ReentrantLock，同时对count递减，进入阻塞队列，然后释放锁当第一个线程被阻塞同时释放锁之后，其他子线程竞争获取锁，操作同1直到最后count为0，执行CyclicBarrier构造函数中的任务，执行完毕之后子线程继续向下执行，计数重置，开始下一轮循环SemaphoreSemaphore叫做信号量，和前面两个不同的是，他的计数器是递增的，信号量这玩意儿在限流中就经常使用到。
public class SemaphoreTest { public static void main(String<> args) throws Exception { ExecutorService executorService = Executors.newFixedThreadPool(10); Semaphore semaphore = new Semaphore(0); executorService.submit(() -> { try { Thread.sleep(1000); System.out.println("写excelA完成"); semaphore.release(); } catch (InterruptedException e) { throw new RuntimeException(e); } }); executorService.submit(() -> { try { Thread.sleep(3000); System.out.println("写excelB完成"); semaphore.release(); } catch (InterruptedException e) { throw new RuntimeException(e); } }); System.out.println("等待excel写入完成"); semaphore.acquire(2); System.out.println("开始打包发送数据.."); executorService.shutdown(); }}//输出等待excel写入完成写excelA完成写excelB完成开始打包发送数据..稍微和前两个有点区别，构造函数接受参数表示可用的许可证的数量，acquire方法表示获取一个许可证，使用完之后release归还许可证。
当子线程调用release()方法时，计数器递增，主线程acquire()传参为2则说明主线程一直阻塞，直到计数器为2才会返回。
Semaphore还还还是基于AQS实现的，同时获取信号量有公平和非公平两种策略，通过构造函数的传参可以修改，默认则是非公平的策略。
先说非公平的策略，主线程调用acquire()方法时，用当前信号量值-需要获取的值，如果小于0，说明还没有达到信号量的要求值，则会进入AQS的阻塞队列，大于0则通过CAS设置当前信号量为剩余值，同时返回剩余值。而对于公平策略来说，如果当前有其他线程在等待获取资源，那么自己就会进入AQS阻塞队列排队。子线程调用release()给当前信号量值计数器+1(增加的值数量由传参决定)，同时不停的尝试唤醒因为调用acquire()进入阻塞的线程ExchangerExchanger用于两个线程之间交换数据，如果两个线程都到达同步点，这两个线程可以互相交换他们的数据。
举个栗子，A和B两个线程需要交换他们自己写的数据以便核对数据是否一致。
public class ExchangerTest { public static void main(String<> args) throws Exception { ExecutorService executorService = Executors.newFixedThreadPool(10); Exchanger<String> exchanger = new Exchanger<>(); executorService.submit(() -> { try { Thread.sleep(1000); System.out.println("写excelA完成"); System.out.println("A获取到数据=" + exchanger.exchange("excelA")); } catch (InterruptedException e) { throw new RuntimeException(e); } }); executorService.submit(() -> { try { Thread.sleep(3000); System.out.println("写excelB完成"); System.out.println("B获取到数据=" + exchanger.exchange("excelB")); } catch (InterruptedException e) { throw new RuntimeException(e); } }); executorService.shutdown(); }}//输出写excelA完成写excelB完成B获取到数据=excelAA获取到数据=excelBA写完之后exchange会一直阻塞等待，直到另外一个线程也exchange之后，才会继续执行。
ThreadLocalRandom通常我们都会用 Random 去生成随机数，但是 Random 有点小问题，在多线程并发的情况下为了保证生成的随机性，通过 CAS 的方式保证生成新种子的原子性，但是这样带来了性能的问题，多线程并发去生成随机数，但是只有一个线程能成功，其他的线程会一直自旋，性能不高，所以 ThreadLocalRandom 就是为了解决这个问题而诞生。
//多线程下通过CAS保证新种子生成的原子性protected int next(int bits) { long oldseed, nextseed; AtomicLong seed = this.seed; do { oldseed = seed.get(); nextseed = (oldseed * multiplier + addend) & mask; } while (!seed.compareAndSet(oldseed, nextseed)); return (int)(nextseed >>> (48 - bits));}ThreadLocalRandom 我们从名字就能看出来，肯定使用了 ThreadLocal，作用就是用 ThreadLocal 保存每个种子的变量，防止在高并发下对同一个种子的争夺。
使用也非常简单：
ThreadLocalRandom.current().nextInt(100);看下源码实现，current 方法获取当前的 ThreadLocalRandom 实例。
public static ThreadLocalRandom current() { if (UNSAFE.getInt(Thread.currentThread(), PROBE) == 0) localInit(); return instance;}nextInt 方法和 Random 看起来差不多，上面是生成新的种子，下面是固定的基于新种子计算随机数，主要看 nextSeed。
public int nextInt(int bound) { if (bound <= 0) throw new IllegalArgumentException(BadBound); int r = mix32(nextSeed()); //生成新种子 int m = bound - 1; if ((bound & m) == 0) // power of two r &= m; else { // reject over-represented candidates for (int u = r >>> 1; u + m - (r = u % bound) < 0; u = mix32(nextSeed()) >>> 1) ; } return r;}r = UNSAFE.getLong(t, SEED) + GAMMA 计算出新的种子，然后使用 UNSAFE 的方法放入当前线程中。
final long nextSeed() { Thread t; long r; // read and update per-thread seed UNSAFE.putLong(t = Thread.currentThread(), SEED, r = UNSAFE.getLong(t, SEED) + GAMMA); return r;}ConcurrentHashMap这个我们就不说了，说的太多了，之前的文章也写过了，可以参考之前写过的。
CopyOnWriteArrayList&CopyOnWriteArraySet这是线程安全的 ArrayList ，从名字我们就能看出来，写的时候复制，这叫做写时复制，也就是写的操作是对拷贝的数组的操作。
先看构造函数，有3个，分别是无参，传参为集合和传参数组，其实都差不多，无参构造函数创建一个新的数组，集合则是把集合类的元素拷贝到新的数组，数组也是一样。
public CopyOnWriteArrayList() { setArray(new Object<0>);}public CopyOnWriteArrayList(Collection<? extends E> c) { Object<> elements; if (c.getClass() == CopyOnWriteArrayList.class) elements = ((CopyOnWriteArrayList<?>)c).getArray(); else { elements = c.toArray(); if (c.getClass() != ArrayList.class) elements = Arrays.copyOf(elements, elements.length, Object<>.class); } setArray(elements);}public CopyOnWriteArrayList(E<> toCopyIn) { setArray(Arrays.copyOf(toCopyIn, toCopyIn.length, Object<>.class));}我们看 add 方法，你一眼就能看出来非常简单的实现，通过 ReentrantLock 加锁，然后拷贝出一个新的数组，数组长度+1，再把新数组赋值，所以这就是名字的由来，写入的时候操作的是数组的拷贝，其他的删除修改就不看了，基本上是一样的。
public boolean add(E e) { final ReentrantLock lock = this.lock; lock.lock(); try { Object<> elements = getArray(); int len = elements.length; Object<> newElements = Arrays.copyOf(elements, len + 1); newElements
public E get(int index) { return get(getArray(), index);}private E get(Object<> a, int index) { return (E) a
public CopyOnWriteArraySet() { al = new CopyOnWriteArrayList<E>();}public boolean add(E e) { return al.addIfAbsent(e);}public boolean addIfAbsent(E e) { Object<> snapshot = getArray(); return indexOf(e, snapshot, 0, snapshot.length) >= 0 ? false : addIfAbsent(e, snapshot);}Fork/JoinFork/Join 是一个并行执行任务的框架，利用的分而治之的思想。
Fork 是把一个大的任务拆分成若干个小任务并行执行，Join 则是合并拆分的子任务的结果集，最终计算出大任务的结果。
所以整个 Fork/Join 的流程可以认为就是两步：
Fork 拆分任务，直到拆分到最小粒度不可拆分为止Join 计算结果，把每个子任务的结果进行合并
fork:join
这里我们需要介绍一下主要的几个类：
ForkJoinTask：就是我们的分治任务的抽象类
RecursiveTask：继承于 ForkJoinTask，用于计算有返回结果的任务
RecursiveAction： 继承于 ForkJoinTask，用于计算没有返回结果的任务
ForkJoinPool：用于执行 ForkJoinTask 任务的线程池，通常我们可以用 ForkJoinPool.commonPool() 去创建一个 Fork/Join 的线程池，然后用 submit 或者 invoke 去提交执行任务。
这里我们写一个测试程序，用于计算<0,999>的求和结果，所以我们写一个类继承 RecursiveTask ，并且实现他的 compute 方法。
invokeAll() 相当于每个任务都执行 fork，fork 之后会再次执行 compute 判断是否要继续拆分，如果无需拆分那么则使用 join 方法计算汇总结果。
public class ForkJoinTest { public static void main(String<> args) throws Exception { List<Integer> list = new LinkedList<>(); Integer sum = 0; for (int i = 0; i < 1000; i++) { list.add(i); sum += i; } CalculateTask task = new CalculateTask(0, list.size(), list); Future<Integer> future = ForkJoinPool.commonPool().submit(task); System.out.println("sum=" + sum + "，Fork/Join result=" + future.get()); } @Data static class CalculateTask extends RecursiveTask<Integer> { private Integer start; private Integer end; private List<Integer> list; public CalculateTask(Integer start, Integer end, List<Integer> list) { this.start = start; this.end = end; this.list = list; } @Override protected Integer compute() { Integer sum = 0; if (end - start < 200) { for (int i = start; i < end; i++) { sum += list.get(i); } } else { int middle = (start + end) / 2; System.out.println(String.format("从<%d,%d>拆分为：<%d,%d>,<%d,%d>", start, end, start, middle, middle, end)); CalculateTask task1 = new CalculateTask(start, middle, list); CalculateTask task2 = new CalculateTask(middle, end, list); invokeAll(task1, task2); sum = task1.join() + task2.join(); } return sum; } }}//输出从<0,1000>拆分为：<0,500>,<500,1000>从<0,500>拆分为：<0,250>,<250,500>从<500,1000>拆分为：<500,750>,<750,1000>从<0,250>拆分为：<0,125>,<125,250>从<250,500>拆分为：<250,375>,<375,500>从<500,750>拆分为：<500,625>,<625,750>从<750,1000>拆分为：<750,875>,<875,1000>sum=499500，Fork/Join result=499500使用完成之后，我们再来谈一下 Fork/Join 的原理。
先看 fork 的代码，调用 fork 之后，使用workQueue.push() 把任务添加到队列中，注意 push 之后调用 signalWork 唤醒一个线程去执行任务。
public final ForkJoinTask<V> fork() { Thread t; if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ((ForkJoinWorkerThread)t).workQueue.push(this); else ForkJoinPool.common.externalPush(this); return this;}final ForkJoinPool.WorkQueue workQueue; // 工作窃取 final void push(ForkJoinTask<?> task) { ForkJoinTask<?><> a; ForkJoinPool p; int b = base, s = top, n; if ((a = array) != null) { // ignore if queue removed int m = a.length - 1; // fenced write for task visibility U.putOrderedObject(a, ((m & s) << ASHIFT) + ABASE, task); U.putOrderedInt(this, QTOP, s + 1); if ((n = s - b) <= 1) { if ((p = pool) != null) p.signalWork(p.workQueues, this); } else if (n >= m) growArray(); }}上面我们看到了 workQueue，这个其实就是我们说的工作队列，它是一个双端队列，并且有一个工作线程和他对应。
@sun.misc.Contendedstatic final class WorkQueue { volatile int base; // 下一个出队列索引 int top; // 下一个入队列索引 ForkJoinTask<?><> array; // 队列中的 task final ForkJoinPool pool; final ForkJoinWorkerThread owner; // 工作队列中的工作线程 volatile Thread parker; // == owner during call to park; else null volatile ForkJoinTask<?> currentJoin; // 当前join的任务 volatile ForkJoinTask<?> currentSteal; // 当前偷到的任务}那如果工作线程自己队列的做完了怎么办？只能傻傻地等待吗？并不是，这时候有一个叫做工作窃取的机制，所以他就会去其他线程的队列里偷一个任务来执行。
为了避免偷任务线程和自己的线程产生竞争，所以自己的工作线程是从队列头部获取任务执行，而偷任务线程则从队列尾部偷任务。
工作窃取
ExecutorExecutor是并发编程中重要的一环，任务创建后提交到Executor执行并最终返回结果。
Executor
任务线程两种创建方式：Runnable和Callable。
Runnable是最初创建线程的方式，在JDK1.1的版本就已经存在，Callable则在JDK1.5版本之后加入，他们的主要区别在于Callable可以返回任务的执行结果。
任务执行任务的执行主要靠Executor，ExecutorService继承自Executor，ThreadPoolExecutor和ScheduledThreadPoolExecutor分别实现了ExecutorService。
那说到线程池之前，我们肯定要提及到线程池的几个核心参数和原理，这个之前的文章也写到过，属于基础中的基础部分。
首先线程池有几个核心的参数概念：
最大线程数maximumPoolSize核心线程数corePoolSize活跃时间keepAliveTime阻塞队列workQueue拒绝策略RejectedExecutionHandler当提交一个新任务到线程池时，具体的执行流程如下：
当我们提交任务，线程池会根据corePoolSize大小创建若干任务数量线程执行任务当任务的数量超过corePoolSize数量，后续的任务将会进入阻塞队列阻塞排队当阻塞队列也满了之后，那么将会继续创建(maximumPoolSize-corePoolSize)个数量的线程来执行任务，如果任务处理完成，maximumPoolSize-corePoolSize额外创建的线程等待keepAliveTime之后被自动销毁如果达到maximumPoolSize，阻塞队列还是满的状态，那么将根据不同的拒绝策略对应处理
拒绝策略主要有四种：
AbortPolicy：直接丢弃任务，抛出异常，这是默认策略CallerRunsPolicy：使用调用者所在的线程来处理任务DiscardOldestPolicy：丢弃等待队列中最老的任务，并执行当前任务DiscardPolicy：直接丢弃任务，也不抛出异常ThreadPoolExecutor通常为了快捷我们会用Executors工具类提供的创建线程池的方法快速地创建一个线程池出来，主要有几个方法，但是一般我们不推荐这样使用，非常容易导致出现问题，生产环境中我们一般推荐自己实现，参数自己定义，而不要使用这些方法。
创建
//创建固定线程数大小的线程池，核心线程数=最大线程数，阻塞队列长度=Integer.MAX_VALUEpublic static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());}//创建只有一个线程的线程池，阻塞队列长度=Integer.MAX_VALUEpublic static ExecutorService newSingleThreadExecutor() { return new FinalizableDelegatedExecutorService (new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()));}//创建核心线程数为0，最大线程数=Integer.MAX_VALUE的线程池，阻塞队列为同步队列public static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());}最好的办法就是自己创建，并且指定线程名称：
new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(), Runtime.getRuntime().availableProcessors()*2,1000L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<>(100),new ThreadFactoryBuilder().setNameFormat("thread-name").build());提交任务
重点说一下几个方法：
submit(Runnable task, T result)：可以用于主线程和子线程之间的通信，数据共享。
submit(Runnable task)：返回null，相当于调用submit(Runnable task, null)。
invokeAll(Collection<? extends Callable> tasks)：批量提交任务，阻塞等待所有任务执行完成之后返回，带超时时间的则是在超时之后返回，并且取消没有执行完成的任务。
invokeAny(Collection<? extends Callable> tasks)：批量提交任务，只要一个任务有返回，那么其他的任务都会被终止。
public void execute(Runnable command); //提交runnable任务，无返回public <T> Future<T> submit(Callable<T> task); //提交callable任务，有返回public Future<?> submit(Runnable task); //提交runnable，有返回public <T> Future<T> submit(Runnable task, T result); //提交runnable，有返回public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks); //批量提交任务public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,long timeout, TimeUnit unit);public <T> T invokeAny(Collection<? extends Callable<T>> tasks);public <T> T invokeAny(Collection<? extends Callable<T>> tasks,long timeout, TimeUnit unit);关闭
shutdown：线程池状态设置为SHUTDOWN，不再接受新任务，直接返回，线程池中任务会执行完成，遍历线程池中的线程，逐个调用interrupt方法去中断线程。
shutdownNow：线程池状态设置为STOP，不再接受新任务，直接返回，线程池中任务会被中断，返回值为被丢弃的任务列表。
isShutdown：只要调用了shutdown或者shutdownNow，都会返回true
isTerminating：所有任务都关闭后，才返回true
public void shutdown();public List<Runnable> shutdownNow();public boolean isShutdown();public boolean isTerminating();ScheduledThreadPoolExecutorScheduledThreadPoolExecutor 继承于 ThreadPoolExecutor，从名字我们也知道，他是用于定时执行任务的线程池。
内部实现了一个DelayedWorkQueue作为任务的阻塞队列，ScheduledFutureTask 作为调度的任务，保存到队列中。
我们先看下他的构造函数，4个构造函数都不支持传队列进来，所以默认的就是使用他的内部类 DelayedWorkQueue，由于 DelayedWorkQueue 是一个无界队列，所以这里最大线程数都是设置的为 Integer.MAX，因为没有意义。
public ScheduledThreadPoolExecutor(int corePoolSize) { super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS, new DelayedWorkQueue());}public ScheduledThreadPoolExecutor(int corePoolSize, ThreadFactory threadFactory) { super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS, new DelayedWorkQueue(), threadFactory);}public ScheduledThreadPoolExecutor(int corePoolSize, RejectedExecutionHandler handler) { super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS, new DelayedWorkQueue(), handler);}public ScheduledThreadPoolExecutor(int corePoolSize, ThreadFactory threadFactory, RejectedExecutionHandler handler) { super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS, new DelayedWorkQueue(), threadFactory, handler);}执行定时任务的方法主要有4个，前面两个 schedule 传参区分 Runnable 和 Callable 其实并没有区别，最终 Runnable 会通过 Executors.callable(runnable, result) 转换为 Callable，本质上我们可以当做只有3个执行方法来看。
public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit); public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit); public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit); public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit);schedule：提交一个延时任务，从时间单位为 unit 的 delay 时间开始执行，并且任务只会执行一次。
scheduleWithFixedDelay：以固定的延迟时间重复执行任务，initialDelay 表示提交任务后多长时间开始执行，delay 表示任务执行时间间隔。
scheduleAtFixedRate：以固定的时间频率重复执行任务，指的是以起始时间开始，然后以固定的时间间隔重复执行任务，initialDelay 表示提交任务后多长时间开始执行，然后从 initialDelay + N * period执行。
这两个特别容易搞混，很难理解到底是个啥意思，记住了。
scheduleAtFixedRate 是上次执行完成之后立刻执行，scheduleWithFixedDelay 则是上次执行完成+delay 后执行。
看个例子，两个任务都会延迟1秒，然后以2秒的间隔开始重复执行，任务睡眠1秒的时间。
scheduleAtFixedRate 由于任务执行的耗时比时间间隔小，所以始终是以2秒的间隔在执行。
scheduleWithFixedDelay 因为任务耗时用了1秒，导致后面的时间间隔都成了3秒。
public class ScheduledThreadPoolTest { public static void main(String<> args) throws Exception { ScheduledExecutorService executorService = new ScheduledThreadPoolExecutor(10); executorService.scheduleAtFixedRate(() -> { try { System.out.println("scheduleAtFixedRate=" + new SimpleDateFormat("HH:mm:ss").format(new Date())); Thread.sleep(1000); } catch (InterruptedException e) { throw new RuntimeException(e); } }, 1000, 2000, TimeUnit.MILLISECONDS); executorService.scheduleWithFixedDelay(() -> { try { System.err.println("scheduleWithFixedDelay=" + new SimpleDateFormat("HH:mm:ss").format(new Date())); Thread.sleep(1000); } catch (InterruptedException e) { throw new RuntimeException(e); } }, 1000, 2000, TimeUnit.MILLISECONDS);// executorService.shutdown(); }}//输出scheduleAtFixedRate=01:17:05scheduleWithFixedDelay=01:17:05scheduleAtFixedRate=01:17:07scheduleWithFixedDelay=01:17:08scheduleAtFixedRate=01:17:09scheduleAtFixedRate=01:17:11scheduleWithFixedDelay=01:17:11scheduleAtFixedRate=01:17:13scheduleWithFixedDelay=01:17:14scheduleAtFixedRate=01:17:15scheduleAtFixedRate=01:17:17scheduleWithFixedDelay=01:17:17scheduleAtFixedRate=01:17:19scheduleWithFixedDelay=01:17:20scheduleAtFixedRate=01:17:21我们把任务耗时调整到超过时间间隔，比如改成睡眠3秒，观察输出结果。
scheduleAtFixedRate 由于任务执行的耗时比时间间隔长，按照规定上次任务执行结束之后立刻执行，所以变成以3秒的时间间隔执行。
scheduleWithFixedDelay 因为任务耗时用了3秒，导致后面的时间间隔都成了5秒。
scheduleWithFixedDelay=01:46:21scheduleAtFixedRate=01:46:21scheduleAtFixedRate=01:46:24scheduleWithFixedDelay=01:46:26scheduleAtFixedRate=01:46:27scheduleAtFixedRate=01:46:30scheduleWithFixedDelay=01:46:31scheduleAtFixedRate=01:46:33scheduleWithFixedDelay=01:46:36scheduleAtFixedRate=01:46:36OK，最后来说说实现原理：
首先我们通过调用 schedule 的几个方法，把任务添加到 ScheduledThreadPoolExecutor 去执行接收到任务之后，会通过请求参数的延迟时间计算出真正需要执行任务的时间，然后把任务封装成 RunnableScheduledFuture然后把封装之后的任务添加到延迟队列中，任务 ScheduledFutureTask 实现了 comparable 接口，把时间越小的任务放在队列头，如果时间一样，则会通过 sequenceNumber 去比较，也就是执行时间相同，先提交的先执行最后线程池会从延迟队列中去获取任务执行，如果是一次性的任务，执行之后删除队列中的任务，如果是重复执行的，则再次计算时间，然后把任务添加到延迟队列中
CompletionService记得上面我将 ThreadPoolExecutor 的方法吗，其中有一个 invokeAny 的方法，批量提交任务，只要有一个完成了，就直接返回，而不用一直傻傻地等，他的实现就是使用了 CompletionService ，我给你看一段源码。
private <T> T doInvokeAny(Collection<? extends Callable<T>> tasks, boolean timed, long nanos) throws InterruptedException, ExecutionException, TimeoutException { if (tasks == null) throw new NullPointerException(); int ntasks = tasks.size(); if (ntasks == 0) throw new IllegalArgumentException(); ArrayList<Future<T>> futures = new ArrayList<Future<T>>(ntasks); ExecutorCompletionService<T> ecs = new ExecutorCompletionService<T>(this);}看到了吧，OK，在我们想试试使用这个类之前，我们先试试 invokeAny 好使不。
public class CompletionServiceTest { private static final int TOTAL = 10; private static ExecutorService executorService = Executors.newFixedThreadPool(TOTAL); public static void main(String<> args) throws Exception { testInvokeAny(); } private static void testInvokeAny() throws Exception { List<TestTask> taskList = new LinkedList<>(); for (int i = 0; i < TOTAL; i++) { taskList.add(new TestTask(i)); } String value = executorService.invokeAny(taskList, 60, TimeUnit.SECONDS); System.out.println("get value = " + value); executorService.shutdown(); } static class TestTask implements Callable<String> { private Integer index; public TestTask(Integer index) { this.index = index; } @Override public String call() throws Exception { long sleepTime = ThreadLocalRandom.current().nextInt(1000, 10000); System.out.println("task-" + index + " sleep " + sleepTime + " Ms"); Thread.sleep(sleepTime); return "task-" + index; } }}//输出task-7 sleep 3072 Mstask-4 sleep 1186 Mstask-3 sleep 6182 Mstask-9 sleep 7411 Mstask-0 sleep 1882 Mstask-1 sleep 8274 Mstask-2 sleep 4789 Mstask-5 sleep 8894 Mstask-8 sleep 7211 Mstask-6 sleep 5959 Msget value = task-4看到效果了吧，耗时最短的任务返回，整个流程就结束了，那我们试试自己用 CompletionService 来实现这个效果看看。
public static void main(String<> args) throws Exception { // testInvokeAny(); testCompletionService();}private static void testCompletionService() { CompletionService<String> completionService = new ExecutorCompletionService(executorService); List<Future> taskList = new LinkedList<>(); for (int i = 0; i < TOTAL; i++) { taskList.add(completionService.submit(new TestTask(i))); } String value = null; try { for (int i = 0; i < TOTAL; i++) { value = completionService.take().get(); if (value != null) { System.out.println("get value = " + value); break; } } } catch (Exception e) { e.printStackTrace(); } finally { taskList.forEach(task -> { task.cancel(true); }); } executorService.shutdown();}//输出task-4 sleep 5006 Mstask-1 sleep 4114 Mstask-2 sleep 4865 Mstask-5 sleep 1592 Mstask-3 sleep 6190 Mstask-7 sleep 2482 Mstask-8 sleep 9405 Mstask-9 sleep 8798 Mstask-6 sleep 2040 Mstask-0 sleep 2111 Msget value = task-5效果是一样的，我们只是实现了一个简化版的 invokeAny 功能，使用起来也挺简单的。
实现原理也挺简单的，哪个任务先完成，就把他丢到阻塞队列里，这样取任务结果的时候直接从队列里拿，肯定是拿到最新的那一个。
异步结果通常，我们都会用 FutureTask 来获取线程异步执行的结果，基于 AQS 实现。
这个没有说太多的必要，看看几个方法就行了。
public V get();public V get(long timeout, TimeUnit unit);public boolean cancel(boolean mayInterruptIfRunning);get 会阻塞的获取线程异步执行的结果，一般不建议直接使用，最好是使用带超时时间的 get 方法。
我们可以通过 cancel 方法去尝试取消任务的执行，参数代表是否支持中断，如果任务未执行，那么可以直接取消，如果任务执行中，使用 cancel(true) 会尝试中断任务。
CompletableFuture之前我们都在使用 Future，要么只能用 get 方法阻塞，要么就用 isDone 来判断，JDK1.8 之后新增了 CompletableFuture 用于异步编程，它针对 Future 的功能增加了回调能力，可以帮助我们简化异步编程。
CompletableFuture 主要包含四个静态方法去创建对象，主要区别在于 supplyAsync 返回计算结果，runAsync 不返回，另外两个方法则是可以指定线程池，如果不指定线程池则默认使用 ForkJoinPool，默认线程数为CPU核数。
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier);public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor);public static CompletableFuture<Void> runAsync(Runnable runnable);public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor);下面看看他的那些恶心人的几十个方法，我估计能疯。
串行
串行就不用解释了，A->B->C 按照顺序执行，下一个任务必须等上一个任务执行完成才可以。
主要包含 thenApply、thenAccept、thenRun 和 thenCompose，以及他们对应的带 async 的异步方法。
为了方便记忆我们要记住，有 apply 的有传参有返回值，带 accept 的有传参但是没有返回值，带 run 的啥也没有，带 compose 的会返回一个新的 CompletableFuture 实例。
public static void main(String<> args) throws Exception { CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> { try { Thread.sleep(1000); System.out.println(Thread.currentThread() + "工作完成"); return "supplyAsync"; } catch (InterruptedException e) { throw new RuntimeException(e); } }); CompletableFuture newFuture = future.thenApply((ret) -> { System.out.println(Thread.currentThread() + "thenApply=>" + ret); return "thenApply"; }).thenAccept((ret) -> { System.out.println(Thread.currentThread() + "thenAccept=>" + ret); }).thenRun(() -> { System.out.println(Thread.currentThread() + "thenRun"); }); CompletableFuture<String> composeFuture = future.thenCompose((ret) -> { System.out.println(Thread.currentThread() + "thenCompose=>" + ret); return CompletableFuture.supplyAsync(() -> { try { Thread.sleep(1000); System.out.println(Thread.currentThread() + "thenCompose工作完成"); return "thenCompose"; } catch (InterruptedException e) { throw new RuntimeException(e); } }); }); System.out.println(future.get()); System.out.println(newFuture.get()); System.out.println(composeFuture.get());}//输出Thread
这个意思是下一个任务执行必须等前两个任务完成可以。
主要包含 thenCombine、thenAcceptBoth、runAfterBoth ，以及他们对应的带 async 的异步方法，区别和上面一样。
public static void main(String<> args) throws Exception { CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> { try { Thread.sleep(1000); System.out.println(Thread.currentThread() + "A工作完成"); return "A"; } catch (InterruptedException e) { throw new RuntimeException(e); } }); CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> { try { Thread.sleep(2000); System.out.println(Thread.currentThread() + "B工作完成"); return "B"; } catch (InterruptedException e) { throw new RuntimeException(e); } }); CompletableFuture newFuture = future.thenCombine(future2, (ret1, ret2) -> { System.out.println(Thread.currentThread() + "thenCombine=>" + ret1 + "," + ret2); return "thenCombine"; }).thenAcceptBoth(future2, (ret1, ret2) -> { System.out.println(Thread.currentThread() + "thenAcceptBoth=>" + ret1 + "," + ret2); }).runAfterBoth(future2, () -> { System.out.println(Thread.currentThread() + "runAfterBoth"); }); System.out.println(future.get()); System.out.println(future2.get()); System.out.println(newFuture.get());}//输出Thread
Or 聚合代表只要多个任务中有一个完成了，就可以继续下面的任务。
主要包含 applyToEither、acceptEither、runAfterEither ，以及他们对应的带 async 的异步方法，区别和上面一样，不再举例了。
回调/异常处理
whenComplete、handle 代表执行完成的回调，一定会执行，exceptionally 则是任务执行发生异常的回调。
public static void main(String<> args) throws Exception { CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> { try { Thread.sleep(1000); int a = 1 / 0; return "success"; } catch (Exception e) { throw new RuntimeException(e); } }); CompletableFuture newFuture = future.handle((ret, exception) -> { System.out.println(Thread.currentThread() + "handle exception=>" + exception.getMessage()); return "handle"; }); future.whenComplete((ret, exception) -> { System.out.println(Thread.currentThread() + "whenComplete exception=>" + exception.getMessage()); }); CompletableFuture exceptionFuture = future.exceptionally((e) -> { System.out.println(Thread.currentThread() + "exceptionally exception=>" + e.getMessage()); return "exception"; }); System.out.println("task future = " + future.get()); System.out.println("handle future = " + newFuture.get()); System.out.println("exception future = " + exceptionFuture.get());}//输出Thread
阻塞队列
阻塞队列中的阻塞包含两层意思：
插入的时候，如果阻塞队列满，插入元素阻塞删除/查询的时候，如果阻塞队列空，删除/查询元素阻塞下面列出队列的一些插入和删除元素的方法，一个个来说：
add：向队列尾部插入元素，插入成功返回 true，队列满则抛出IllegalStateException("Queue full")异常
offer：向队列尾部插入元素，队列满返回 false，否则返回 true，带超时的则是会阻塞，达到超时时间后返回
put：向队列尾部插入元素，队列满会一直阻塞
remove：删除队列头部元素，删除成功返回 true，队列空则抛出NoSuchElementException异常
poll：删除队列头部元素，删除成功返回队列头部元素，队列空返回null，带超时的则是会阻塞，达到超时时间后返回
take：删除队列头部元素，队列空会一直阻塞
element：查询队列头部元素，并且返回，队列空则抛出NoSuchElementException异常
peek：查询队列头部元素，并且返回
ArrayBlockingQueueArrayBlockingQueue 从名字就知道，基于数组实现的有界阻塞队列，基于AQS支持公平和非公平策略。
还是看构造函数吧，可以传入初始数组大小，一旦设置之后大小就不能改变了，传参可以支持公平和非公平，最后一个构造函数可以支持传入集合进行初始化，但是长度不能超过 capacity，否则抛出ArrayIndexOutOfBoundsException异常。
public ArrayBlockingQueue(int capacity);public ArrayBlockingQueue(int capacity, boolean fair);public ArrayBlockingQueue(int capacity, boolean fair, Collection<? extends E> c);这个其实在上面介绍 Condition 的时候我们就已经实现过他了，这里就不再说了，可以参考上面 Condition 的部分。
LinkedBlockingQueueLinkedBlockingQueue 基于链表实现的有界阻塞队列。
使用无参构造函数则链表长度为 Integer.MAX_VALUE，另外两个构造函数和 ArrayBlockingQueue 差不多。
public LinkedBlockingQueue();public LinkedBlockingQueue(int capacity);public LinkedBlockingQueue(Collection<? extends E> c);我们可以看看 put 和 take 的源码。
首先加锁中断然后判断如果达到了队列的最大长度，那么就阻塞等待，否则就把元素插入到队列的尾部注意这里和 ArrayBlockingQueue 有个区别，这里再次做了一次判断，如果队列没满，唤醒因为 put 阻塞的线程，为什么要做判断，因为他们不是一把锁最后的逻辑是一样的，notEmpty 唤醒public void put(E e) throws InterruptedException { if (e == null) throw new NullPointerException(); int c = -1; Node<E> node = new Node<E>(e); final ReentrantLock putLock = this.putLock; final AtomicInteger count = this.count; putLock.lockInterruptibly(); try { while (count.get() == capacity) { notFull.await(); } enqueue(node); c = count.getAndIncrement(); if (c + 1 < capacity) notFull.signal(); } finally { putLock.unlock(); } if (c == 0) signalNotEmpty();}private void enqueue(Node<E> node) { // assert putLock.isHeldByCurrentThread(); // assert last.next == null; last = last.next = node;}private void signalNotEmpty() { final ReentrantLock takeLock = this.takeLock; takeLock.lock(); try { notEmpty.signal(); } finally { takeLock.unlock(); }}take的逻辑也是非常类似啊。
加锁中断判断队列是不是空了，空了的话就阻塞等待，否则就从队列移除一个元素然后再次做一次判断，队列要是不空，就唤醒阻塞的线程最后唤醒 notFull 的线程public E take() throws InterruptedException { E x; int c = -1; final AtomicInteger count = this.count; final ReentrantLock takeLock = this.takeLock; takeLock.lockInterruptibly(); try { while (count.get() == 0) { notEmpty.await(); } x = dequeue(); c = count.getAndDecrement(); if (c > 1) notEmpty.signal(); } finally { takeLock.unlock(); } if (c == capacity) signalNotFull(); return x;}private E dequeue() { // assert takeLock.isHeldByCurrentThread(); // assert head.item == null; Node<E> h = head; Node<E> first = h.next; h.next = h; // help GC head = first; E x = first.item; first.item = null; return x;}private void signalNotFull() { final ReentrantLock putLock = this.putLock; putLock.lock(); try { notFull.signal(); } finally { putLock.unlock(); }}PriorityBlockingQueuePriorityBlockingQueue 是支持优先级的无界阻塞队列，默认排序按照自然排序升序排列。
几个构造函数，无参构造函数初始容量为11，可以自定义，也可以在创建的时候传入 comparator 自定义排序规则。
public PriorityBlockingQueue();public PriorityBlockingQueue(int initialCapacity);public PriorityBlockingQueue(int initialCapacity, Comparator<? super E> comparator);public PriorityBlockingQueue(Collection<? extends E> c);直接看 put 和 take 方法吧，后面都这样，其他的就忽略好了，找到 put 之后，发现直接就是调用的 offer，那我们就直接看 offer 的实现。
首先还是加锁，然后看当前元素个数是否达到了数组的上限，到了就调用 tryGrow 去扩容。看是否实现了 Comparator 接口，是的话就用 Comparator 去排序，否则就用 Comparable 去比较，如果两个都没有，会报错notEmpty 唤醒，最后解锁public void put(E e) { offer(e); // never need to block}public boolean offer(E e) { if (e == null) throw new NullPointerException(); final ReentrantLock lock = this.lock; lock.lock(); int n, cap; Object<> array; while ((n = size) >= (cap = (array = queue).length)) tryGrow(array, cap); try { Comparator<? super E> cmp = comparator; if (cmp == null) siftUpComparable(n, e, array); else siftUpUsingComparator(n, e, array, cmp); size = n + 1; notEmpty.signal(); } finally { lock.unlock(); } return true;}这里，我们要继续关注一下这个扩容的逻辑，到底是怎么处理的？代码不长，但是看着很方的样子。
首先，先释放锁，因为下面用 CAS 处理，估计怕扩容时间太长阻塞的线程太多然后 CAS 修改 allocationSpinLock 为1CAS 成功的话，进行扩容的逻辑，如果长度小于64就扩容一倍，否则扩容一半之前我们说他无界，其实不太对，这里就判断是否超过了最大长度，MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8，判断一下有可能会抛出内存溢出异常然后创建一个新的对象数组，并且 allocationSpinLock 重新恢复为0执行了一次 Thread.yield()，让出 CPU，因为有可能其他线程正在扩容，让大家争抢一下最后确保新的对象数组创建成功了，也就是扩容是没有问题的，再次加锁，数组拷贝，结束private void tryGrow(Object<> array, int oldCap) { lock.unlock(); // must release and then re-acquire main lock Object<> newArray = null; if (allocationSpinLock == 0 && UNSAFE.compareAndSwapInt(this, allocationSpinLockOffset, 0, 1)) { try { int newCap = oldCap + ((oldCap < 64) ? (oldCap + 2) : // grow faster if small (oldCap >> 1)); if (newCap - MAX_ARRAY_SIZE > 0) { // possible overflow int minCap = oldCap + 1; if (minCap < 0 || minCap > MAX_ARRAY_SIZE) throw new OutOfMemoryError(); newCap = MAX_ARRAY_SIZE; } if (newCap > oldCap && queue == array) newArray = new Object
public E take() throws InterruptedException { final ReentrantLock lock = this.lock; lock.lockInterruptibly(); E result; try { while ( (result = dequeue()) == null) notEmpty.await(); } finally { lock.unlock(); } return result;}DelayQueueDelayQueue 是支持延时的无界阻塞队列，这个在我们聊 ScheduledThreadPoolExecutor 也谈到过，里面也使用了延迟队列，只不过是它自己的一个内部类，DelayQueue 内部其实使用 PriorityQueue 来实现。
DelayQueue 的用法是添加元素的时候可以设置一个延迟时间，当时间到了之后才能从队列中取出来，使用 DelayQueue 中的对象必须实现 Delayed 接口，重写 getDelay 和 compareTo 方法，就像这样，那实现其实可以看 ScheduledThreadPoolExecutor 里面是怎么做的，这里我就不管那么多，示意一下就好了。
public class Test { public static void main(String<> args) throws Exception { DelayQueue<User> delayQueue = new DelayQueue<>(); delayQueue.put(new User(1, "a")); } @Data @NoArgsConstructor @AllArgsConstructor static class User implements Delayed { private Integer id; private String username; @Override public long getDelay(TimeUnit unit) { return 0; } @Override public int compareTo(Delayed o) { return 0; } }}我们可以看看他的属性和构造函数，呐看到了吧，使用的 PriorityQueue，另外构造函数比较简单了，不说了。
private final transient ReentrantLock lock = new ReentrantLock();private final PriorityQueue<E> q = new PriorityQueue<E>();private Thread leader = null;private final Condition available = lock.newCondition();public DelayQueue();public DelayQueue(Collection<? extends E> c);OK，没啥毛病，这里我们要先看 take 方法，不能先看 put，否则我觉得闹不明白。
来第一步加锁，如果头结点是空的，也就是队列是空的话，阻塞，没啥好说的反之队列有东西，我们就要去取了嘛，但是这里要看对象自己实现的 getDelay 方法获得延迟的时间，如果延迟的时间小于0，那说明到时间了，可以执行了，poll 返回第一次，leader 线程肯定是空的，线程阻塞 delay 的时间之后才开始执行，完全没毛病，然后 leader 重新 置为 null当 leader 不是 null 的时候，说明其他线程在操作了，所以阻塞等待唤醒最后，leader 为 null，唤醒阻塞中的线程，解锁public E take() throws InterruptedException { final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try { for (;;) { E first = q.peek(); if (first == null) available.await(); else { long delay = first.getDelay(NANOSECONDS); if (delay <= 0) return q.poll(); first = null; // don't retain ref while waiting if (leader != null) available.await(); else { Thread thisThread = Thread.currentThread(); leader = thisThread; try { available.awaitNanos(delay); } finally { if (leader == thisThread) leader = null; } } } } } finally { if (leader == null && q.peek() != null) available.signal(); lock.unlock(); }}然后再来看 put 就会简单多了，put 还是直接调用的 offer，看 offer 方法。
这里使用的是 PriorityQueue 的 offer 方法，其实和我们上面说到的 PriorityBlockingQueue 差不多，不再多说了，添加到队列头部之后，leader 置为 null，唤醒，结束了。
public boolean offer(E e) { final ReentrantLock lock = this.lock; lock.lock(); try { q.offer(e); if (q.peek() == e) { leader = null; available.signal(); } return true; } finally { lock.unlock(); }}SynchronousQueue&LinkedTransferQueue为什么这两个放一起说呢。。。因为这源码真的不想在这里说一遍，这俩源码可以单独出一个专题来写，长篇精悍文章不适合他他们，就简单先了解下。
SynchronousQueue 是一个不存储元素的阻塞队列，每个 put 必须等待 take，否则不能继续添加元素。
如果你还记得我们上面说到线程池的地方，newCachedThreadPool 默认就是使用的 SynchronousQueue。
他就两个构造方法，你一看就知道，对吧，支持公平和非公平，当然你也别问默认是啥，问就是非公平。
public SynchronousQueue();public SynchronousQueue(boolean fair);主要靠内部抽象类 Transferer，他的实现主要有两个，TransferQueue 和 TransferStack。
注意：如果是公平模式，使用的是 TransferQueue 队列，非公平则使用 TransferStack 栈。
abstract static class Transferer<E> { abstract E transfer(E e, boolean timed, long nanos);}LinkedTransferQueue 是链表组成的无界阻塞队列，看他内部类就知道了，这是个链表实现。
static final class Node { final boolean isData; // 标记生产者或者消费者 volatile Object item; // 值 volatile Node next; // 下一个节点 volatile Thread waiter;}LinkedBlockingDequeLinkedBlockingDeque 是链表组成的双向阻塞队列，它支持从队列的头尾进行进行插入和删除元素。
构造函数有3个，不传初始容量就是 Integer 最大值。
public LinkedBlockingDeque() { this(Integer.MAX_VALUE);}public LinkedBlockingDeque(int capacity);public LinkedBlockingDeque(Collection<? extends E> c);看下双向链表的结构：
static final class Node<E> { E item; Node<E> prev; Node<E> next; Node(E x) { item = x; }}因为是双向链表，所以比其他的队列多了一些方法，比如 add、addFirst、addLast，add 其实就是 addLast，offer、put 也是类似。
我们可以区分看一下 putFirst 和 putLast ，主要区别就是 linkFirst 和 linkLast，分别去队列头部和尾部添加新节点，其他基本一致。
public void putFirst(E e) throws InterruptedException { if (e == null) throw new NullPointerException(); Node<E> node = new Node<E>(e); final ReentrantLock lock = this.lock; lock.lock(); try { while (!linkFirst(node)) notFull.await(); } finally { lock.unlock(); }}public void putLast(E e) throws InterruptedException { if (e == null) throw new NullPointerException(); Node<E> node = new Node<E>(e); final ReentrantLock lock = this.lock; lock.lock(); try { while (!linkLast(node)) notFull.await(); } finally { lock.unlock(); }}结尾本次长篇内容参考书籍和文档
Java 并发编程的艺术Java 并发编程之美Java 并发编程实战Java 8实战极客时间：Java 并发编程实战OK，本期内容到此结束，我是艾小仙，我们过两个月再见。
