JDK1.8并发包之 -- CountDownLatch

Posted 2023-04-09

参考技术A 上篇 JDK1.8并发包之--Semaphore 发现一个方法，搞懂Jdk的类注释，就能快速了解该类的用法，于是从CountDownLatch的英文注释出发，GO！

一句话告诉我们CountDownLatch的目标是，等待一组线程执行完成。场景如下，10个线程并发执行一系列任务，肯定有的线程先跑完，那么先跑完的就等省下没跑完的线程执行完毕，再执行下面任务。

CountDownLatch初始化构造传递整形count参数，await方法直到count降为0才会立即返回，调用countDown方法使count减一，count等于0后，所有的等待线程被唤醒。这个过程中count不能被重置，如果要重置count，考虑使用CycliBarrier。

CountDownLatch有种用途，可以用来作为门栓或者门岗，所有的线程调用await方法阻塞，直到一个线程调用countDown减1，CountDownLatch初始化为N，保证一个线程等待N个线程完成动作后，或者执行N次后，再接着执行。

CountDownLatch有个非常有用的特性就是调用countDown减1不用等待减到0 。仅仅是避免其他线程都执行完了才能通过await方法。

说的两处使用CountDownLatch的场景，第一个CountDownLatch避免worker执行任务，直到driver准备好了，worker才开始执行。第二个CountDownLatch是信号量，让driver等待worker执行完毕。

还有另外一种用法是，把一个问题拆分成N个部分，每个部分都是一个线程任务，然后执行countDwon直到为0，后续的请求放入队列中缓存起来。当N个部分全部执行完成，协调线程就会通过await，进一步执行任务。另外，如果线程需要重复执行countDown，就要用CyclicBarrier。

总结，CountDownLatch有两种用法，第一、协调任务执行关卡，如第一个例子那样，Driver-Worker的关系，好比司机-搬运工，搬运工必须等司机把车开过来才能工作，司机把车开到后，必须等搬运工把所有东西都装完才能把车开到目的地。第二、分别大任务，如第二个例子，把问题分解成N个小份，再分配给N个线程执行，等N个线程执行完毕后再分配下一批任务。笔者用第二种方式同步数据，效率尤其快。

Java多线程--JDK并发包

Java多线程--JDK并发包（2）

线程池

在使用线程池后，创建线程变成了从线程池里获得空闲线程，关闭线程变成了将线程归坏给线程池。

JDK有一套Executor框架，大概包括Executor、ExecutorService、AbstractExeccutorService、ThreadPoolExecutor、Executors等成员，位于java.util.concurrent包下。它们之间的关系如下：

Executor是顶层的接口，ExecutorService接口继承了它，AbstrctExecutorService继承了ExecutorService，ThreadPoolExecutor继承了AbstrctExecutorService。如果用<——表示继承，<--表示实现接口，它们的关系可表示如下：

Executor(接口) <—— ExecutorService(接口) <-- AbstrctExecutorService(抽象类) <—— ThreadPoolExecutor(类)

Executors是单独的一个类，可以看成是“线程池工厂”，它有很多静态方法，比如：

• newFixedThreadPool(int nThread)
• newSingleThreadExecutor()
• newCachedThreadPool()
• newSingleThreadScheduledExecutor()
• newScheduledThreadPool(int corePoolSize)

newFixedThreadPool该方法返回一个固定线程数的线程池。当有新任务提交时，如果线程池中有空闲线程就立即执行，否则会进入任务队列中，等到有空闲线程了才能执行。

newSingleThreadExecutor，该方法返回只有一个线程的线程池，处理策略和上面一样。实际上就是上面的参数指定为1而已。

newCachedThreadPool该方法返回一个可根据实际情况调整线程数的线程池，任务提交后，如果有空闲线程可以复用，则优先复用。若线程池中的线程全部在工作，而此时有新任务，则会创建新的线程来处理任务，所有线程执行完后会将线程归还给线程池。

newScheduledThreadPool返回一个ScheduledExecutorService对象，可以有计划地执行任务，比如在某个延时之后开始执行，或者周期性地执行某个任务。可以指定线程数量。

newSingleThreadScheduledExecutor实现了和上面一样的功能，不过线程池的大小为1。

ScheduledExecutorService有三个方法可以有计划地执行任务。如：

• schedule(Runnable command, long delay, TimeUnit unit);该方法可以在给定的延时后，执行一个任务；
• scheduleAtFixedRate(Runnable command,long initialDelay,long period,TimeUnit unit);该方法以任务开始执行的时间为initialDelay，加上周期period，就是下一个任务开始执行的时间，以此类推，因此这个方法任务调度的频率是一定的；
• scheduleWithFixedDelay(Runnable command,long initialDelay,long delay,TimeUnit unit);该方法表示每执行完一个任务，延迟delay的时间后，开始执行下一个任务，initialDelay还是表示任务开始的初始时延，上一个任务结束的时间点与下一个任务开始的时间点之差是固定的，固定为delay。

