异步和线程池-阿里云开发者社区

初始化线程的四种方式

继承 Thread

public static void main(String[] args) {
        Thread01 thread01=new Thread01();
        thread01.start();
    }
    public  static  class  Thread01 extends  Thread{
        @Override
        public void run() {
            System.out.println("当前线程"+Thread.currentThread().getId());
            int i=10/2;
            System.out.println("返回结果："+i);
        }
    }

实现 Runnable 接口

public static void main(String[] args) {
        Runnable01 runnable01=new Runnable01();
        new Thread(runnable01).start();
    }
        public  static  class  Runnable01 implements   Runnable{
        @Override
        public void run() {
            System.out.println("当前线程"+Thread.currentThread().getId());
            int i=10/2;
            System.out.println("返回结果："+i);
        }
    }

实现 Callable 接口 + FutureTask （可以拿到返回结果，可以处理异常）

public static void main(String[] args) throws ExecutionException, InterruptedException {
        FutureTask<Integer> futureTask=new FutureTask<>(new Callable01());
        new Thread(futureTask).start();
        Integer integer = futureTask.get();
        System.out.println("main---返回"+integer);
    }
        public  static  class  Callable01 implements Callable<Integer>{
        @Override
        public Integer call() throws Exception {
            System.out.println("当前线程"+Thread.currentThread().getId());
            int i=10/2;
            System.out.println("返回结果："+i);
            return  i;
        }
    }

线程池

ExecutorService executorService = Executors.newFixedThreadPool(10);
executorService.execute(new Runnable01());

方式 1 和方式 2：主进程无法获取线程的运算结果。不适合当前场景

方式 3：主进程可以获取线程的运算结果，但是不利于控制服务器中的线程资源。可以导致服务器资源耗尽。

方式 4：可以控制资源

通过线程池性能稳定，也可以获取执行结果，并捕获异常。但是，在业务复杂情况下，一个异步调用可能会依赖于另一个异步调用的执行结果。

线程池详解

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

corePoolSize: 池中一直保持的线程的数量，即使线程空闲。除非设置了 allowCoreThreadTimeOut
maximumPoolSize: 池中允许的最大的线程数
keepAliveTime: 当线程数大于核心线程数的时候，超出核心线程数的线程在最大长时间没有接到新任务就会终止释放，最终线程池维持在 corePoolSize 大小
unit: 时间单位
workQueue: 阻塞队列，用来存储等待执行的任务，如果当前对线程的需求超过了corePoolSize大小，就会放在这里等待空闲线程执行.
threadFactory：创建线程的工厂，比如指定线程名等
handler：拒绝策略，如果线程满了，线程池就会使用拒绝策略。

运行流程

1、线程池创建，准备好 core 数量的核心线程，准备接受任务

2、新的任务进来，用 core 准备好的空闲线程执行。

(1) 、core 满了，就将再进来的任务放入阻塞队列中。空闲的 core 就会自己去阻塞队列获取任务执行

(2) 、阻塞队列满了，就直接开新线程执行，最大只能开到 max 指定的数量

(3) 、max 都执行好了。Max-core 数量空闲的线程会在 keepAliveTime 指定的时间后自动销毁。最终保持到core 大小

(4) 、如果线程数开到了 max 的数量，还有新任务进来，就会使用 reject 指定的拒绝策略进行处理

3、所有的线程创建都是由指定的 factory 创建的。

面试：

一个线程池 core 7；； max 20 ，queue ：50 ，100 并发进来怎么分配的？

先有 7 个能直接得到执行，接下来 50 个进入队列排队，在多开 13 个继续执行。现在 70 个被安排上了。剩下 30 个使用拒绝策略。

具体参考：多线程之线程池 - 炒焖煎糖板栗 - 博客园 (cnblogs.com)

常见的 4 种线程池

newCachedThreadPool

创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。

newFixedThreadPool

创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。

newScheduledThreadPool

创建一个定长线程池，支持定时及周期性任务执行。

newSingleThreadExecutor

创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。

开发中为什么使用线程池

降低资源的消耗

通过重复利用已经创建好的线程降低线程的创建和销毁带来的损耗

提高响应速度

因为线程池中的线程数没有超过线程池的最大上限时，有的线程处于等待分配任务的状态，当任务来时无需创建新的线程就能执行

提高线程的可管理性

线程池会根据当前系统特点对池内的线程进行优化处理，减少创建和销毁线程带来的系统开销。无限的创建和销毁线程不仅消耗系统资源，还降低系统的稳定性，使用线程池进行统一分配

CompletableFuture 异步编排

业务场景：

