Day22 CompletableFuture与ThreadLocal

CompletableFuture

▼
text
复制代码
**CompletableFuture是Future接口的扩展和增强**。CompletableFuture实现了Future接口，并在此基础上进行了丰富地扩展，**实现了对任务的编排能力。
优势：**不需要调用get()阻塞等待结果，实现真正的非阻塞；强大的链式组合能力；完善的异常处理机制；

- 应用场景
    
    **描述依赖关系：**
    
    1. thenApply() 把前面异步任务的结果，交给后面的Function
    2. thenCompose()用来连接两个有依赖关系的任务，结果由第二个任务返回
    
    **描述and聚合关系：**
    
    1. thenCombine:任务合并，有返回值
    2. thenAccepetBoth:两个任务执行完成后，将结果交给thenAccepetBoth消耗，无返回值。
    3. runAfterBoth:两个任务都执行完成后，执行下一步操作（Runnable）。
    
    **描述or聚合关系：**
    
    1. applyToEither:两个任务谁执行的快，就使用那一个结果，有返回值。
    2. acceptEither: 两个任务谁执行的快，就消耗那一个结果，无返回值。
    3. runAfterEither: 任意一个任务执行完成，进行下一步操作(Runnable)。
    
    **并行执行：**
    
    CompletableFuture类自己也提供了anyOf()和allOf()用于支持多个CompletableFuture并行执行
    
- 创建异步操作
    
    有四个静态方法
    
    - runAsync 方法以Runnable函数式接口类型为参数，没有返回结果，supplyAsync 方法Supplier函数式接口类型为参数，返回结果类型为U；Supplier 接口的 get() 方法是有返回值的（**会阻塞**）
    - 没有指定Executor的方法会使用ForkJoinPool.commonPool() 作为它的线程池执行异步代码。如果指定线程池，则使用指定的线程池运行。ForkJoinPool默认创建的线程数是 CPU 的核数（也可以通过 JVM option:-D java.util.concurrent.ForkJoinPool.common.parallelism 来设置）
    
    ```java
    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)
    
    Runnable runnable = () -> System.out.println("执行无返回结果的异步任务");
    CompletableFuture.runAsync(runnable);
    
    CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
        System.out.println("执行有返回值的异步任务");
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "Hello World";
    });
    String result = future.get();
    System.out.println(result);
    ```
    
- 结果处理
    
    方法不以Async结尾，意味着Action使用相同的线程执行，而Async可能会使用其它的线程去执行(如果使用相同的线程池，也可能会被同一个线程选中执行)。
    
    ```java
    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)
    
    CompletableFuture.supplyAsync(() -> {
    				           try {
                            TimeUnit.SECONDS.sleep(1);
                        } catch (InterruptedException e) {
                        }
                        if (new Random().nextInt(10) % 2 == 0) {
                            int i = 12 / 0;
                        }
                        System.out.println("执行结束！");
                        return "test";
                    })
                    .whenComplete((t, action) -> System.out.println(t + " 执行完成！"))
                    .exceptionally(new Function<Throwable, String>() {
                        @Override
                        public String apply(Throwable t) {
                            System.out.println("执行失败：" + t.getMessage());
                            return "异常xxxx";
                        }
                    }).join();
    ```
    