即使单个任务的执行时间超过调度周期，scheduleAtFixedRate也不会让多个任务堆叠，比如任务执行需要8s，而调度周期是2s，调度第二个任务时，第一个还没执行完，因此为了避免任务堆叠，此时调度周期会变成8s；而采用scheduleWithFixedDelay，两个任务之间的实际间隔会变成10s，8s的执行+2s的delay。

线程池的内部实现

• newFixedThreadPool(int nThread)
• newSingleThreadExecutor()
• newCachedThreadPool()

这三个内部都是通过返回ThreadPoolExecutor产生线程池的。所以我们来重点关注它的构造方法。

public ThreadPoolExecutor(
    int corePoolSize,
    int maximumPoolSize,
    long keepAliveTime,
    TimeUnit unit,
    BlockingQueue<Runnable> workQueue,
    ThreadFactory threadFactory,
    RejectedExecutionHandler handler)

• corePoolSize表示线程池中的线程数；
• maximumPoolSize表示线程池中的最大线程数；
• keepAliveTime表示当线程数超过corePoolSize时，多余的空闲线程的存活时间；
• unit是keepAliveTime的单位
• workQueue任务队列，保存那些已经提交但还没有开始执行的任务（在等待空闲线程）；
• threadFactory，线程工厂，可自定义，一般默认；
• handler拒绝策略，当任务多得来不及处理时，如何拒绝任务。

workQueue是一个BlockingQueue接口的对象，存放的是Runnable对象。根据功能的不同，ThreadPoolExecutor中可以使用以下几种BlockingQueue

• 直接提交的队列：对应SynchronousQueue对象，它没有容量，每一个插入都要等待一个相应的删除操作；每一个删除操作都要等待对应的插入操作。使用该对象，提交的任务不会被真实保存，而总是将任务交给线程执行。如果没有空闲线程就创建新线程，如果线程数已经达到最大值，就执行拒绝策略。
• 有界的任务队列：使用ArrayBlockingQueue实现。当有任务提交时，判断线程池中当前的实际线程数，如果小于corePoolSize，则优先创建新线程；若大于corePoolSize，就将任务加入到等待队列中；若此时等待队列也满，创建新线程；若实际线程已经达到maxPoolSize，就开始执行拒绝策略。可以看出有界的任务队列只有在任务队列满时，才会创建新线程，通常情况下实际线程数可以稳定在corePoolSize。
• 无界的任务队列：使用LinkedBlockingQueue实现。和上面ArrayBlockingQueue相比，区别在于，任务队列没有大小限制，当实际线程数超过corePoolSize时，直接进入任务队列。
• 优先任务队列：使用PriorityBlockingQueue实现。前面的几种都是按照先进先出的顺序来处理任务，而该对象实现的任务队列可根据任务自身的优先级顺序执行。

newFixedThreadPool因为它的corePoolSize和maxPoolSize大小一样，固定大小的线程不存在当实际线程数超过corePoolSize时要新增线程的可能，所以它使用了LinkedBlockingQueue，当有新任务且实际线程数已经达到最大时，会直接进入等待队列。

newSingleThreadExecutor是newFixedThreadPool的一种特殊情况，即取corePoolSize和maxPoolSize都为1

而newCachedThreadPool的corePoolSize为0，maxPoolSize为Integer.MAX_VALUE，任务队列使用SynchronousQueue直接提交，新任务提交后，若有空闲线程就直接用，若没有就进入等待队列——但是这是个直接提交的队列，所有会新增线程执行该任务！由于corePoolSize为0，所以任务执行完毕后60s（构造函数指定）就会被回收。

拒绝策略

当实际线程数超过maxPoolSize时，该采取什么样的策略？

• AbortPolicy：丢弃任务并抛出异常；
• CallerRunPolicy：该任务被线程池拒绝，由调用execute方法的线程执行该任务；
• DiscardOldestPolicy：丢弃最老的一个，也就是马上要执行的一个任务；
• DiscardPolicy：默默丢弃被拒绝的任务，体现在代码中就是什么也不做。

下面看看CallerRunPolicy怎么拒绝的

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {
            r.run();
        }
    }

DiscardOldestPolicy是这样做的

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {
            e.getQueue().poll(); // 最老的一个请求在队列头部
            e.execute(r);
        }
    }

线程的创建--线程工厂

ThreadFactory只有一个方法Thread newThread(Runnable r);，线程池中的线程就是由它创建的。

Fork/Join框架

fork也就是分支、分叉的意思，可以将大任务分解成小任务；join表示等待的意思，必须等待fork后的小任务执行完毕，得到执行后的部分结果，才能将部分结果合并成最终结果。