查询商品详情页的逻辑比较复杂，有些数据还需要远程调用，必然需要花费更多的时间。

获取sku的基本信息 0.5s
获取sku的图片信息 0.5s
获取sku的促销信息 1s
获取spu的所有销售属性 1s
获取规格参数 1.5s
spu详情 1s

假如商品详情页的每个查询，需要如下标注的时间才能完成，那么，用户需要 5.5s 后才能看到商品详情页的内容。很显然是不能接受的。如果有多个线程同时完成这 6 步操作，也许只需要 1.5s 即可完成响应。

在 Java 8 中, 新增加了一个包含 50 个方法左右的类: CompletableFuture，提供了非常强大的Future 的扩展功能，可以帮助我们简化异步编程的复杂性，提供了函数式编程的能力，可以通过回调的方式处理计算结果，并且提供了转换和组合 CompletableFuture 的方法。

CompletableFuture 和 FutureTask 同属于 Future 接口的实现类，都可以获取线程的执行结果。

创建异步对象

CompletableFuture 提供了四个静态方法来创建一个异步操作。

public static CompletableFuture<Void> runAsync(Runnable runnable)
public static CompletableFuture<Void> runAsync(Runnable runnable,Executor executor)
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier,Executor executor)

runXxxx 都是没有返回结果的，supplyXxx 都是可以获取返回结果的
可以传入自定义的线程池，否则就用默认的线程池；

启动异步任务

public static void main(String[] args) {
      System.out.println("Main方法开始");
        //runAsync没有返回结果
        CompletableFuture.runAsync(()->{
            System.out.println("当前线程"+Thread.currentThread().getId());
            int i=10/2;
            System.out.println("返回结果："+i);
        },executorService);
        //supplyAsync 有返回结果
        CompletableFuture<Integer> integerCompletableFuture = CompletableFuture.supplyAsync(() -> {
            System.out.println("当前线程" + Thread.currentThread().getId());
            int i = 10 / 2;
            System.out.println("返回结果：" + i);
            return i;
        }, executorService);
        System.out.println("返回值："+integerCompletableFuture.get());
        System.out.println("Main方法结束");
    }

计算完成时回调方法

public CompletableFuture<T> whenComplete(BiConsumer<? super T, ? super Throwable> action)
public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action)
public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action, Executor executor)
public CompletableFuture<T> exceptionally(Function<Throwable, ? extends T> fn)

whenComplete 可以处理正常和异常的计算结果，exceptionally 处理异常情况

whenComplete 和 whenCompleteAsync 的区别：

whenComplete：是执行当前任务的线程执行继续执行 whenComplete 的任务。
whenCompleteAsync：是执行把 whenCompleteAsync 这个任务继续提交给线程池来进行执行

方法不以 Async 结尾，意味着 Action 使用相同的线程执行，而 Async 可能会使用其他线程执行（如果是使用相同的线程池，也可能会被同一个线程选中执行）

方法完成后的感知

public class CompletableFutureDemo {
    public static void main(String[] args) {
      CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("当前线程" + Thread.currentThread().getId());
            int i = 10 / 0;
            System.out.println("返回结果：" + i);
            return i;
        }, executorService).whenCompleteAsync((res,expect)->{
            //虽然能得到异常信息但是没法处理结果
            System.out.println("异步完成返回结果："+res+"，异常信息："+expect);
        }).exceptionally(throwable -> {
            //可以感知异常，并返回默认值
            return  10;
        });
        System.out.println("返回值："+future.get());
        System.out.println("Main方法结束");
    }
}

handle 方法

和complete方法一样，不过可以处理异常，并返回结果

public <U> CompletableFuture<U> handle(BiFunction<? super T, Throwable, ? extends U> fn) {
        return uniHandleStage(null, fn);
    }
public <U> CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn) {
        return uniHandleStage(asyncPool, fn);
    }
public <U> CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn, Executor executor) {
        return uniHandleStage(screenExecutor(executor), fn);
    }

方法完成后的处理

public class CompletableFutureDemo {
    public static void main(String[] args) {
     CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("当前线程" + Thread.currentThread().getId());
            int i = 10 / 0;
            System.out.println("返回结果：" + i);
            return i;
        }, executorService).handle((res,except)->{
            //如果返回成功
            if(res!=null) {
                res=res*2;
            }
            //如果返回失败
            if(except!=null){
                return  0;
            }
            return  0;
        });
        System.out.println("返回值："+future.get());
        System.out.println("Main方法结束");
    }
}

线程串行化方法

