package com.chezhibao.mockserver.mulitithread;

import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

/**
 * @author hepei
 * @date 2019/8/26 10:05
 **/
public class CompletableFutureDemo {

    /***
     * 无返回值
     */
    private static void runAsync() throws ExecutionException, InterruptedException {
        System.out.println( "start ..." + System.currentTimeMillis() );
        CompletableFuture<Void> future = CompletableFuture.runAsync( () -> {
            try {
                TimeUnit.SECONDS.sleep( 1 );
            } catch (InterruptedException ignored) {
            }
            System.out.println( "end ..." + System.currentTimeMillis() );
        } );
        future.get();
    }

    /***
     * 有返回值
     */
    private static void supplyAsync() throws ExecutionException, InterruptedException {
        System.out.println( "start ..." + System.currentTimeMillis() );
        CompletableFuture<Long> future = CompletableFuture.supplyAsync( () -> {
            try {
                TimeUnit.SECONDS.sleep( 1 );
            } catch (InterruptedException ignored) {
            }
            System.out.println( "end ..." + System.currentTimeMillis() );
            return System.currentTimeMillis();
        } );

        Long aLong = future.get();
        System.out.println( "time = " + aLong );
    }


    private static void whenComplete() throws InterruptedException {
        CompletableFuture<Void> future = CompletableFuture.runAsync( () -> {
            try {
                TimeUnit.SECONDS.sleep( 1 );
            } catch (InterruptedException ignored) {
            }
            System.out.println( "......" );
            int i = 12 / 0;
            System.out.println( "run end ..." );
        } );
        // 执行当前任务的线程执行继续执行 whenComplete 的任务
        future.whenComplete( new BiConsumer<Void, Throwable>() {
            @Override
            public void accept(Void aVoid, Throwable throwable) {
                System.out.println( "执行完成:" + throwable.getMessage() );
            }
        } );
        //当future执行发生异常后执行
        future.exceptionally( new Function<Throwable, Void>() {
            @Override
            public Void apply(Throwable throwable) {
                System.out.println( "执行失败" + throwable.getMessage() );
                return null;
            }
        } );
        TimeUnit.SECONDS.sleep( 2 );
    }

    /***
     * 第二个任务依赖第一个任务的结果
     * thenApply 只可以执行正常的任务，任务出现异常则不执行 thenApply 方法
     */
    private static void thenApply() throws ExecutionException, InterruptedException {
        CompletableFuture<Long> future = CompletableFuture.supplyAsync( new Supplier<Long>() {
            @Override
            public Long get() {
                long result = new Random().nextInt( 100 );
                System.out.println( "result1=" + result );
                return result;
            }
        } ).thenApply( new Function<Long, Long>() {
            @Override
            public Long apply(Long t) {
                long result = t * 5;
                System.out.println( "result2=" + result );
                return result;
            }
        }).exceptionally( new Function<Throwable, Long>() {
            @Override
            public Long apply(Throwable throwable) {
                System.out.println("异常执行:"+throwable.getMessage());
                return 0L;
            }
        } );
        long result = future.get();
        System.out.println( result );
    }

    /***
     * 任务完成后再执行，还可以处理异常的任务
     */
    private static void handle() throws ExecutionException, InterruptedException {
        CompletableFuture<Long> future = CompletableFuture.supplyAsync( new Supplier<Long>() {
            @Override
            public Long get() {
                int i = 10 / 0;
                return new Random().nextLong();
            }
        } ).handle( new BiFunction<Long, Throwable, Long>() {
            @Override
            public Long apply(Long param, Throwable throwable) {
                Long result = -1L;
                if (throwable == null) {
                    result = param * 2;
                } else {
                    System.out.println( throwable.getMessage() );
                }
                return result;
            }
        } );
        System.out.println( future.get() );
    }