比如计算1到10000的和，就可以分成10个分支，每个分支计算一千个数的和，得到一个部分和，等待这10个部分和的结果都计算完毕，最后将其全部合并，得到最终的结果。

通常一个物理线程需要处理多个逻辑任务，所以每一个线程都有一个任务队列。若线程A的任务都执行完了，B还有很多任务没执行，此时A会“帮助”B执行它的任务，A帮助B执行B的任务时，从队列的尾部拿数据；而B自己执行任务时从队列头部拿数据，这就像是两个指针一个往左移动一个往右移动，避免了A、B之间对数据的竞争。

JDK中有ForkJoinPool，该接口有个方法public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task)

ForkJoinTask支持fork()和join()方法，它有两个重要的子类，没有返回值的RecursiveAction和有返回值的RecursiveTask，它们都有个方法compute()，在这个方法中进行主要的计算。对于RecursiveAction来说签名是void，而对于RecursiveTask来说有返回值所以签名是<T>

JDK并发容器

• ConcurrentHashMap：高效的并发HashMap，可看作线程安全的HashMap；
• CopyOnWriteArrayList：读-读，读-写不会阻塞，只有在写-写下会进行同步。在读多写少的场合，性能很好；
• ConcurrentLinkedQueue：高效的并发队列，链表实现，使用了CAS操作（Compare and Swap），可看作线程安全的LinkedList；
• BlockingQueue：接口，实现了Queue；数组实现的ArrayBlockingQueue和链表实现的LinkedBlockingQueue实现了这个接口。
• ConcurrentSkipListMap：使用跳表的数据结构实现的Map。

CopyOnWriteArrayList原理

CopyOnWriteArrayList的原理主要是：读的时候正常读，写-写需要同步，所以在写之前要使用Lock，然后为了读-写不阻塞，CopyOnWriteArrayList在写入操作时，先将原数组复制一份，然后在新数组末尾追加要添加的值，添加成功后再用新数组覆盖旧数组。

JDK中的该类的add方法是这样实现的：

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

而数组的定义是这样的：

 private transient volatile Object[] array;

注意有volatile关键字，说明当写数据的线程修改数组后，其他读取线程能立即“察觉”到。

BlockingQueue原理

BlockingQueue可以在并发环境下高效传输数据，本质上还是一个队列，数据从队列尾部入，从队列头部出。队列都有的offer()和pull()就不说了，没什么特别的。BlockingQueue还有put()和take()方法，正是这两个方法实现了阻塞。

以ArrayBlockingQueue来说：当队列为空时，take()方法会等待，直到队列不为空；当队列满时，put()方法会等待，直到队列有空闲位置。这是怎么实现的呢？来看代码

/** Main lock guarding all access */
final ReentrantLock lock;

/** Condition for waiting takes */
private final Condition notEmpty;

/** Condition for waiting puts */
private final Condition notFull;

首先读和写都用的同一个锁lock，因此任何时候读和写只能有一个在执行。然后是条件notNull，等待非满，以便put；notEmpty等待非空，以便take。

public void put(E e) throws InterruptedException {
    checkNotNull(e);
    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上等待的线程（通知其他线程可以进行take啦）
    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;
    @SuppressWarnings("unchecked")
    E x = (E) items[takeIndex];
    items[takeIndex] = null;
    if (++takeIndex == items.length)
        takeIndex = 0;
    count--;
    if (itrs != null)
        itrs.elementDequeued();
    // 关键！有元素出列了，等待在notFull上的线程可以被唤醒，可以进行put操作了
    notFull.signal();
    return x;
}

LinkedBlockingQueue和ArrayBlockingQueue原理大同小异，不过LinkedBlockingQueue读和写分别用一把锁，因此读和写可以同时进行。

/** Lock held by take, poll, etc */
private final ReentrantLock takeLock = new ReentrantLock();

/** Wait queue for waiting takes */
private final Condition notEmpty = takeLock.newCondition();

/** Lock held by put, offer, etc */
private final ReentrantLock putLock = new ReentrantLock();

/** Wait queue for waiting puts */
private final Condition notFull = putLock.newCondition();

跳表

ConcurrentSkipMap使用跳表实现。是一种可以进行快速查找的数据结构，时间复杂度是$O(lg n)$

跳表形象点说像个“直角三角形一样的金字塔”，每一层都是一条链表，最底层的链表包含了Map中的所有数据，每上一层都是下面一层的子集，越到上面结点越少。层与层之间通过值相同的元素链接起来，因此结点除了有指向本层的下一个结点的right，还有指向下层中具有相同值的元素的down（实际上通过数据结构Index表示）。另外，跳表中所有链表的元素都是排序的。

查找时，先从顶层开始查找，如果找到就结束了；否则当发现查找的值大于当前层的最大值（链表末尾），就会“跳到”下一层链表接着向前查找，查找朝着下面和右面两个方向进行，有点像“下台阶”...

---

by @sunhaiyu

2108.4.26