当前位置：自学咖网 > 技术教程 > 编程最新 > 并发编程从入门到放弃系列开始和结束
wenwen 编程最新

并发编程从入门到放弃系列开始和结束

并发编程从入门到放弃系列开始和结束

对于 Java 部分的面试来说，突然想到并发这一块的内容是不太完整的，这篇文章会通篇把多线程和并发都大致阐述一遍，至少能够达到了解原理和使用的目的，内容会比较多，从最基本的线程到我们常用的类会统一说一遍，慢慢看。

并发编程

进程&线程

对于基本的概念，大家应该都很熟悉了，进程是资源分配的单位，线程是CPU调度的单位，线程是进程中的一个实体。

对于我们的Java程序来说，天生就是多线程的，我们通过main方法启动，就是启动了一个JVM的进程，同时创建一个名为main的线程，main就是JVM进程中的一个实体线程。

进程&线程

线程生命周期

线程几种基本状态：

  1. New，初始状态，就是New了一个线程，但是还没有调用start方法
  2. Runnable，可运行Ready或者运行Running状态，线程的就绪和运行中状态我们统称为Runnable运行状态
  3. Blocked/Wating/Timed_Wating，这些状态统一就叫做休眠状态
  4. Terminated，终止状态

几个状态之间的转换我们分别来说。

New：我们创建一个线程，但是线程没有调用start方法，就是初始化状态。

Runnable：调用start()启动线程进入Ready可运行状态，等待CPU调度之后进入到Running状态。

Blocked：阻塞状态，当线程在等待进入synchronized锁的时候，进入阻塞状态。

Waiting：等待状态需要被显示的唤醒，进入该状态分为三种情况，在synchonized中调用Object.wait()，调用Thread.join()，调用LockSupport.park()。

Timed_Waiting：和Waiting的区别就是多了超时时间，不需要显示唤醒，达到超时时间之后自动唤醒，调用图中的一些带有超时参数的方法则会进入该状态。

Terminated：终止状态，线程执行完毕。

守护线程&用户线程

Java中的线程分为守护线程和用户线程，上面我们提到的main线程其实就是一个用户线程。

他们最主要的区别就在于，只要有非守护线程没有结束，JVM就不会正常退出，而守护线程则不会影响JVM的退出。

可以通过简单的方法设置一个线程为守护线程。

 Thread t = new Thread();
 t.setDaemon(true);

锁

锁是控制多线程并发访问共享资源的方式，为了更简单快速的了解Java中的锁，我们可以按照显示锁和隐式锁来做一个大致的区分。

锁

隐式锁

在没有Lock接口之前，加锁通过synchronzied实现，在之前的Java基础系列中我已经说过了，就不在这里过多的阐述，此处引用之前写过的，更多详细可以看《我想进大厂》之Java基础夺命连环16问。

synchronized是java提供的原子性内置锁，这种内置的并且使用者看不到的锁也被称为监视器锁，使用synchronized之后，会在编译之后在同步的代码块前后加上monitorenter和monitorexit字节码指令，他依赖操作系统底层互斥锁实现，主要作用就是实现原子性操作和解决共享变量的内存可见性问题。

执行monitorenter指令时会尝试获取对象锁，如果对象没有被锁定或者已经获得了锁，锁的计数器+1。此时其他竞争锁的线程则会进入等待队列中。

执行monitorexit指令时则会把计数器-1，当计数器值为0时，则锁释放，处于等待队列中的线程再继续竞争锁。

如果再深入到源码来说，synchronized实际上有两个队列waitSet和entryList。

  1. 当多个线程进入同步代码块时，首先进入entryList
  2. 有一个线程获取到monitor锁后，就赋值给当前线程，并且计数器+1
  3. 如果线程调用wait方法，将释放锁，当前线程置为null，计数器-1，同时进入waitSet等待被唤醒，调用notify或者notifyAll之后又会进入entryList竞争锁
  4. 如果线程执行完毕，同样释放锁，计数器-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的区别。

  1. 可中断加锁lockInterruptibly，synchronized可能会有死锁的问题，那么解决方案就是能响应中断。当前线程加锁时，如果其他线程调用当前线程的中断方法，则会抛出异常。
  2. 非阻塞加锁tryLock，调用后立刻返回，获取锁则返回true，否则返回false
  3. 支持超时加锁tryLock(long time, TimeUnit unit)，超时时间内获取锁返回true，否则返回false
  4. 支持公平和非公平锁，公平指的是获取锁按照请求锁的时间顺序决定，先到先得，非公平则是直接竞争锁，先到不一定先得
  5. 支持Condition