    /***
     * 接收任务的处理结果，并消费处理，无返回结果。
     */
    private static void thenAccept() throws ExecutionException, InterruptedException {
        CompletableFuture<Void> future = CompletableFuture.supplyAsync( new Supplier<Integer>() {
            @Override
            public Integer get() {
                return new Random().nextInt( 10 );
            }
        } ).thenAccept( integer -> {
            System.out.println( integer + System.currentTimeMillis() );
        } );
        future.get();
    }

    /***
     * 不关心任务的处理结果。只要上面的任务执行完成，就开始执行 thenRun 。
     */
    private static void thenRun() throws ExecutionException, InterruptedException {
        CompletableFuture future = CompletableFuture.supplyAsync( new Supplier<Integer>() {
            @Override
            public Integer get() {
                return new Random().nextInt( 10 );
            }
        } ).thenRun( () -> System.out.println( "thenRun...." ) );
        future.get();
    }

    /***
     * 把两个任务的结果一块交给 thenCombine 来处理
     */
    private static void thenCombine() throws ExecutionException, InterruptedException {
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync( new Supplier<String>() {
            @Override
            public String get() {
                return "hello";
            }
        } );

        CompletableFuture<String> future2 = CompletableFuture.supplyAsync( new Supplier<String>() {
            @Override
            public String get() {
                return "world";
            }
        } );

        CompletableFuture<String> result = future1.thenCombine( future2, new BiFunction<String, String, String>() {
            @Override
            public String apply(String s, String s2) {
                return s + " " + s2;
            }
        } );
        System.out.println( result.get() );
    }

