Fork/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.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。

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

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

最大线程数maximumPoolSize
核心线程数corePoolSize
活跃时间keepAliveTime
阻塞队列workQueue
拒绝策略RejectedExecutionHandler

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

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

拒绝策略主要有四种：

AbortPolicy：直接丢弃任务，抛出异常，这是默认策略
CallerRunsPolicy：使用调用者所在的线程来处理任务
DiscardOldestPolicy：丢弃等待队列中最老的任务，并执行当前任务
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，最后来说说实现原理：

首先我们通过调用 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 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

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

Fork/Join

Executor

任务

任务执行

ThreadPoolExecutor

ScheduledThreadPoolExecutor

CompletionService

异步结果

CompletableFuture

热门文章

最新文章

相关课程

相关电子书