public <U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn)
public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn)
public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn, Executor executor)
public CompletableFuture<Void> thenAccept(Consumer<? super T> action)
public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action)
public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action,Executor executor)
public CompletableFuture<Void> thenRun(Runnable action)
public CompletableFuture<Void> thenRunAsync(Runnable action)
public CompletableFuture<Void> thenRunAsync(Runnable action, Executor executor)

有两个线程A、B先后执行

thenRun：A线程执行完不需要A的返回值，直接执行B

public static void main(String[] args) {
         CompletableFuture.supplyAsync(() -> {
            System.out.println("当前线程" + Thread.currentThread().getId());
            int i = 10 / 5;
            System.out.println("返回结果：" + i);
            return i;
        }, executorService).thenRunAsync(()->{
            System.out.println("方法二执行");
        },executorService);
    }

假如有异常不能获取上一步的返回结果

thenAccept：A线程执行完，B需要A的返回值，然后执行B，无返回结果

public static void main(String[] args) {
            CompletableFuture.supplyAsync(() -> {
            System.out.println("当前线程" + Thread.currentThread().getId());
            int i = 10 / 5;
            System.out.println("返回结果：" + i);
            return i;
        }, executorService).thenAcceptAsync((res)->{
            System.out.println("任务二获取任务一结果："+res);
        },executorService);
    }

thenApply：当一个线程依赖另一个线程时，获取上一个任务返回的结果，并返回当前任务的返回值

public static void main(String[] args) {
  CompletableFuture<String> stringCompletableFuture = CompletableFuture.supplyAsync(() -> {
            System.out.println("当前线程" + Thread.currentThread().getId());
            int i = 10 / 5;
            System.out.println("返回结果：" + i);
            return i;
        }, executorService).thenApplyAsync((res) -> {
            System.out.println("任务二获取任务一结果：" + res);
            return "hello," + res;
        }, executorService);
        System.out.println("方法最终返回结果："+stringCompletableFuture.get());
}

两任务组合

public <U,V> CompletableFuture<V> thenCombine(
        CompletionStage<? extends U> other,
        BiFunction<? super T,? super U,? extends V> fn) {
        return biApplyStage(null, other, fn);
    }
    public <U,V> CompletableFuture<V> thenCombineAsync(
        CompletionStage<? extends U> other,
        BiFunction<? super T,? super U,? extends V> fn) {
        return biApplyStage(asyncPool, other, fn);
    }
    public <U,V> CompletableFuture<V> thenCombineAsync(
        CompletionStage<? extends U> other,
        BiFunction<? super T,? super U,? extends V> fn, Executor executor) {
        return biApplyStage(screenExecutor(executor), other, fn);
    }
 public <U> CompletableFuture<Void> thenAcceptBoth(
        CompletionStage<? extends U> other,
        BiConsumer<? super T, ? super U> action) {
        return biAcceptStage(null, other, action);
    }
    public <U> CompletableFuture<Void> thenAcceptBothAsync(
        CompletionStage<? extends U> other,
        BiConsumer<? super T, ? super U> action) {
        return biAcceptStage(asyncPool, other, action);
    }
    public <U> CompletableFuture<Void> thenAcceptBothAsync(
        CompletionStage<? extends U> other,
        BiConsumer<? super T, ? super U> action, Executor executor) {
        return biAcceptStage(screenExecutor(executor), other, action);
    }
   public CompletableFuture<Void> runAfterBoth(CompletionStage<?> other,
                                                Runnable action) {
        return biRunStage(null, other, action);
    }
    public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other,
                                                     Runnable action) {
        return biRunStage(asyncPool, other, action);
    }
    public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other,
                                                     Runnable action,
                                                     Executor executor) {
        return biRunStage(screenExecutor(executor), other, action);
    }

两个任务必须都完成，触发该任务

thenCombine：组合两个 future，获取两个 future 的返回结果，并返回当前任务的返回值

public static void main(String[] args) {
         CompletableFuture<Integer> future01 = CompletableFuture.supplyAsync(() -> {
            System.out.println("任务一线程" + Thread.currentThread().getId());
            int i = 10 / 5;
            System.out.println("任务一返回结果：" + i);
            return i;
        }, executorService);
        CompletableFuture<String> future02 = CompletableFuture.supplyAsync(() -> {
            System.out.println("任务二线程" + Thread.currentThread().getId());
            System.out.println("任务二返回");
            return  "hello";
        }, executorService);
        CompletableFuture<String> stringCompletableFuture = future01.thenCombineAsync(future02, (f1, f2) -> {
            return f1 + ":" + f2 + "=>hello";
        }, executorService);
        System.out.println("最终处理结果："+stringCompletableFuture.get());
}

thenAcceptBoth：组合两个 future，获取两个 future 任务的返回结果，然后处理任务，没有返回值。

