1 Future

从 Java 1.5开始，提供了Callable和Future，通过它们可以在任务执行完毕之后得到任务执行结果。

1.1 Callable 与 Runnable

1.1.1 Runnable

package java.lang;

@FunctionalInterface
public interface Runnable {

    public abstract void run();
    
}

1.1.2 Callable

package java.util.concurrent;

@FunctionalInterface
public interface Callable<V> {

    V call() throws Exception;
    
}

1.1.3 Callable 与 Runnable 的差异

1.2 Future + Callable

1.2.1 Future

Future就是对于具体的Runnable或者Callable任务的执行结果进行取消、查询是否完成、获取结果。必要时可以通过get方法获取执行结果，该方法会阻塞直到任务返回结果。或者说，Future 提供了另一种在操作完成时通知应用程序的方式。这个对象可以看作是一个异步操作的结果的占位符；它将在未来的某个时刻完成，并提供对其结果的访问。

package java.util.concurrent;

public interface Future<V> {

    boolean cancel(boolean mayInterruptIfRunning);

    boolean isCancelled();

    boolean isDone();

    V get() throws InterruptedException, ExecutionException;

    V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;
}

1.2.2 FutureTask

1 FutureTask 类实现了RunnableFuture接口

public class FutureTask<V> implements RunnableFuture<V>

2 RunnableFuture 继承了Runnable接口和Future接口，而FutureTask实现了RunnableFuture接口。所以它既可以作为Runnable被线程执行，又可以作为Future得到Callable的返回值。

public interface RunnableFuture<V> extends Runnable, Future<V> {

1.2.3 Future + Callable 并发

1 作为 Runnable 执行

package com.xu.thread;

import java.util.List;
import java.util.ArrayList;
import java.util.concurrent.FutureTask;

/**
 * All rights Reserved, Designed By Hyacinth
 *
 * @Title: Main.java
 * @Package com.xu.thread
 * @Description:
 * @author: hyacinth
 * @date: 2021年1月26日 下午9:45:55   
 * @version V1.0
 * @Copyright:
 */
public class Main {

    public static void main(String[] args) throws Exception {

        FutureTask<List<Integer>> task = new FutureTask<>(() -> {
            List<Integer> list = new ArrayList<>();
            for (int i = 0; i < 10; i++) {
                list.add(i);
                Thread.sleep(500);
            }
            return list;
        });

        new Thread(task).start();
        task.get(20, TimeUnit.SECONDS).stream().forEach(System.out::println);

    }

}

2 作为 Callable 执行

1 Executors

1 Executors.newFixedThreadPool()

public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>(),
                                  threadFactory);
}

2 Executors.newSingleThreadExecutor()

public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) {
    return new FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>(),
                                threadFactory));
}

3 Executors.newCachedThreadPool()

public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>(),
                                  threadFactory);
}

4 Executors.newScheduledThreadPool()

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize, 
															  ThreadFactory threadFactory) {
    return new ScheduledThreadPoolExecutor(corePoolSize, threadFactory);
}

package com.xu.thread;

import java.util.List;
import java.util.ArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;

/**
 * All rights Reserved, Designed By Hyacinth
 *
 * @Title: Main.java
 * @Package com.xu.thread
 * @Description:
 * @author: hyacinth
 * @date: 2021年1月26日 下午9:45:55
 * @version V1.0
 * @Copyright:
 */
public class Main {

    public static void main(String[] args) throws Exception {

        ExecutorService executor = Executors.newCachedThreadPool();
        FutureTask<List<Integer>> task = new FutureTask<>(() -> {
            List<Integer> list = new ArrayList<>();
            for (int i = 0; i < 10; i++) {
                list.add(i);
                Thread.sleep(500);
            }
            return list;
        });

        executor.submit(task);
        task.get(20, TimeUnit.SECONDS).stream().forEach(System.out::println);

    }

}

2 ThreadPoolExecutor

public ThreadPoolExecutor(int corePoolSize,// 核心线程池大小
                          int maximumPoolSize,// 最大线程池大小 
                          long keepAliveTime,// 线程最大空闲时间
                          TimeUnit unit,// 时间单位
                          BlockingQueue<Runnable> workQueue,// 线程等待队列
                          ThreadFactory threadFactory,// 线程创建工厂
                          RejectedExecutionHandler handler)// 拒绝策略