如果你看过阻塞队列的源码，那么你对 Condition 应该挺了解了，我们举个栗子来看看，我们需要实现：

  1. 如果队列满了，那么写入阻塞
  2. 如果队列空了，那么删除（取元素）阻塞

我们给阻塞队列提供一个 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[capacity];
            lock = new ReentrantLock();
            notEmpty = lock.newCondition();
            notFull = lock.newCondition();
        }

        public void put(E e) throws InterruptedException {
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly();
            try {
                while (count == items.length) {
                    notFull.await();
                }
                enqueue(e);
            } finally {
                lock.unlock();
            }
        }

        private void enqueue(E x) {
            final Object[] items = this.items;
            items[putIndex] = x;
            if (++putIndex == items.length){
                putIndex = 0;
            }
            count++;
            notEmpty.signal();
        }

        public E take() throws InterruptedException {
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly();
            try {
                while (count == 0) {
                    notEmpty.await();
                }
                return dequeue();
            } finally {
                lock.unlock();
            }
        }

        private E dequeue() {
            final Object[] items = this.items;
            E x = (E) items[takeIndex];
            items[takeIndex] = null;
            if (++takeIndex == items.length){
                takeIndex = 0;
            }
            count--;
            notFull.signal();
            return x;
        }
    }
}

读写锁

读写锁，也可以称作共享锁，区别于独占锁，共享锁则可以允许多个线程同时持有，如ReentrantReadWriteLock允许多线程并发读，要简单概括就是：读读不互斥，读写互斥，写写互斥。

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);
        }
    }
}

LockSupport

LockSupport是一个比较基础的工具类，基于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);
    }
}
//输出如下
主线程sleep3000
Thread-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;
    }
}
//输出
test
newPwd

累加器

累加器有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[i]) != 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);
    }
}

工具类&容器类

这里要说到一些我们在平时开发中经常使用到的一些类以及他们的实现原理。

工具类&容器类

CountDownLatch

CountDownLatch适用于在多线程的场景需要等待所有子线程全部执行完毕之后再做操作的场景。

假设现在我们有一个业务场景，我们需要调用多个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变量维持倒数状态，多线程共享变量可见。

  1. CountDownLatch通过构造函数初始化传入参数实际为AQS的state变量赋值，维持计数器倒数状态
  2. 当主线程调用await()方法时，当前线程会被阻塞，当state不为0时进入AQS阻塞队列等待。
  3. 其他线程调用countDown()时，通过CAS修改state值-1，当state值为0的时候，唤醒所有调用await()方法阻塞的线程

CyclicBarrier

CyclicBarrier叫做回环屏障，它的作用是让一组线程全部达到一个状态之后再全部同时执行，他和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，这就是复用的原理。

  1. 当子线程调用await()方法时，获取独占锁ReentrantLock，同时对count递减，进入阻塞队列，然后释放锁
  2. 当第一个线程被阻塞同时释放锁之后，其他子线程竞争获取锁，操作同1
  3. 直到最后count为0，执行CyclicBarrier构造函数中的任务，执行完毕之后子线程继续向下执行，计数重置，开始下一轮循环

Semaphore

Semaphore叫做信号量，和前面两个不同的是，他的计数器是递增的，信号量这玩意儿在限流中就经常使用到。

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实现的，同时获取信号量有公平和非公平两种策略，通过构造函数的传参可以修改，默认则是非公平的策略。

  1. 先说非公平的策略，主线程调用acquire()方法时，用当前信号量值-需要获取的值，如果小于0，说明还没有达到信号量的要求值，则会进入AQS的阻塞队列，大于0则通过CAS设置当前信号量为剩余值，同时返回剩余值。而对于公平策略来说，如果当前有其他线程在等待获取资源，那么自己就会进入AQS阻塞队列排队。
  2. 子线程调用release()给当前信号量值计数器+1(增加的值数量由传参决定)，同时不停的尝试唤醒因为调用acquire()进入阻塞的线程