public static void main(String[] args) {
           CompletableFuture<Integer> future01 = CompletableFuture.supplyAsync(() -> {
            System.out.println("任务一线程" + Thread.currentThread().getId());
            int i = 10 / 5;
            System.out.println("任务一返回结果：" + i);
            return i;
        }, executorService);
        CompletableFuture<String> future02 = CompletableFuture.supplyAsync(() -> {
            System.out.println("任务二线程" + Thread.currentThread().getId());
            System.out.println("任务二返回");
            return  "hello";
        }, executorService);
        future01.thenAcceptBothAsync(future02,(f1,f2)->{
            System.out.println("任务三开始,得到之前的结果f1："+f1+",f2:"+f2);
        },executorService);
    }

runAfterBoth：组合两个 future，不需要获取 future 的结果，只需两个 future 处理完任务后，处理该任务

public static void main(String[] args) { 
CompletableFuture<Integer> future01 = CompletableFuture.supplyAsync(() -> {
            System.out.println("任务一线程" + Thread.currentThread().getId());
            int i = 10 / 5;
            System.out.println("任务一返回结果：" + i);
            return i;
        }, executorService);
        CompletableFuture<String> future02 = CompletableFuture.supplyAsync(() -> {
            System.out.println("任务二线程" + Thread.currentThread().getId());
            System.out.println("任务二返回");
            return  "hello";
        }, executorService);
        future01.runAfterBothAsync(future02,()->{
            System.out.println("任务三开始");
        },executorService);
    }

两任务组合一任务完成

public <U> CompletableFuture<U> applyToEither(
        CompletionStage<? extends T> other, Function<? super T, U> fn) {
        return orApplyStage(null, other, fn);
    }
    public <U> CompletableFuture<U> applyToEitherAsync(
        CompletionStage<? extends T> other, Function<? super T, U> fn) {
        return orApplyStage(asyncPool, other, fn);
    }
    public <U> CompletableFuture<U> applyToEitherAsync(
        CompletionStage<? extends T> other, Function<? super T, U> fn,
        Executor executor) {
        return orApplyStage(screenExecutor(executor), other, fn);
    }
    public CompletableFuture<Void> acceptEither(
        CompletionStage<? extends T> other, Consumer<? super T> action) {
        return orAcceptStage(null, other, action);
    }
    public CompletableFuture<Void> acceptEitherAsync(
        CompletionStage<? extends T> other, Consumer<? super T> action) {
        return orAcceptStage(asyncPool, other, action);
    }
    public CompletableFuture<Void> acceptEitherAsync(
        CompletionStage<? extends T> other, Consumer<? super T> action,
        Executor executor) {
        return orAcceptStage(screenExecutor(executor), other, action);
    }
    public CompletableFuture<Void> runAfterEither(CompletionStage<?> other,
                                                  Runnable action) {
        return orRunStage(null, other, action);
    }
    public CompletableFuture<Void> runAfterEitherAsync(CompletionStage<?> other,
                                                       Runnable action) {
        return orRunStage(asyncPool, other, action);
    }
    public CompletableFuture<Void> runAfterEitherAsync(CompletionStage<?> other,
                                                       Runnable action,
                                                       Executor executor) {
        return orRunStage(screenExecutor(executor), other, action);
    }

当两个任务中，任意一个 future 任务完成的时候，执行任务

applyToEither：两个任务有一个执行完成，获取它的返回值，处理任务并有新的返回值。

acceptEither：两个任务有一个执行完成，获取它的返回值，处理任务，没有新的返回值。

runAfterEither：两个任务有一个执行完成，不需要获取 future 的结果，处理任务，也没有返回值。

多任务组合

public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs) {
        return andTree(cfs, 0, cfs.length - 1);
    }
        public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs) {
        return orTree(cfs, 0, cfs.length - 1);
    }

allOf：等待所有任务完成

anyOf：只要有一个任务完成

异步和线程池

初始化线程的四种方式

线程池详解

运行流程

常见的 4 种线程池

开发中为什么使用线程池

CompletableFuture 异步编排

创建异步对象

启动异步任务

计算完成时回调方法

handle 方法

线程串行化方法

两任务组合

两任务组合一任务完成

多任务组合

热门文章

最新文章

相关电子书

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

异步和线程池

初始化线程的四种方式

线程池详解

运行流程

常见的 4 种线程池

开发中为什么使用线程池

CompletableFuture 异步编排

创建异步对象

启动异步任务

计算完成时回调方法

handle 方法

线程串行化方法

两任务组合

两任务组合一任务完成

多任务组合

热门文章

最新文章

相关电子书