JUC并发编程之CompletableFuture基础用法

Posted 2022-11-07 知道什么是码怪吗？

目录

实现多线程的四种方式

方式一：继承Thread类

方式二：实现Runnable接口

方式三：实现Callable接口

方式四：线程池

创建异步对象

回调方法

handle方法 

线程串行化 

任务组合

组合任务单任务完成及执行

---

实现多线程的四种方式

方式一：继承Thread类

public static class Thread01 extends Thread 
    @Override
    public void run() 
    

执行任务方式： 

Thread thread = new Thread01();
thread.start();

方式二：实现Runnable接口

public static class Runable01 implements Runnable 
    @Override
    public void run() 
    

执行任务方式：  

Runable01 runable01 = new Runable01();
new Thread(runable01).start();

方式三：实现Callable接口

    public static class Callable01 implements Callable<Object> 
        @Override
        public Integer call() throws Exception 
            return 0;
        
    

 执行任务方式：

FutureTask<Integer> futureTask = new FutureTask<>(new Callable01());
new Thread(futureTask).start();// 执行任务
System.out.println(futureTask.get());// 得到返回值

方式四：线程池

前三种方式执行的过程相当于每次新任务都会新增一个线程，这是十分消耗内存的，为了统一管理线程，引入了线程池。执行任务的线程从线程池中拿。

查看线程池构造方法的源码，其七个构造参数分别表示的含义如下：

参数名	参数类型	参数含义
corePoolSize	int	核心线程数，即线程池中一直保持的线程数量。
maximumPoolSize	int	允许的最大线程数
keepAliveTime	long	线程数大于核心线程数时，线程在该时间下没有收到任务就会自动释放。
unit	TimeUnit	时间单位
workQueue	BlockingQueue<Runnable>	阻塞队列，存储等待执行的任务
threadFactory	ThreadFactory	线程工厂，创造线程
handler	RejectedExecutionHandler	拒绝策略

ThreadPoolExecutor executor = new ThreadPoolExecutor(
                20,// 核心线程数
                200,// 最大线程数
                10,// 线程关闭时间
                TimeUnit.SECONDS,// 时间：秒
                new LinkedBlockingDeque<>(100000),// 存储的异步任务最大数
                Executors.defaultThreadFactory(),// 线程工厂
                new ThreadPoolExecutor.AbortPolicy());//拒绝策略

 接下来为了方便测试，我们开启一个固定10个线程的线程池。

ExecutorService executor = Executors.newFixedThreadPool(10);// 开启一个固定10个线程的线程池

创建异步对象

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

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)

这四种方法的区别在于：

方法名	区别
supplyAsync(Supplier<U> supplier)	可以返回结果
runAsync(Runnable runnable)	不能返回结果
supplyAsync(Supplier<U> supplier, Executor executor)	可以返回结果，可指定线程池
runAsync(Runnable runnable, Executor executor)	不能返回结果，可指定线程池

CompletableFuture.runAsync(() -> 
    System.out.println("runAsync方法，当前线程：" + Thread.currentThread().getId());
, executor);

CompletableFuture.supplyAsync(() -> 
    System.out.println("supplyAsync方法，当前线程：" + Thread.currentThread().getId());
    return "这是结果";
, executor);

回调方法

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)

方法以Async结尾，表示此任务可能被其他线程执行，并且也可以指定线程池。如果不以Async结尾，则是继续用当前的线程执行任务。

方法名	区别
whenComplete	继续以当前线程执行本次任务
whenCompleteAsync	将当前任务提交给线程池处理

whenCompleteAsync方法

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> 
    System.out.println("supplyAsync方法，当前线程：" + Thread.currentThread().getId());
    int i = 10 / 0;// 制造数学异常
    return i;
, executor).whenCompleteAsync((res, excption) -> // 只能感知异常，无法处理异常
    System.out.println("res:   " + res + "excption:   " + excption);
, executor).exceptionally(throwable -> // 处理异常
    return 10;// 返回默认值
);
System.out.println("结果为：" + future.get());// 获取返回值

whenComplete方法 

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> 
    System.out.println("supplyAsync方法，当前线程：" + Thread.currentThread().getId());
    int i = 10 / 0;// 制造数学异常
    return i;
, executor).whenComplete((res, excption) -> // 只能感知异常，无法处理异常
    System.out.println("res:   " + res + "excption:   " + excption);
).exceptionally(throwable -> // 处理异常
    return 10;// 返回默认值
);
System.out.println("结果为：" + future.get());

执行结果为： 

handle方法 

public <U> CompletableFuture<U> handle(BiFunction<? super T, Throwable, ? extends U> fn)

public <U> CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn)

public <U> CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn, Executor executor)

方法以Async结尾，表示此任务可能被其他线程执行，并且也可以指定线程池。如果不以Async结尾，则是继续用当前的线程执行任务。

该方法可对结果做最后的处理，可以处理异常并且有返回值。

下面这段代码，如果无异常，返回500，有异常返回0。

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> 
    System.out.println("supplyAsync方法，当前线程：" + Thread.currentThread().getId());
    int i = 10 / 2 ;
    return i;
, executor).handle((res, thr) -> // 感知异常并处理异常
    System.out.println("res:   " + res + "    thr:   " + thr);
    if (res != null) return res * 100;// 如果结果不为空
    if (thr != null) return 0;// 如果异常不为空
    return 100;
);
System.out.println(future.get());