    /***
     * 当两个CompletionStage都执行完成后，把结果一块交给thenAcceptBoth来进行消耗
     */
    private static void thenAcceptBoth() throws ExecutionException, InterruptedException {
        CompletableFuture<Integer> f1 = CompletableFuture.supplyAsync( new Supplier<Integer>() {
            @Override
            public Integer get() {
                int t = new Random().nextInt( 3 );
                try {
                    TimeUnit.SECONDS.sleep( t );
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println( "f1=" + t );
                return t;
            }
        } );

        CompletableFuture<Integer> f2 = CompletableFuture.supplyAsync( new Supplier<Integer>() {
            @Override
            public Integer get() {
                int t = new Random().nextInt( 3 );
                try {
                    TimeUnit.SECONDS.sleep( t );
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println( "f2=" + t );
                return t;
            }
        } );

        f1.thenAcceptBoth( f2, new BiConsumer<Integer, Integer>() {
            @Override
            public void accept(Integer integer, Integer integer2) {
                System.out.println( "f1=" + integer + ";f2=" + integer2 + ";" );
            }
        } );
    }

    /***
     * 两个CompletionStage，谁执行返回的结果快，就用那个CompletionStage的结果进行下一步的转化操作。
     */
    private static void applyToEither() throws ExecutionException, InterruptedException {
        CompletableFuture<Integer> f1 = CompletableFuture.supplyAsync( new Supplier<Integer>() {
            @Override
            public Integer get() {
                int t = new Random().nextInt( 3 );
                try {
                    TimeUnit.SECONDS.sleep( t );
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println( "f1=" + t );
                return t;
            }
        } );
        CompletableFuture<Integer> f2 = CompletableFuture.supplyAsync( new Supplier<Integer>() {
            @Override
            public Integer get() {
                int t = new Random().nextInt( 3 );
                try {
                    TimeUnit.SECONDS.sleep( t );
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println( "f2=" + t );
                return t;
            }
        } );

        CompletableFuture<Integer> result = f1.applyToEither( f2, new Function<Integer, Integer>() {
            @Override
            public Integer apply(Integer t) {
                System.out.println( t );
                return t * 2;
            }
        } );
        System.out.println( result.get() );
    }

    /***
     * 两个CompletionStage，谁执行返回的结果快，我就用那个CompletionStage的结果进行下一步的消耗操作。
     */
    private static void acceptEither() {
        CompletableFuture<Integer> f1 = CompletableFuture.supplyAsync( new Supplier<Integer>() {
            @Override
            public Integer get() {
                int t = new Random().nextInt( 3 );
                try {
                    TimeUnit.SECONDS.sleep( t );
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println( "f1=" + t );
                return t;
            }
        } );

        CompletableFuture<Integer> f2 = CompletableFuture.supplyAsync( new Supplier<Integer>() {
            @Override
            public Integer get() {
                int t = new Random().nextInt( 3 );
                try {
                    TimeUnit.SECONDS.sleep( t );
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println( "f2=" + t );
                return t;
            }
        } );
        f1.acceptEither( f2, new Consumer<Integer>() {
            @Override
            public void accept(Integer t) {
                System.out.println( t );
            }
        } );
    }

    /***
     * 两个CompletionStage，任何一个完成了都会执行下一步的操作（Runnable）
     */
    private static void runAfterEither() {
        CompletableFuture<Integer> f1 = CompletableFuture.supplyAsync( new Supplier<Integer>() {
            @Override
            public Integer get() {
                int t = new Random().nextInt( 3 );
                try {
                    TimeUnit.SECONDS.sleep( t );
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println( "f1=" + t );
                return t;
            }
        } );

        CompletableFuture<Integer> f2 = CompletableFuture.supplyAsync( new Supplier<Integer>() {
            @Override
            public Integer get() {
                int t = new Random().nextInt( 3 );
                try {
                    TimeUnit.SECONDS.sleep( t );
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println( "f2=" + t );
                return t;
            }
        } );
        f1.runAfterEither( f2, new Runnable() {
            @Override
            public void run() {
                System.out.println( "上面有一个已经完成了。" );
            }
        } );
    }

    /***
     * 两个CompletionStage，都完成了计算才会执行下一步的操作（Runnable）
     */
    private static void runAfterBoth() throws ExecutionException, InterruptedException {
        CompletableFuture<Integer> f1 = CompletableFuture.supplyAsync( new Supplier<Integer>() {
            @Override
            public Integer get() {
                int t = new Random().nextInt( 3 );
                try {
                    TimeUnit.SECONDS.sleep( t );
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println( "f1=" + t );
                return t;
            }
        } );

        CompletableFuture<Integer> f2 = CompletableFuture.supplyAsync( new Supplier<Integer>() {
            @Override
            public Integer get() {
                int t = new Random().nextInt( 3 );
                try {
                    TimeUnit.SECONDS.sleep( t );
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println( "f2=" + t );
                return t;
            }
        } );
        f1.get();
        f2.get();
        f1.runAfterBoth( f2, new Runnable() {
            @Override
            public void run() {
                System.out.println( "上面两个任务都执行完成了。" );
            }
        } );
    }

    /***
     * thenCompose 方法允许你对两个 CompletionStage 进行流水线操作，第一个操作完成时，将其结果作为参数传递给第二个操作。
     */
    private static void thenCompose() throws ExecutionException, InterruptedException {
        CompletableFuture<Integer> f = CompletableFuture.supplyAsync( new Supplier<Integer>() {
            @Override
            public Integer get() {
                int t = new Random().nextInt( 3 );
                System.out.println( "t1=" + t );
                return t;
            }
        } ).thenCompose( new Function<Integer, CompletionStage<Integer>>() {
            @Override
            public CompletionStage<Integer> apply(Integer param) {
                return CompletableFuture.supplyAsync( new Supplier<Integer>() {
                    @Override
                    public Integer get() {
                        int t = param * 2;
                        System.out.println( "t2=" + t );
                        return t;
                    }
                } );
            }

        } );
        System.out.println( "thenCompose result : " + f.get() );
    }


    public static void main(String[] args) throws ExecutionException, InterruptedException {
        runAsync();
        supplyAsync();
        whenComplete();
        thenApply();
        handle();
        thenAccept();
        thenRun();
        thenCombine();
        thenAcceptBoth();
        applyToEither();
        acceptEither();
        runAfterEither();
        runAfterBoth();
        thenCompose();
    }
}