package com.xu.thread;

import java.util.List;
import java.util.ArrayList;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * All rights Reserved, Designed By Hyacinth
 *
 * @version V1.0
 * @Title: Main.java
 * @Package com.xu.thread
 * @Description:
 * @author: hyacinth
 * @date: 2021年1月26日 下午9:45:55
 * @Copyright:
 */
public class Main {

    public static void main(String[] args) throws Exception {

        ThreadPoolExecutor executor = new ThreadPoolExecutor(15,// 核心线程池大小
                20,// 最大线程池大小
                10,// 线程最大空闲时间
                TimeUnit.MILLISECONDS,// 时间单位
                new ArrayBlockingQueue<Runnable>(5),// 线程等待队列
                Executors.defaultThreadFactory(),// 线程创建工厂
                new ThreadPoolExecutor.AbortPolicy());// 拒绝策略
        FutureTask<List<Integer>> task = new FutureTask<>(() -> {
            List<Integer> list = new ArrayList<>();
            for (int i = 0; i < 10; i++) {
                list.add(i);
                Thread.sleep(500);
            }
            return list;
        });

        executor.submit(task);
        task.get(20, TimeUnit.SECONDS).stream().forEach(System.out::println);

    }

}

1.3 Future 的局限性

2 CompletableFuture

2.1 创建 CompletableFuture 对象

CompletableFuture 提供了四个静态方法用来创建 CompletableFuture 对象：

2.1.1 CompletableFuture.runAsync

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.junit.Test;

public class UnitTest {

    ThreadPoolExecutor executor = new ThreadPoolExecutor(15, // 核心线程池大小
            20, // 最大线程池大小
            10, // 线程最大空闲时间
            TimeUnit.MILLISECONDS, // 线程最大空闲时间单位
            new ArrayBlockingQueue<>(100), // 线程等待队列
            Executors.defaultThreadFactory(), // 线程创建工厂
            new ThreadPoolExecutor.AbortPolicy()); // 拒绝策略

    @Test
    public void testCompletableFuture() throws Exception {
        CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
            System.out.println("a");
        }, executor);
    }

}

2.1.2 CompletableFuture.supplyAsync

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.junit.Test;

public class UnitTest {

    ThreadPoolExecutor executor = new ThreadPoolExecutor(15, // 核心线程池大小
            20, // 最大线程池大小
            10, // 线程最大空闲时间
            TimeUnit.MILLISECONDS, // 线程最大空闲时间单位
            new ArrayBlockingQueue<>(100), // 线程等待队列
            Executors.defaultThreadFactory(), // 线程创建工厂
            new ThreadPoolExecutor.AbortPolicy()); // 拒绝策略

    @Test
    public void testCompletableFuture() throws Exception {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            return "a";
        }, executor);
    }

}

2.2 获取 CompletableFuture 结果

2.2.1 自动获取结果

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.junit.Test;

public class UnitTest {

    ThreadPoolExecutor executor = new ThreadPoolExecutor(15, // 核心线程池大小
            20, // 最大线程池大小
            10, // 线程最大空闲时间
            TimeUnit.MILLISECONDS, // 线程最大空闲时间单位
            new ArrayBlockingQueue<>(100), // 线程等待队列
            Executors.defaultThreadFactory(), // 线程创建工厂
            new ThreadPoolExecutor.AbortPolicy()); // 拒绝策略

    @Test
    public void testCompletableFuture() throws Exception {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "a";
        }, executor).thenApply(x -> {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return x + "b";
        }).thenCompose(e -> CompletableFuture.supplyAsync(e::toUpperCase));
        System.out.println(future.get());
        System.out.println(future.join());
        System.out.println(future.getNow("C"));
        System.out.println(future.get(1, TimeUnit.SECONDS));
    }

}

AB
AB
AB
AB

2.2.2 主动获取结果

future.complete("直接使用这个结果")
future.completeExceptionally(new Throwable("运行超时"))