- 结果转换
    - thenApply **接收一个普通函数**作为参数，使用该函数处理上一个CompletableFuture 调用的结果，并返回一个具有处理结果的Future对象，把前一个任务的结果T变成U后返回，不产生新的异步任务。
        
        ```java
        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)
        
        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
            int result = 100;
            System.out.println("一阶段：" + result);
            return result;
        }).thenApply(number -> {
            int result = number * 3;
            System.out.println("二阶段：" + result);
            return result;
        });
        
        System.out.println("最终结果：" + future.get());
        ```
        
    - thenCompose **接收一个返回 CompletableFuture 实例的函数**作为参数，该函数的参数是先前计算步骤的结果，把前一个任务的结果T作为参数传到**新的**异步任务里，得到结果后再拆掉一层包装返回（≈flatMap）。
        
        ```java
        public <U> CompletableFuture<U> thenCompose(Function<? super T, ? extends CompletionStage<U>> fn);
        public <U> CompletableFuture<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn) ;
        public <U> CompletableFuture<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn, Executor executor) ;
        
        CompletableFuture<Integer> future = CompletableFuture
                        .supplyAsync(() -> {
                            int number = new Random().nextInt(30);
                            System.out.println("第一阶段：" + number);
                            return number;
                        })
                        .thenCompose(param -> CompletableFuture.supplyAsync(() -> {
                            int number = param * 2;
                            System.out.println("第二阶段：" + number);
                            return number;
                        }));
         
        System.out.println("最终结果: " + future.get());
        ```
        
- 结果消费
    - thenAccept 对单个结果进行消费
        
        ```java
        public CompletionStage<Void> thenAccept(Consumer<? super T> action);
        public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action);
        public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action,Executor executor);
        
        CompletableFuture<Void> future = CompletableFuture
                .supplyAsync(() -> {
                    int number = new Random().nextInt(10);
                    System.out.println("第一阶段：" + number);
                    return number;
                }).thenAccept(number ->
                        System.out.println("第二阶段：" + number * 5));
        
        System.out.println("最终结果：" + future.get());
        ```
        
    - thenAcceptBoth 对两个结果进行消费
        
        ```java
        public <U> CompletionStage<Void> thenAcceptBoth(CompletionStage<? extends U> other,BiConsumer<? super T, ? super U> action);
        public <U> CompletionStage<Void> thenAcceptBothAsync(CompletionStage<? extends U> other,BiConsumer<? super T, ? super U> action);
        public <U> CompletionStage<Void> thenAcceptBothAsync(CompletionStage<? extends U> other,BiConsumer<? super T, ? super U> action,     Executor executor);
        
        CompletableFuture<Integer> futrue1 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
            @Override
            public Integer get() {
                int number = new Random().nextInt(3) + 1;
                try {
                    TimeUnit.SECONDS.sleep(number);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("第一阶段：" + number);
                return number;
            }
        });
        
        CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
            @Override
            public Integer get() {
                int number = new Random().nextInt(3) + 1;
                try {
                    TimeUnit.SECONDS.sleep(number);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("第二阶段：" + number);
                return number;
            }
        });
        
        futrue1.thenAcceptBoth(future2, new BiConsumer<Integer, Integer>() {
            @Override
            public void accept(Integer x, Integer y) {
                System.out.println("最终结果：" + (x + y));
            }
        }).join();
        ```
        
    - thenRun 不关心结果，只处理后续动作，并拿不到结果值
        
        ```java
        public CompletionStage<Void> thenRun(Runnable action);
        public CompletionStage<Void> thenRunAsync(Runnable action);
        public CompletionStage<Void> thenRunAsync(Runnable action,Executor executor);
        
        CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {
            int number = new Random().nextInt(10);
            System.out.println("第一阶段：" + number);
            return number;
        }).thenRun(() ->
        				// runnable不使用上一个任务计算的结果
                System.out.println("thenRun 执行"));
        System.out.println("最终结果：" + future.get());
        ```
        
- 结果组合
    - thenCombine 合并两个线程任务的结果进行处理，返回一个CompletableFuture
        
        ```java
        public <U,V> CompletionStage<V> thenCombine(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn);
        public <U,V> CompletionStage<V> thenCombineAsync(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn);
        public <U,V> CompletionStage<V> thenCombineAsync(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn,Executor executor);
        
        CompletableFuture<Integer> future1 = CompletableFuture
                .supplyAsync(new Supplier<Integer>() {
                    @Override
                    public Integer get() {
                        int number = new Random().nextInt(10);
                        System.out.println("第一阶段：" + number);
                        return number;
                    }
                });
        CompletableFuture<Integer> future2 = CompletableFuture
                .supplyAsync(new Supplier<Integer>() {
                    @Override
                    public Integer get() {
                        int number = new Random().nextInt(10);
                        System.out.println("第二阶段：" + number);
                        return number;
                    }
                });        
        CompletableFuture<Integer> result = future1
                .thenCombine(future2, new BiFunction<Integer, Integer, Integer>() {
                    @Override
                    public Integer apply(Integer x, Integer y) {
                        return x + y;
                    }
                });
        System.out.println("最终结果：" + result.get());
        ```
        