Exchanger

Exchanger用于两个线程之间交换数据，如果两个线程都到达同步点，这两个线程可以互相交换他们的数据。

举个栗子，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获取到数据=excelA
A获取到数据=excelB

A写完之后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[len] = e;
        setArray(newElements);
        return true;
    } finally {
        lock.unlock();
    }
}

再看看 get 方法，也非常简单，直接获取数组当前索引的值，这里需要注意的是，读数据是没有加锁的，所以会有一致性的问题，它并不能保证读到的一定是最新的数据。

public E get(int index) {
    return get(getArray(), index);
}

private E get(Object[] a, int index) {
	return (E) a[index];
}

final Object[] getArray() {
	return array;
}

至于 CopyOnWriteArraySet ，他就是基于 CopyOnWriteArrayList 实现的，这里我们不再赘述。

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/Join

Fork/Join 是一个并行执行任务的框架，利用的分而治之的思想。

Fork 是把一个大的任务拆分成若干个小任务并行执行，Join 则是合并拆分的子任务的结果集，最终计算出大任务的结果。

所以整个 Fork/Join 的流程可以认为就是两步：

  1. Fork 拆分任务，直到拆分到最小粒度不可拆分为止
  2. 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.Contended
static 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; // 当前偷到的任务
}

那如果工作线程自己队列的做完了怎么办？只能傻傻地等待吗？并不是，这时候有一个叫做工作窃取的机制，所以他就会去其他线程的队列里偷一个任务来执行。

为了避免偷任务线程和自己的线程产生竞争，所以自己的工作线程是从队列头部获取任务执行，而偷任务线程则从队列尾部偷任务。

工作窃取

Executor

Executor是并发编程中重要的一环，任务创建后提交到Executor执行并最终返回结果。

Executor

任务

线程两种创建方式：Runnable和Callable。

Runnable是最初创建线程的方式，在JDK1.1的版本就已经存在，Callable则在JDK1.5版本之后加入，他们的主要区别在于Callable可以返回任务的执行结果。

任务执行

任务的执行主要靠Executor，ExecutorService继承自Executor，ThreadPoolExecutor和ScheduledThreadPoolExecutor分别实现了ExecutorService。

那说到线程池之前，我们肯定要提及到线程池的几个核心参数和原理，这个之前的文章也写到过，属于基础中的基础部分。

首先线程池有几个核心的参数概念：

  1. 最大线程数maximumPoolSize

  2. 核心线程数corePoolSize

  3. 活跃时间keepAliveTime

  4. 阻塞队列workQueue

  5. 拒绝策略RejectedExecutionHandler

当提交一个新任务到线程池时，具体的执行流程如下：

  1. 当我们提交任务，线程池会根据corePoolSize大小创建若干任务数量线程执行任务
  2. 当任务的数量超过corePoolSize数量，后续的任务将会进入阻塞队列阻塞排队
  3. 当阻塞队列也满了之后，那么将会继续创建(maximumPoolSize-corePoolSize)个数量的线程来执行任务，如果任务处理完成，maximumPoolSize-corePoolSize额外创建的线程等待keepAliveTime之后被自动销毁
  4. 如果达到maximumPoolSize，阻塞队列还是满的状态，那么将根据不同的拒绝策略对应处理

拒绝策略主要有四种：

  1. AbortPolicy：直接丢弃任务，抛出异常，这是默认策略
  2. CallerRunsPolicy：使用调用者所在的线程来处理任务
  3. DiscardOldestPolicy：丢弃等待队列中最老的任务，并执行当前任务
  4. DiscardPolicy：直接丢弃任务，也不抛出异常

ThreadPoolExecutor

通常为了快捷我们会用Executors工具类提供的创建线程池的方法快速地创建一个线程池出来，主要有几个方法，但是一般我们不推荐这样使用，非常容易导致出现问题，生产环境中我们一般推荐自己实现，参数自己定义，而不要使用这些方法。

创建

//创建固定线程数大小的线程池，核心线程数=最大线程数，阻塞队列长度=Integer.MAX_VALUE
public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
}
//创建只有一个线程的线程池，阻塞队列长度=Integer.MAX_VALUE
public 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();