2.3 CompletableFuture 异步回调

2.3.1 thenAccept…

上一个任务需要有返回值

@Test
public void testCompletableFuture() throws Exception {
    CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "a";
    }).thenAcceptAsync(x -> {
        System.out.println("thenRun");
    }).thenApplyAsync(x -> {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return x + "c";
    }).thenCompose(x -> CompletableFuture.supplyAsync(x::toUpperCase));
    System.out.println(future.join());
}

2.3.2 thenRun…

上一个任务有无返回值均可

@Test
public void testCompletableFuture() throws Exception {
    CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "a";
    }).thenRun(() -> {
        System.out.println("thenRun");
    }).thenApplyAsync(x -> {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return x + "c";
    }).thenCompose(x -> CompletableFuture.supplyAsync(x::toUpperCase));
    System.out.println(future.join());
}

2.3.3 thenApply…

上一个任务需要有返回值

 @Test
 public void testCompletableFuture() throws Exception {
     CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
         try {
             Thread.sleep(2000);
         } catch (InterruptedException e) {
             e.printStackTrace();
         }
         return "a";
     }).thenApplyAsync(x -> {
         try {
             Thread.sleep(1000);
         } catch (InterruptedException e) {
             e.printStackTrace();
         }
         return x + "c";
     }).thenCompose(x -> CompletableFuture.supplyAsync(x::toUpperCase));
     System.out.println(future.join());
 }

2.3.4 handle

@Test
public void testCompletableFuture() throws Exception {
    CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "a";
    }).exceptionally((thread) -> {
        System.out.println(thread.getMessage());
        return "ERROR";
    }).thenApplyAsync(x -> {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return x + "c";
    }).handleAsync((a, b) -> {
        System.out.println(a + "handle" + b);
        return "K";
    });
    System.out.println(future.join());
}

2.3.5 whenComplete

@Test
public void testCompletableFuture() throws Exception {
    CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "a";
    }).exceptionally((thread) -> {
        System.out.println(thread.getMessage());
        return "ERROR";
    }).thenApplyAsync(x -> {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return x + "c";
    }).whenComplete((a, b) -> {
        System.out.println(a + "handle" + b);
    });
    System.out.println(future.join());
}

2.3.6 exceptionally

exceptionally 方法指定某个任务执行异常时执行的回调方法，会将抛出异常作为参数传递到回调方法中。

@Test
public void testCompletableFuture() throws Exception {
    CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "a";
    }).exceptionally((thread) -> {
        System.out.println(thread.getMessage());
        return "ERROR";
    }).thenApplyAsync(x -> {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return x + "c";
    }).thenCompose(x -> CompletableFuture.supplyAsync(x::toUpperCase));
    System.out.println(future.join());
}

2.4 CompletableFuture 组合处理

2.4.1 thenCompose

thenCompose 方法会在某个任务执行完成后，将该任务的执行结果作为方法入参然后执行指定的方法，该方法会返回一个新的CompletableFuture实例，空则返回null，否则返回新的CompletableFuture实例。

@Test
public void testCompletableFuture() throws Exception {
    CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "a";
    }).exceptionally((thread) -> {
        System.out.println(thread.getMessage());
        return "ERROR";
    }).thenApplyAsync(x -> {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return x + "c";
    }).thenCompose(x -> CompletableFuture.supplyAsync(x::toUpperCase));
    System.out.println(future.join());
}

2.4.2 thenCombine / thenAcceptBoth / runAfterBoth

@Test
public void testCompletableFuture() throws Exception {
    CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "a";
    });
    CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "b";
    });
    // thenCombine
    CompletableFuture<String> future3 = future1.thenCombine(future2, (a, b) -> {
        return a + b + "c";
    });
    // thenAcceptBoth
    CompletableFuture<Void> future4 = future1.thenAcceptBoth(future2, (a, b) -> {
        System.out.println(a + b + "c");
    });
    // runAfterBoth
    CompletableFuture<Void> future5 = future1.runAfterBoth(future2, () -> {
        System.out.println("c");
    });
}

2.4.3 applyToEither / acceptEither / runAfterEither

