Java-SDK-并发包全面总结 | 乐趣区

一、Lock 和 Condition

Java 并发包中的 Lock 和 Condition 主要解决的是线程的互斥和同步问题，这两者的配合使用，相当于 synchronized、wait()、notify() 的使用。

1. Lock 的优势

比起传统的 synchronized 关键字，Lock 最大的不同（或者说优势）在于：

阻塞的线程能够响应中断，这样能够有机会释放自己持有的锁，避免死锁
支持超时，如果线程在一定时间内未获取到锁，不是进入阻塞状态，而是抛出异常
非阻塞的获取锁，如果未获取到锁，不进入阻塞状态，而是直接返回

三种情况分别对应 Lock 的三个方法：void lockInterruptibly()，boolean tryLock(long time, TimeUnit unit)，boolean tryLock()。

Lock 最常用的一个实现类是 ReentrantLock，代表可重入锁，意思是可以反复获取同一把锁。
除此之外，Lock 的构造方法可以传入一个 boolean 值，表示是否是公平锁。

2. Lock 和 Condition 的使用

前面实现的简单的阻塞队列就是使用 Lock 和 Condition ，现在其含义已经非常明确了：

public class BlockingQueue<T> {    private int capacity;    private int size;    //定义锁和条件    private final Lock lock = new ReentrantLock();    private final Condition notFull = lock.newCondition();    private final Condition notEmpty = lock.newCondition();    /**     * 入队列     */    public void enqueue(T data){        lock.lock();        try {            //如果队列满了，需要等待，直到队列不满            while (size >= capacity){                notFull.await();            }            //入队代码，省略            //入队之后，通知队列已经不为空了            notEmpty.signal();        } catch (InterruptedException e) {            e.printStackTrace();        } finally {            //在finally块中释放锁，避免死锁            lock.unlock();        }    }    /**     * 出队列     */    public T dequeue(){        lock.lock();        try {            //如果队列为空，需要等待，直到队列不为空            while (size <= 0){                notEmpty.await();            }            //出队代码，省略            //出队列之后，通知队列已经不满了            notFull.signal();        } catch (InterruptedException e) {            e.printStackTrace();        } finally {            lock.unlock();        }        //实际应该返回出队数据        return null;    }}

可以看到，Lock 需要手动的加锁和解锁，并且解锁操作是放在 finally 块中的，这是一种编程范式，尽量遵守。

二、ReadWriteLock

ReadWriteLock 表示读写锁，适用于读多写少的情况，读写锁一般有几个特征：

读锁与读锁之间不互斥，即允许多个线程同时读变量。
写锁与读锁之间互斥，一个线程在写时，不允许读操作。
写锁与写锁之间互斥，只允许一个线程写操作。

读写锁减小了锁的粒度，在读多写少的场景下，对性能的提升较为明显。ReadWriteLock 的简单使用示例如下：

public class ReadWriteLockTest {    private final ReadWriteLock lock = new ReentrantReadWriteLock();    private final Lock readLock =lock.readLock();    private final Lock writeLock =lock.writeLock();    private int value;    //加写锁    private void addValue(){        writeLock.lock();        try {            value += 1;        }        finally {            writeLock.unlock();        }    }    //加读锁    private int getValue(){        readLock.lock();        try {            return value;        }        finally {            readLock.unlock();        }    }}

读写锁的升级与降级

Java 中不允许锁的升级，即加写锁时必须释放读锁。

但是允许锁的降级，即加读锁时，可以不释放写锁，最后读锁和写锁一起释放。

三、StampedLock

1. StampedLock 的使用及特点

StampedLock 是 Java 1.8 版本中提供的锁，主要支持三种锁模式：写锁、悲观读锁、乐观读。

其中写锁和悲观读锁跟 ReadWriteLock 中的写锁和读锁的概念类似。StampedLock 在使用的时候不一样，加锁的时候会返回一个参数，解锁的时候需要传入这个参数，示例如下：

public class StampedLockTest {    private final StampedLock lock = new StampedLock();    private int value;    private void addValue(){        long stamp = lock.writeLock();        try {            value += 1;        }        finally {            lock.unlockWrite(stamp);        }    }}

StampedLock 最主要的特点是支持“乐观读”，即当进行读操作的时候，并不是所有的写操作都被阻塞，允许一个线程获取写锁。乐观读的使用示例如下：

public class StampedLockTest {    private final StampedLock lock = new StampedLock();    private int value;    private void getValue(){        //乐观读，读入变量        long stamp = lock.tryOptimisticRead();        int a = value;        //如果验证失败        if (!lock.validate(stamp)){            //升级为悲观读锁，继续读入变量            stamp = lock.readLock();            try {                a = value;            }            finally {                lock.unlockRead(stamp);            }        }    }}

需要注意的是，这里使用 validate() 方法进行验证，如果乐观读失败，则升级为悲观读锁，继续获取变量。

2. StampedLock 的注意事项

StampedLock 不支持重入，即不可反复获取同一把锁。

在使用 StampedLock 的时候，不要调用中断操作。如果需要支持中断，可以调用 readLockInterruptibly 和 writeLockInterruptibly 方法。

四、Semaphore

Semaphore 表示信号量，初始化对象的时候，需要传一个参数，表示信号量的计数器值。acquire() 方法将计数器加 1，release() 方法减 1，这两个方法都能够保证原子性。

信号量的简单示例：

public class SemaphoreTest {    private final Semaphore semaphore = new Semaphore(1);    private int value;    public void addValue() {        try {            semaphore.acquire();            value += 1;        }        catch (InterruptedException e) {            e.printStackTrace();        }        finally {            semaphore.release();        }    }

程序中使用信号量实现了一个线程安全的方法，初始值设为了 1，当多个方法访问 addValue 方法的时候，由于 acquire 方法保证原子性，所以只能有一个线程将计数器减 1 并进入临界区，另一个线程等待。

一个线程执行完后，调用 release 方法，计数器加 1，另一个等待的线程被唤醒。

Semaphore 与 Lock 的一个不同点便是信号量允许多个线程同时进入临界区，例如将初始值设置的更大一些。例如下面这个例子：

public class SemaphoreTest {    //初始值 2，表示 2 个线程可同时进入临界区    private final Semaphore semaphore = new Semaphore(2);    public void test() {        try {            semaphore.acquire();            System.out.println("线程" + Thread.currentThread().getName() + " 进入临界区 : " + System.currentTimeMillis());            Thread.sleep(1000);        }        catch (InterruptedException e) {            e.printStackTrace();        }        finally {            semaphore.release();        }    }}

五、CountDownLatch

CountDownLatch 是一个线程同步的工具，主要实现一个线程等待多个线程的功能。在原始的 Thread 中，可以调用 join() 方法来等待线程执行完毕，而 CountDownLatch 则可以用在线程池中的线程等待。

下面是 CountDownLatch 的使用示例：

public class CountDownLatchTest {    //实际生产中不推荐使用这种创建线程的方式    private final ExecutorService threadPool = Executors.newFixedThreadPool(2);    public void test() throws InterruptedException {        CountDownLatch latch = new CountDownLatch(2);        threadPool.execute(() -> {            System.out.println("线程1执行完毕");            latch.countDown();        });        threadPool.execute(() -> {            System.out.println("线程2执行完毕");            latch.countDown();        });        latch.await();        System.out.println("两个线程都执行完毕");        threadPool.shutdown();    }}

CountDownLatch 的初始值为 2，线程执行完毕则调用 countDown 方法，计数器减 1。减到 0 的时候，会唤醒主线程继续执行。

六、CyclicBarrier

CyclicBarrier 也是一个线程同步工具类，主要实现多个线程之间的互相等待。

CyclicBarrier 有两个构造函数，可以传一个计数器的初始值，还可以加上一个 Runnable，表示计数器执行减到 0 的时候，需要执行的回调方法。

public class CyclicBarrierTest {    private final ExecutorService threadPool = Executors.newFixedThreadPool(2);    private final CyclicBarrier barrier = new CyclicBarrier(2, this::note);    public void print(){        threadPool.execute(() -> {            System.out.println("线程1执行完毕");            try {                barrier.await();            } catch (InterruptedException | BrokenBarrierException e) {                e.printStackTrace();            }        });        threadPool.execute(() -> {            System.out.println("线程2执行完毕");            try {                barrier.await();            } catch (InterruptedException | BrokenBarrierException e) {                e.printStackTrace();            }        });        threadPool.shutdown();    }    public void note(){        System.out.println("两个线程执行完毕");    }}

示例中设置 CyclicBarrier 的初始值为 2，线程执行完毕调用 await 方法，计数器减 1。print() 方法中的两个线程执行完后，计数器减到 0，就会调用 note 方法。

七、ThreadPoolExecutor

1. 线程池的工作原理

由于线程是一种重量级对象，频繁的创建和销毁比较消耗系统资源，因此线程池的优势就显现出来了。线程池可有降低资源消耗，因为不用频繁创建和销毁线程；提高响应速度，需要执行任务时，可直接使用线程池中的线程资源；还能够有效的管理、监控线程池中的线程。

Java 中的线程池的实现是一种很典型的生产者-消费者模式，使用线程的一方是生产者，主要提供需要执行的任务，线程池是消费者，消费生产者提供的任务。

下面这段代码能够帮助理解线程池的实现原理(仅用于帮助理解，实际执行结果有出入)：

public class ThreadPool {    //保存任务的阻塞队列    private BlockingQueue<Runnable> workQueue;    //保存工作线程的列表    private List<WorkThread> threadList = new ArrayList<>();    //构造方法    public ThreadPool(int poolSize, BlockingQueue<Runnable> workQueue) {        this.workQueue = workQueue;        //根据poolSize的数量创建工作线程，并执行线程        for (int i = 0; i < poolSize; i++) {            WorkThread thread = new WorkThread();            thread.start();            threadList.add(thread);        }    }    //执行任务的方法，主要是将任务添加到队列中    public void execute(Runnable task) {        try {            workQueue.put(task);        }        catch (InterruptedException e) {            e.printStackTrace();        }    }    //工作线程    class WorkThread extends Thread{        @Override        public void run() {            //循环取出任务执行            while (!workQueue.isEmpty()) {                try {                    Runnable task = workQueue.take();                    task.run();                } catch (InterruptedException e) {                    e.printStackTrace();                }            }        }    }}

上面的代码注释很详细了，主要是使用了一个阻塞队列，用来存储生产者的任务。然后在构造器中创建线程，并循环从队列中取出任务执行。

2. Java 中的线程池

Java 中提供了 Executors 这个类来快速创建线程池，简单使用示例如下：

Executors.newSingleThreadExecutor();//创建一个线程的线程池Executors.newFixedThreadPool(5);//创建固定数量线程Executors.newCachedThreadPool();//创建可调整数量的线程Executors.newScheduledThreadPool(5);//创建定时任务线程池

但是在《阿里巴巴Java开发手册》中，明确禁止使用 Executors 创建线程池(甚至也不建议使用 Thread 显式创建线程)，主要原因是 Executors 的默认方法都是使用的无界队列，在高负载的情况下，很容易导致 OOM(Out Of Memory)。

所以在 Java 中创建线程池的正确姿势是使用 ThreadPoolExecutor ，其构造函数有七个：

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

corePoolSize：线程池中最少的线程数
maximumPoolSize：线程池中创建的最大的线程数
keepAliveTime：表示线程池中线程的活跃时间，如果线程在这个活跃时间内没有执行任务，并且线程数量超过了 corePoolSize，那么线程池就会回收多余的线程。
TimeUnit：上一个参数的时间单位
workQueue：保存任务的队列，为了避免 OOM，建议使用有界队列
threadFactory：可选参数，不传的话就是默认值。也可以自己传一个实现了 ThreadFactory 接口的类，表示自定义线程，例如给线程指定名字，线程组等。
handler：可选参数。定义任务的拒绝策略，表示无空闲线程时，并且队列中的任务满了的，怎么拒绝新的任务。目前的拒绝策略有四种：
- AbortPolicy：默认的拒绝策略，抛出 RejectedExecutionException 异常
- CallerRunsPolicy：让提交任务的线程自己去执行这个任务
- DiscardOldestPolicy：丢弃最老的任务，及最先加入队列中的任务，并添加新的任务
- DiscardPolicy：直接丢弃任务，并且不会抛出任何异常

调用 ThreadPoolExecutor

线程池创建好了之后，就需要执行任务，ThreadPoolExecutor 提供了两个方法，一是 execute，二是 submit。execute 没有返回值，也就是说无法获取执行结果。使用示例如下：

public static void main(String[] args) {    BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>(5);    ThreadPoolExecutor threadPool = new ThreadPoolExecutor(5, 10, 60L, TimeUnit.SECONDS, queue);    threadPool.execute(() -> {        System.out.println("In this world");    });    threadPool.shutdown();}

而 submit 方法有一个 Future 接口的返回值，Future 接口有五个方法：

cancle：取消任务
isCancled：任务是否已取消
isDone：任务是否已执行完
get：获取任务执行结果
get(long timeout, TimeUnit unit)：支持超时获取任务执行结果

下面代码展示了取消任务的方法：

public static void main(String[] args) {        BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>(5);    ThreadPoolExecutor threadPool = new ThreadPoolExecutor(5, 10, 60L, TimeUnit.SECONDS, queue);    Future<?> future = threadPool.submit(() -> {        System.out.println("I am roseduan");        try {            Thread.sleep(5000);        } catch (InterruptedException e) {            e.printStackTrace();        }    });    future.cancel(false);    threadPool.shutdown();}

程序的本意是打印语句然后休眠 5 秒，但由于调用了 cancle 方法，因此程序直接结束，不会有任何输出。

八、FutureTask

FutureTask 也是一个支持获取任务执行结果的工具类，FutureTask 实现了 Runnable 和 Future 接口。

所以可以将 FutureTask 作为任务提交给 ThreadPoolExecutor 或者 Thread 执行，并且可以获取执行结果。简单的使用如下：

public static void main(String[] args) throws ExecutionException, InterruptedException {    //创建任务    FutureTask<String> task = new FutureTask<>(() -> "Java and " + "Python");    BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>(5);    ThreadPoolExecutor threadPool = new ThreadPoolExecutor(5, 10, 60L, TimeUnit.SECONDS, queue);        threadPool.execute(task);    //获取执行结果    System.out.println(task.get());    threadPool.shutdown();}

传给 Thread 作为参数的使用示例如下：

public static void main(String[] args) throws ExecutionException, InterruptedException {        FutureTask<Integer> task = new FutureTask<>(() -> 1 + 2);        Thread thread = new Thread(task);        thread.start();        System.out.println(task.get());//输出3    }

九、CompletableFuture

CompletableFuture 是一个异步编程的工具类，异步化能够最大化并行程序的执行，是多线程性能优化的基础。

1. 创建 CompletableFuture 对象

Completable 有四个静态方法，可以用来创建对象：

runAsync(Runnable runnable);//无返回值runAsync(Runnable runnable, Executor executor);//无返回值，可指定线程池supplyAsync(Supplier<U> supplier);//有返回值supplyAsync(Supplier<U> supplier, Executor executor);//有返回值，可指定线程池

可以看到，四个方法分为了是否有返回值，和是否自定义线程池。如果不自定义线程池，那么 CompletableFuture 会使用公共的线程池，默认创建 CPU 核数的数量的线程池，当有多个任务的时候，还是建议根据每个任务自定义线程池。

一个简单的使用示例如下，其中 task3 会等待两个任务都执行完毕：

public static void main(String[] args) {    CompletableFuture<Void> task1 = CompletableFuture.runAsync(() -> {        System.out.println("任务1执行完毕");    });    CompletableFuture<Void> task2 = CompletableFuture.runAsync(() -> {        try {            Thread.sleep(2000);        } catch (InterruptedException e) {            e.printStackTrace();        }        System.out.println("任务2执行完毕");    });    CompletableFuture<String> task3 = task1.thenCombine(task2, (__, res) -> "两个任务执行完毕");    System.out.println(task3.join());}

CompletableFuture 实现了 Future 接口，因此可以查看任务执行的情况，并且可以获取返回值。

2. CompletionStage 接口中的方法

CompletableFuture 还实现了 CompletionStage 接口。这个接口描述了任务之间的时序关系，分别有串行、并行、聚合三种关系。需要注意的是，并行本就是其所具有的特性，所以不再探讨了，并且聚合关系又分为了 AND 聚合关系和 OR 聚合关系。下面依次介绍串行、AND 聚合、OR 聚合这三种关系。

首先是串行关系，串行很简单，一个任务执行完后再执行另一个任务，例如下图：

描述串行关系的几个方法是：thenApply、thenAccept、thenRun、thenCompose。

thenApply 既支持接收参数，又能够支持返回值。

thenAccept 支持接收参数，但是不支持返回值。

thenRun 既不能接收参数，也不能有返回值。

CompletionStage 中的大部分方法都有带有 Async 后缀的方法，表示可能会使用其他的线程来执行主体中的内容，后面介绍的方法都类似这样，不再赘述。

简单的使用示例如下：

public static void main(String[] args) throws ExecutionException, InterruptedException {    CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {        try {            Thread.sleep(2000);        } catch (InterruptedException e) {            e.printStackTrace();        }        System.out.println("任务1执行完毕");        return "Task1";    }).thenApply((s) -> "接收到的参数 : " + s);;    System.out.println(future.get());}

其次是 AND 汇聚关系，典型的场景便是一个线程等待两个线程都执行完后再执行，例如下图：

描述 AND 聚合关系的有三个方法：thenCombine、thenAcceptBoth、runAfterBoth，其是否接收参数和支持返回值，和上面的三个方法对应。一个简单的使用示例如下：

public static void main(String[] args) {    CompletableFuture<String> task1 = CompletableFuture.supplyAsync(() -> {        try {            Thread.sleep(2000);        } catch (InterruptedException e) {            e.printStackTrace();        }        System.out.println("任务1执行完毕");        return "task1";    });    CompletableFuture<String> task2 = CompletableFuture.supplyAsync(() -> {        System.out.println("任务2执行完毕");        return "task2";    });    CompletableFuture<String> task3 = task1.thenCombine(task2, (r,s) -> r + " " + s);    System.out.println(task3.join());}

任务 1 休眠了 2 秒，任务 3 会等待前面两个任务执行完成之后再执行。

最后是 OR 聚合关系，表示线程等待其中一个线程满足条件之后，就可以继续执行了，不用等待全部的线程。

描述 OR 聚合关系的是 applyToEither、acceptEither、runAfterEither。使用示例和上面的类似，只需要将方法改一下就是了，这里不再赘述了。

3. 处理异常

在异步编程中，CompletionStage 接口还提供了几个可以处理异常的方法，和 try() catch() finally() 类似。

这几个方法分别是：

exceptionally：相当于 catch
whenComplete：相当于 finally
handle：相当于 finally ，支持返回值

使用示例如下：

public static void main(String[] args) {    CompletableFuture<Integer> task = CompletableFuture.supplyAsync(() -> {        String str = null;        return str.length();        //相当于catch    }).exceptionally((e) -> {        System.out.println("发生异常");        return 0;    });    //相当于 finally    task.whenComplete((s, r) -> {        System.out.println("执行结束");    });    System.out.println(task.join());}

十、CompletionService

CompletionService 是一个批量执行异步任务的工具类，先来看一个例子：

public static void main(String[] args) throws ExecutionException, InterruptedException {    StringBuffer sb = new StringBuffer();    ThreadPoolExecutor threadPool = new ThreadPoolExecutor(            5, 5,            10, TimeUnit.SECONDS,            new LinkedBlockingQueue<>(5));    Future<String> task1 = threadPool.submit(() -> {        Thread.sleep(2000);        return "Task1";    });    Future<String> task2 = threadPool.submit(() -> "Task2");    Future<String> task3 = threadPool.submit(() -> "Task3");        sb.append(task1.get());    sb.append(task2.get());    sb.append(task3.get());}

程序的意思是，依次执行三个任务，并将其结果存储到 StringBuffer 中，由于 task1 休眠了 2 秒，所以 sb 会在这里阻塞。

由于这三个任务之间没有关联，所以等待的消耗完全是没必要的，解决的办法便是利用一个阻塞队列，先执行完的任务将结果保存在队列中，sb 从队列中取出就行了。

CompletionService 实际上就是将线程池和阻塞队列的功能整合了起来，解决了类似上面的问题。CompletionService 的实现类是 ExecutorCompletionService，这个类有两个构造方法：

public ExecutorCompletionService(Executor executor) {}public ExecutorCompletionService(Executor executor,    BlockingQueue<Future<V>> completionQueue) {}

如果不传一个阻塞队列，则会使用默认的无界队列。

CompletionService 主要有这几个方法：

submit() 提交任务、take() 从阻塞队列中获取执行结果（如果队列为空，线程阻塞）、poll() 也是从队列中获取执行结果（如果队列为空，则返回 null），另外 poll 还支持超时获取。

使用 CompletionService 改造后的程序示例如下：

public static void main(String[] args) throws ExecutionException, InterruptedException {    StringBuffer sb = new StringBuffer();    ThreadPoolExecutor threadPool = new ThreadPoolExecutor(            5, 5,            10, TimeUnit.SECONDS,            new LinkedBlockingQueue<>(5));    CompletionService<String> service = new ExecutorCompletionService<>(threadPool);    service.submit(() -> {        Thread.sleep(2000);        return "Task1";    });    service.submit(() -> "Task2");    service.submit(() -> "Task3");    System.out.println(sb.append(service.take().get()).toString());    System.out.println(sb.append(service.take().get()).toString());    System.out.println(sb.append(service.take().get()).toString());}

十一、Fork/Join

1. Fork/Join 使用

Fork/Join 是一个处理分治任务的计算框架，所谓分治，即分而治之，将一个任务分解成子任务，求解子任务，然后将子任务的结果合并，就得到了最后的结果。分治思想的应用十分的广泛，例如常见的快速排序、归并排序，还有流行的大数据计算框架 MapReduce，都应用了分治思想。

Java 中，Fork 对应的是任务分解，Join 则表示子任务的结果合并。

Fork/Join 主要包含两个主要的实现类：

一是线程池 ForkJoinPool，默认会创建 CPU核数数量的线程
二是 ForkJoinTask，这是一个抽象类，主要的方法有 fork() 和 join()，前者表示执行子任务，后者表示阻塞等待子任务的执行结果。ForkJoinTask 还有两个子类：
- RecursiveTask
- RecursiveAction

这两个类也是抽象的，我们需要自定义并继承这个类，并覆盖其 compute 方法。其中 RecursiveTask 有返回值，而 RecursiveAction 没有返回值。

下面是一个使用 ForkJoin 的示例，实现了 n 的阶乘，注释写得比较详细。

public class ForkJoinTest {    public static void main(String[] args) {        //创建线程池        ForkJoinPool forkJoinPool = new ForkJoinPool(4);        //创建任务        Factorial task = new Factorial(6);        //invoke 方法执行任务（还可以使用 execute、submit），得到执行的结果        Integer res = forkJoinPool.invoke(task);        System.out.println(res);    }    static class Factorial extends RecursiveTask<Integer> {        private final int n;        Factorial(int n) {            this.n = n;        }        @Override        protected Integer compute() {            if (n == 0){                return 1;            }            Factorial f = new Factorial(n - 1);            //执行子任务            f.fork();            //等待子任务结果            return n * factorial.join();        }    }}

2. ForkJoinPool 原理

和普通的线程池类似，ForkJoinPool 是一个特殊的线程池，并且也采用的是生产者 - 消费者模式。跟普通线程池共享一个队列不同，ForkJoinPool 其中维护了多个双端队列，当一个线程对应的任务队列为空的时候，线程并不会空闲，而是“窃取”其他队列的任务执行。

由于是双端队列，正常执行任务和“窃取任务”可以从两端进行出队，这样避免了数据竞争。

采用“任务窃取”这种模式，也是 ForkJoinPool 比普通线程池更加智能的体现。