- 任务交互
将两个线程任务获取结果的速度相比较，按一定的规则进行下一步处理。
    - applyToEither 哪个任务执行得快就拿哪个的执行结果进行下一步操作
        
        ```java
        public <U> CompletionStage<U> applyToEither(CompletionStage<? extends T> other,Function<? super T, U> fn);
        public <U> CompletionStage<U> applyToEitherAsync(CompletionStage<? extends T> other,Function<? super T, U> fn);
        public <U> CompletionStage<U> applyToEitherAsync(CompletionStage<? extends T> other,Function<? super T, U> fn,Executor executor);
        
        CompletableFuture<Integer> future1 = CompletableFuture
                .supplyAsync(new Supplier<Integer>() {
                    @Override
                    public Integer get() {
                        int number = new Random().nextInt(10);
                        System.out.println("第一阶段start：" + number);
                        try {
                            TimeUnit.SECONDS.sleep(number);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        System.out.println("第一阶段end：" + number);
                        return number;
                    }
                });
        CompletableFuture<Integer> future2 = CompletableFuture
                .supplyAsync(new Supplier<Integer>() {
                    @Override
                    public Integer get() {
                        int number = new Random().nextInt(10);
                        System.out.println("第二阶段start：" + number);
                        try {
                            TimeUnit.SECONDS.sleep(number);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        System.out.println("第二阶段end：" + number);
                        return number;
                    }
                });
        
        future1.applyToEither(future2, new Function<Integer, Integer>() {
            @Override
            public Integer apply(Integer number) {
                System.out.println("最快结果：" + number);
                return number * 2;
            }
        }).join();
        ```
        
    - acceptEither 哪个执行得快就用哪个的结果做副作用
        
        ```java
        public CompletionStage<Void> acceptEither(CompletionStage<? extends T> other,Consumer<? super T> action);
        public CompletionStage<Void> acceptEitherAsync(CompletionStage<? extends T> other,Consumer<? super T> action);
        public CompletionStage<Void> acceptEitherAsync(CompletionStage<? extends T> other,Consumer<? super T> action,Executor executor);
        ```
        
    - runAfterEither 哪个任务执行完成就进行下一步操作，不关心运行结果
        
        ```java
        public CompletionStage<Void> runAfterEither(CompletionStage<?> other,Runnable action);
        public CompletionStage<Void> runAfterEitherAsync(CompletionStage<?> other,Runnable action);
        public CompletionStage<Void> runAfterEitherAsync(CompletionStage<?> other,Runnable action,Executor executor);
        ```
        
    - runAfterBoth 两个任务全部完成才进行下一步操作，不关心运行结果
        
        ```java
        public CompletionStage<Void> runAfterBoth(CompletionStage<?> other,Runnable action);
        public CompletionStage<Void> runAfterBothAsync(CompletionStage<?> other,Runnable action);
        public CompletionStage<Void> runAfterBothAsync(CompletionStage<?> other,Runnable action,Executor executor);
        ```
        
    - CompletableFuture.anyOf 任意一个先完成，就返回这个任务的CompletableFuture
    - CompletableFuture.allOf 所有任务都完成才返回，返回CompletableFuture<Void>
        
        ```java
        public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs)
        
        CompletableFuture<String> future1 = CompletableFuture
                .supplyAsync(() -> {
                    try {
                        TimeUnit.SECONDS.sleep(2);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println("future1完成！");
                    return "future1完成！";
                });
        
        CompletableFuture<String> future2 = CompletableFuture
                .supplyAsync(() -> {
                    System.out.println("future2完成！");
                    return "future2完成！";
                });
        
        CompletableFuture<Void> combindFuture = CompletableFuture
                .allOf(future1, future2);
        try {
            combindFuture.get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        System.out.println("future1: " + future1.isDone() + "，future2: " + future2.isDone());
        ```