ScheduledThreadPoolExecutor

ScheduledThreadPoolExecutor 继承于 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:05
scheduleWithFixedDelay=01:17:05
scheduleAtFixedRate=01:17:07
scheduleWithFixedDelay=01:17:08
scheduleAtFixedRate=01:17:09
scheduleAtFixedRate=01:17:11
scheduleWithFixedDelay=01:17:11
scheduleAtFixedRate=01:17:13
scheduleWithFixedDelay=01:17:14
scheduleAtFixedRate=01:17:15
scheduleAtFixedRate=01:17:17
scheduleWithFixedDelay=01:17:17
scheduleAtFixedRate=01:17:19
scheduleWithFixedDelay=01:17:20
scheduleAtFixedRate=01:17:21

我们把任务耗时调整到超过时间间隔，比如改成睡眠3秒，观察输出结果。

scheduleAtFixedRate 由于任务执行的耗时比时间间隔长，按照规定上次任务执行结束之后立刻执行，所以变成以3秒的时间间隔执行。

scheduleWithFixedDelay 因为任务耗时用了3秒，导致后面的时间间隔都成了5秒。

scheduleWithFixedDelay=01:46:21
scheduleAtFixedRate=01:46:21
scheduleAtFixedRate=01:46:24
scheduleWithFixedDelay=01:46:26
scheduleAtFixedRate=01:46:27
scheduleAtFixedRate=01:46:30
scheduleWithFixedDelay=01:46:31
scheduleAtFixedRate=01:46:33
scheduleWithFixedDelay=01:46:36
scheduleAtFixedRate=01:46:36

OK，最后来说说实现原理：

  1. 首先我们通过调用 schedule 的几个方法，把任务添加到 ScheduledThreadPoolExecutor 去执行
  2. 接收到任务之后，会通过请求参数的延迟时间计算出真正需要执行任务的时间，然后把任务封装成 RunnableScheduledFuture
  3. 然后把封装之后的任务添加到延迟队列中，任务 ScheduledFutureTask 实现了 comparable 接口，把时间越小的任务放在队列头，如果时间一样，则会通过 sequenceNumber 去比较，也就是执行时间相同，先提交的先执行
  4. 最后线程池会从延迟队列中去获取任务执行，如果是一次性的任务，执行之后删除队列中的任务，如果是重复执行的，则再次计算时间，然后把任务添加到延迟队列中

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 Ms
task-4 sleep 1186 Ms
task-3 sleep 6182 Ms
task-9 sleep 7411 Ms
task-0 sleep 1882 Ms
task-1 sleep 8274 Ms
task-2 sleep 4789 Ms
task-5 sleep 8894 Ms
task-8 sleep 7211 Ms
task-6 sleep 5959 Ms
get 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 Ms
task-1 sleep 4114 Ms
task-2 sleep 4865 Ms
task-5 sleep 1592 Ms
task-3 sleep 6190 Ms
task-7 sleep 2482 Ms
task-8 sleep 9405 Ms
task-9 sleep 8798 Ms
task-6 sleep 2040 Ms
task-0 sleep 2111 Ms
get 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[ForkJoinPool.commonPool-worker-9,5,main]工作完成
Thread[ForkJoinPool.commonPool-worker-9,5,main]thenCompose=>supplyAsync
Thread[main,5,main]thenApply=>supplyAsync
Thread[main,5,main]thenAccept=>thenApply
Thread[main,5,main]thenRun
supplyAsync
null
Thread[ForkJoinPool.commonPool-worker-2,5,main]thenCompose工作完成
thenCompose

AND 聚合

这个意思是下一个任务执行必须等前两个任务完成可以。

主要包含 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[ForkJoinPool.commonPool-worker-9,5,main]A工作完成
A
Thread[ForkJoinPool.commonPool-worker-2,5,main]B工作完成
B
Thread[ForkJoinPool.commonPool-worker-2,5,main]thenCombine=>A,B
Thread[ForkJoinPool.commonPool-worker-2,5,main]thenAcceptBoth=>thenCombine,B
Thread[ForkJoinPool.commonPool-worker-2,5,main]runAfterBoth
null