2.4.4 allOf / anyOf

@Test
public void testCompletableFuture() throws Exception {
    CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "a";
    });
    CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "b";
    });
    CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> {
        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "c";
    }).thenCompose(x -> CompletableFuture.supplyAsync(x::toUpperCase));
    // allOf
    CompletableFuture future4 = CompletableFuture.allOf(future1, future2, future3).whenCompleteAsync((a, b) -> {
        if (null != b) {
            System.out.println("出现异常");
        } else {
            System.out.println("全部 CompletableFuture 都完成");
        }
    });
    // anyOf
    CompletableFuture future5 = CompletableFuture.anyOf(future1, future2, future3).whenCompleteAsync((a, b) -> {
        if (null != b) {
            System.out.println("出现异常");
        } else {
            System.out.println("最先完成的 CompletableFuture 任务的结果为 " + a);
        }
    });
}

3 CompletableFuture 多任务

package com.xu.thread;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class TaskTest {

    private static final ThreadPoolExecutor executor = new ThreadPoolExecutor(10,// 核心线程池大小
            20,// 最大线程池大小
            10,// 线程最大空闲时间
            TimeUnit.MILLISECONDS,// 时间单位
            new ArrayBlockingQueue<>(1000),// 线程等待队列
            Executors.defaultThreadFactory(),// 线程创建工厂
            new ThreadPoolExecutor.AbortPolicy());// 拒绝策略

    private AtomicInteger atomic = new AtomicInteger();

    public static void main(String[] args) throws Exception {
        TaskTest task = new TaskTest();
        task.run().get().forEach(System.out::println);
        executor.shutdown();
    }

    public CompletableFuture<List<String>> run() {

        List<CompletableFuture<List<String>>> future = new ArrayList<>();

        for (int i = 0; i < 10; i++) {
            future.add(task());
        }

        return CompletableFuture.supplyAsync(() -> {
            List<String> list = new ArrayList<>();
            future.stream().parallel().forEach(task1 -> {
                list.addAll(task1.join());
            });
            return list;
        });
    }

    public CompletableFuture<List<String>> task() {
        return CompletableFuture.supplyAsync(() -> {
            // TODO:任务
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return new ArrayList<String>() {{
                add(atomic.getAndIncrement() + "->");
            }};
        }, executor);
    }

}

"C:\Program Files\Java\jdk-11.0.8\bin\java.exe" "-javaagent:D:\IDE\IntelliJ IDEA 2020.3.1\lib\idea_rt.jar=12191:D:\IDE\IntelliJ IDEA 2020.3.1\bin" -Dfile.encoding=UTF-8 -classpath E:\SourceCode\Idea-2020.2\OpenCV\out\production\OpenCV com.xu.task.CompletableFutureTest
1->
0->
7->
2->
9->
6->
3->
8->


Process finished with exit code 0

4 Future 多任务

package com.xu.task;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;

public class FutureTest {

    ExecutorService executor = Executors.newCachedThreadPool();

    AtomicInteger atomic = new AtomicInteger();

    public static void main(String[] args) {
        FutureTest test = new FutureTest();
        long t1 = System.currentTimeMillis();
        test.task().stream().forEach(e -> {
            try {
                System.out.println(e.get());
            } catch (Exception e1) {
            }
        });
        long t2 = System.currentTimeMillis();
        System.out.println(t2 - t1);
        test.executor.shutdown();
    }

    public List<Future<String>> task() {
        List<Future<String>> futures = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            futures.add(executor.submit(() -> {
                Thread.sleep(1000);
                return atomic.getAndIncrement() + "";
            }));
        }
        return futures;
    }

}

"C:\Program Files\Java\jdk-11.0.8\bin\java.exe" "-javaagent:D:\IDE\IntelliJ IDEA 2020.3.1\lib\idea_rt.jar=12239:D:\IDE\IntelliJ IDEA 2020.3.1\bin" -Dfile.encoding=UTF-8 -classpath E:\SourceCode\Idea-2020.2\OpenCV\out\production\OpenCV com.xu.task.FutureTest
0
2
1
6
9
8
7
5
4
3
1030

Process finished with exit code 0