ThreadLocal

为什么需要ThreadLocal？

ThreadLocal类用来提供线程内部的局部变量，解决的核心问题是线程隔离，让每个线程都有自己独立的副本，不需要加锁也不会有锁竞争
- 线程安全不存在锁竞争
- 数据传递每个线程都有一份副本，共享的数据直接从副本取（例如spring的事物管理）
- 线程隔离每个线程变量都独立

常用方法

ThreadLocal() 创建Thread Local对象

public void set( T value) 设置当前线程局部变量

public T get() 获取当前线程绑定的局部变量

public void remove() 移除当前线程绑定的局部变量

▼
java
复制代码
public class MyThreadLocal {
    static class WithoutThreadLocal {
        private String content;
        public String getContent() {
            return content;
        }
        public void setContent(String content) {
            this.content = content;
        }
    }
    private static ThreadLocal<String> threadLocal = new ThreadLocal<>();
    static class WithThreadLocal {
        private String content;
        private String getContent() {
            return threadLocal.get();
        }
        private void setContent(String content) {
            threadLocal.set(content);
        }

    }

    public static void main(String[] args) throws InterruptedException {
        withoutThreadLocal();
        Thread.sleep(100);
        System.out.println("---------------------------");
        withThreadLocal();
    }
    
    private static void withoutThreadLocal() {
        WithoutThreadLocal demo = new WithoutThreadLocal();
        for (int i = 0; i < 5; i++) {
            Thread thread = new Thread(() -> {
                demo.setContent(Thread.currentThread().getName() + "的数据");
                System.out.println(Thread.currentThread().getName() + "--->" + demo.getContent());
            });
            thread.setName("线程" + i);
            thread.start();
        }
    }
    private static void withThreadLocal() {
        WithThreadLocal demo = new WithThreadLocal();
        for (int i = 0; i < 5; i++) {
            Thread thread = new Thread(() -> {
                demo.setContent(Thread.currentThread().getName() + "的数据");
                System.out.println(Thread.currentThread().getName() + "--->" + demo.getContent());
            });
            thread.setName("线程" + i);
            thread.start();
        }
    }
}

ThreadLocal与synchronized的区别
- ThreadLocal 以空间换时间，每个线程都有一份副本，互不干扰；侧重于让多线程中每个线程之间的数据都相互隔离
- synchronized 以时间换空间，只有一份变量，谁先抢到就让谁先访问，访问完了才能进行下一轮抢占；侧重于多个线程之间访问资源的同步
ThreadLocal的结构
- 每个Thread里有一个类型为ThreadLocalMap的threadLocals字段，这个Map的key是ThreadLocal对象本身，value是线程本地变量副本值。
  
  调用get()方法的时候，先拿到当前的ThreadLocalMap，在用this作为key去查，所以每次取值都是拿到自己的，完全隔离其他线程。
```
▼
java
复制代码
public T get() {
    Thread t = Thread.currentThread();
    ThreadLocalMap map = getMap(t);  // 拿当前线程的 map
    if (map != null) {
        ThreadLocalMap.Entry e = map.getEntry(this);  // 用 this 当 key
        if (e != null) {
            return (T) e.value;
        }
    }
    return setInitialValue();
}
```
- 内存泄漏问题
  
  ThreadLocalMap的Entry继承了WeakReference，key是弱引用，如果ThreadLocal没有被强引用，GC会把key进行回收，但实际上value还在，此时Entry就变成了key为nul，value还占用着内存的“脏数据”。如果是在线程池，线程生命周期很长，就容易造成内存泄漏了，所以，要真正解决内存泄漏问题，每次使用完记得调用一下remove方法

CompletableFuture

ThreadLocal

目录

目录