Or 聚合

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[ForkJoinPool.commonPool-worker-9,5,main]exceptionally exception=>java.lang.RuntimeException: java.lang.ArithmeticException: / by zero
Thread[main,5,main]whenComplete exception=>java.lang.RuntimeException: java.lang.ArithmeticException: / by zero
Thread[ForkJoinPool.commonPool-worker-9,5,main]handle exception=>java.lang.RuntimeException: java.lang.ArithmeticException: / by zero
Exception in thread "main" java.util.concurrent.ExecutionException: java.lang.RuntimeException: java.lang.ArithmeticException: / by zero
	at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357)
	at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1908)
	at com.example.demo.CompletableFutureTest3.main(CompletableFutureTest3.java:31)
Caused by: java.lang.RuntimeException: java.lang.ArithmeticException: / by zero
	at com.example.demo.CompletableFutureTest3.lambda$main$0(CompletableFutureTest3.java:13)
	at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1604)
	at java.util.concurrent.CompletableFuture$AsyncSupply.exec(CompletableFuture.java:1596)
	at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
	at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1067)
	at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1703)
	at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:172)
Caused by: java.lang.ArithmeticException: / by zero
	at com.example.demo.CompletableFutureTest3.lambda$main$0(CompletableFutureTest3.java:10)
	... 6 more

阻塞队列

并发编程中，队列是其中不可缺少的一环，其实前面在说到线程池的时候，就已经提及到了阻塞队列了，这里我们要一起看看 JUC 包下提供的这些队列。

阻塞队列

阻塞队列中的阻塞包含两层意思：

  1. 插入的时候，如果阻塞队列满，插入元素阻塞
  2. 删除/查询的时候，如果阻塞队列空，删除/查询元素阻塞

下面列出队列的一些插入和删除元素的方法，一个个来说：

add：向队列尾部插入元素，插入成功返回 true，队列满则抛出IllegalStateException("Queue full")异常

offer：向队列尾部插入元素，队列满返回 false，否则返回 true，带超时的则是会阻塞，达到超时时间后返回

put：向队列尾部插入元素，队列满会一直阻塞

remove：删除队列头部元素，删除成功返回 true，队列空则抛出NoSuchElementException异常

poll：删除队列头部元素，删除成功返回队列头部元素，队列空返回null，带超时的则是会阻塞，达到超时时间后返回

take：删除队列头部元素，队列空会一直阻塞

element：查询队列头部元素，并且返回，队列空则抛出NoSuchElementException异常

peek：查询队列头部元素，并且返回

ArrayBlockingQueue

ArrayBlockingQueue 从名字就知道，基于数组实现的有界阻塞队列，基于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 的部分。

LinkedBlockingQueue

LinkedBlockingQueue 基于链表实现的有界阻塞队列。

使用无参构造函数则链表长度为 Integer.MAX_VALUE，另外两个构造函数和 ArrayBlockingQueue 差不多。

public LinkedBlockingQueue();
public LinkedBlockingQueue(int capacity);
public LinkedBlockingQueue(Collection<? extends E> c);

我们可以看看 put 和 take 的源码。

  1. 首先加锁中断
  2. 然后判断如果达到了队列的最大长度，那么就阻塞等待，否则就把元素插入到队列的尾部
  3. 注意这里和 ArrayBlockingQueue 有个区别，这里再次做了一次判断，如果队列没满，唤醒因为 put 阻塞的线程，为什么要做判断，因为他们不是一把锁
  4. 最后的逻辑是一样的，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的逻辑也是非常类似啊。

  1. 加锁中断
  2. 判断队列是不是空了，空了的话就阻塞等待，否则就从队列移除一个元素
  3. 然后再次做一次判断，队列要是不空，就唤醒阻塞的线程
  4. 最后唤醒 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();
  	}
}

PriorityBlockingQueue

PriorityBlockingQueue 是支持优先级的无界阻塞队列，默认排序按照自然排序升序排列。

几个构造函数，无参构造函数初始容量为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 的实现。

  1. 首先还是加锁，然后看当前元素个数是否达到了数组的上限，到了就调用 tryGrow 去扩容。
  2. 看是否实现了 Comparator 接口，是的话就用 Comparator 去排序，否则就用 Comparable 去比较，如果两个都没有，会报错
  3. 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;
}