正常执行结果：

异常执行结果：

线程串行化 

主要就是以下9个方法，以Async结尾的表示可以被其他线程执行，而参数中有Executor的表示可以指定线程池。

 三种方式的区别：

方法名	区别
thenApply	获取上一个任务的结果，并且返回当前任务的结果
thenAccept	获取上一个任务的结果，无返回值
thenRun	无法获取上一个任务的结果，无返回值

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> 
    System.out.println("supplyAsync方法，当前线程：" + Thread.currentThread().getId());
    int i = 10 / 2;
    return i;
, executor).thenApplyAsync((res) -> // 前一个任务处理完成之后，后续处理
    System.out.println("任务二启动");
    return res * 10;
, executor);
System.out.println(future.get());

执行结果：

任务组合

其意思就是将两个任务组合在一起，当两个任务都执行完成之后在执行给定的任务。

其方法名和参数值至此以及无需赘述了。 

方法名	区别
thenCombine	获取两个任务的返回结果，并且返回当前任务的返回值
thenAcceptBoth	获取两个任务的返回结果，并且无返回值
runAfterBoth	不获取两个任务的结果，并且无返回值

下面编写了两个任务

CompletableFuture<Integer> future01 = CompletableFuture.supplyAsync(() -> 
    System.out.println("任务一开始：" + Thread.currentThread().getId());
    System.out.println("任务一结束");
    return 10;
, executor);

CompletableFuture<Integer> future02 = CompletableFuture.supplyAsync(() -> 
    System.out.println("任务二开始：" + Thread.currentThread().getId());
    System.out.println("任务二结束");
    return 123;
, executor);

 runAfterAsync方法

future01.runAfterBothAsync(future02, () -> // 无法获取任务的返回值，也不返回结果
    System.out.println("开始执行，无返回值也无参数");
, executor);

执行结果 

thenAcceptBothAsync方法

future01.thenAcceptBothAsync(future02, (f1, f2) -> // 获取两个任务的返回值，不返回结果
    System.out.println("两个任务结果之和为：" + f1 * f2);
, executor);

执行结果  

  

 thenCombineAsync方法

CompletableFuture<String> future = future01.thenCombineAsync(future02, (f1, f2) -> // 获取两个任务的返回值，并返回结果
    System.out.println("两个任务结果之和为：" + f1 * f2);
    return "hello world";
, executor);
System.out.println("thenCombineAsync返回值为： "+future.get());

执行结果  

 

组合任务单任务完成及执行

其意思就是将两个任务组合在一起，只要有一个任务完成了就执行给定的任务。

方法名	区别
applyToEither	获取先完成任务的结果，有返回值
acceptEither	获取先完成任务的结果，无返回值
runAfterEither	不获取任务的结果，无返回值

下面编写了两个任务，其中任务一睡眠500ms

CompletableFuture<Integer> future01 = CompletableFuture.supplyAsync(() -> 
    System.out.println("任务一开始：" + Thread.currentThread().getId());
    try 
        Thread.sleep(500);// 线程睡眠500ms
     catch (InterruptedException e) 
        e.printStackTrace();
    
    System.out.println("任务一结束");
    return 10;
, executor);

CompletableFuture<Integer> future02 = CompletableFuture.supplyAsync(() -> 
    System.out.println("任务二开始：" + Thread.currentThread().getId());
    System.out.println("任务二结束");
    return 123;
, executor);

runAfterEither方法

future01.runAfterEitherAsync(future02, () -> 
    System.out.println("runAfterEitherAsync开始执行");
, executor);

执行结果

acceptEitherAsync方法

future01.acceptEitherAsync(future02, (res) -> 
    System.out.println("acceptEitherAsync方法开始执行    " + res);
, executor);

执行结果 

 

applyToEitherAsync方法

CompletableFuture<String> result = future01.applyToEitherAsync(future02, (res) -> 
    System.out.println("applyToEitherAsync方法开始执行    " + res);
    return "res:   " + res.toString();
, executor);
System.out.println(result.get());

执行结果 

多任务组合

将多个任务进行组合。 

allOf：等待所有任务完成。

anyOf：只要有一个任务完成。

下面代码编写了3个任务

CompletableFuture<Integer> future01 = CompletableFuture.supplyAsync(() -> 
    System.out.println("任务一开始：" + Thread.currentThread().getId());
    try 
        Thread.sleep(500);// 线程睡眠500ms
     catch (InterruptedException e) 
        e.printStackTrace();
    
    System.out.println("任务一结束");
    return 10;
, executor);

CompletableFuture<Integer> future02 = CompletableFuture.supplyAsync(() -> 
    System.out.println("任务二开始：" + Thread.currentThread().getId());
    System.out.println("任务二结束");
    return 20;
, executor);

CompletableFuture<Integer> future03 = CompletableFuture.supplyAsync(() -> 
    System.out.println("任务三开始：" + Thread.currentThread().getId());
    System.out.println("任务三结束");
    return 30;
, executor);

allOf & anyOf 

CompletableFuture<Void> allOf = future01.allOf(future02, future03);
CompletableFuture<Object> anyOf = future01.anyOf(future02, future03);