这里，我们要继续关注一下这个扩容的逻辑，到底是怎么处理的？代码不长，但是看着很方的样子。

  1. 首先，先释放锁，因为下面用 CAS 处理，估计怕扩容时间太长阻塞的线程太多
  2. 然后 CAS 修改 allocationSpinLock 为1
  3. CAS 成功的话，进行扩容的逻辑，如果长度小于64就扩容一倍，否则扩容一半
  4. 之前我们说他无界，其实不太对，这里就判断是否超过了最大长度，MAX_ARRAY_SIZE = Integer.MAX_VALUE – 8，判断一下有可能会抛出内存溢出异常
  5. 然后创建一个新的对象数组，并且 allocationSpinLock 重新恢复为0
  6. 执行了一次 Thread.yield()，让出 CPU，因为有可能其他线程正在扩容，让大家争抢一下
  7. 最后确保新的对象数组创建成功了，也就是扩容是没有问题的，再次加锁，数组拷贝，结束
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[newCap];
            } finally {
                allocationSpinLock = 0;
            }
        }
        if (newArray == null) // back off if another thread is allocating
            Thread.yield();
        lock.lock();
        if (newArray != null && queue == array) {
            queue = newArray;
            System.arraycopy(array, 0, newArray, 0, oldCap);
        }
}

take 的逻辑基本一样，最多有个排序的逻辑在里面，就不再多说了。

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;
}

DelayQueue

DelayQueue 是支持延时的无界阻塞队列，这个在我们聊 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，否则我觉得闹不明白。

  1. 来第一步加锁，如果头结点是空的，也就是队列是空的话，阻塞，没啥好说的
  2. 反之队列有东西，我们就要去取了嘛，但是这里要看对象自己实现的 getDelay 方法获得延迟的时间，如果延迟的时间小于0，那说明到时间了，可以执行了，poll 返回
  3. 第一次，leader 线程肯定是空的，线程阻塞 delay 的时间之后才开始执行，完全没毛病，然后 leader 重新 置为 null
  4. 当 leader 不是 null 的时候，说明其他线程在操作了，所以阻塞等待唤醒
  5. 最后，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;
}

LinkedBlockingDeque

LinkedBlockingDeque 是链表组成的双向阻塞队列，它支持从队列的头尾进行进行插入和删除元素。

构造函数有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();
  	}
}

结尾

本次长篇内容参考书籍和文档

  1. Java 并发编程的艺术
  2. Java 并发编程之美
  3. Java 并发编程实战
  4. Java 8实战
  5. 极客时间：Java 并发编程实战

OK，本期内容到此结束，我是艾小仙，我们过两个月再见。

hmoban主题是根据ripro二开的主题，极致后台体验，无插件，集成会员系统
自学咖网 » 并发编程从入门到放弃系列开始和结束

wen 普通

分享到：

相关推荐

04python基础知识02

这一篇文章主要介绍函数、模块、运行python文件等知识。 函数 函数犹如小程序，可以用来执行特定的操作。Python本身提供了很多函数，我们称这些函数为内置函数。内置函数可以用于编写数值表达式 >>>...

Gateway网关匹配规则

Gateway匹配规则 重要概念 路由 route 断言 predicate 过滤器 fliter 一、时间匹配 在 after 时间之后的所有请求转发到 URI中的地址 gateway: discovery: loca...

Java数据类型

一、八大基本类型 1. 整型类 类型 占用存储空间 范围 byte 1字节=8bit -128~127 2^8 short 2字节 -2^15~2^15-1 int 4字节 -2^31~2^31-1(约21亿) long...

Spring Boot + MybatisX = 王炸！！

1.什么是MybatisX？ MybatisX 是一款基于 IDEA 的快速开发插件，方便在使用mybatis以及mybatis-plus开始时简化繁琐的重复操作，提高开发速率。 2.使用MybatisX的好处 节省大量...

HarmonyOS实战—实现相亲APP

简易版 相亲APP 实现如下效果： 如果喜欢就点击下面的“联系方式”，不喜欢就点击“下一个” 1. 布局实现 新建项目：MakeFriendsApplication 把以下九个girl图片复制到 media 文件夹下 有...