CompletableFuture用法介绍

一、CompletableFuture用法入门介绍

入门介绍的一个例子：

package com.cy.java8;

import java.util.Random;
import java.util.concurrent.CompletableFuture;

public class CompletableFutureInAction {
    private final static Random RANDOM = new Random(System.currentTimeMillis());

    public static void main(String[] args){
        CompletableFuture<Double> completableFuture = new CompletableFuture<>();
        new Thread(() -> {
            double value = get();
            completableFuture.complete(value);
        }).start();

        System.out.println("do other things...");

        completableFuture.whenComplete((t, e) -> {
            System.out.println("complete. value = "+ t);
            if(e != null){
                e.printStackTrace();
            }
        });
    }

    private static double get(){
        try {
            Thread.sleep(RANDOM.nextInt(3000));
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return RANDOM.nextDouble();
    }
}

console打印：

do other things...
complete. value = 0.8244376567363494

　

二、CompletableFuture.supplyAsync

CompletableFuture很少有直接new出来的方式去用的，一般都是通过提供的静态方法来使用。

1.使用CompletableFuture.supplyAsync来构造CompletableFuture：

package com.cy.java8;

import java.util.concurrent.*;

import static com.cy.java8.CompletableFutureInAction.get;

public class CompletableFutureInAction2 {

    public static void main(String[] args) {
        /**
         * 可以发现value=..没有被打印，为什么呢？
         * 因为此方法构造出来的线程是demon的，守护进程，main执行结束之后就消失了，所以
         * 根本没来得及执行whenComplete中的语句
         */
        CompletableFuture.supplyAsync(() -> get())
                .whenComplete((v, e) -> {
                    System.out.println("value = " + v);
                    if (e != null) {
                        e.printStackTrace();
                    }
                });

        System.out.println("do other things...");
    }


}

2.要将上面whenComplete中的语句执行，进行改造：

package com.cy.java8;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import static com.cy.java8.CompletableFutureInAction.get;

public class CompletableFutureInAction2 {

    public static void main(String[] args) throws InterruptedException {
        AtomicBoolean finished = new AtomicBoolean(false);

        CompletableFuture.supplyAsync(() -> get())
                .whenComplete((v, e) -> {
                    System.out.println("value = " + v);
                    if (e != null) {
                        e.printStackTrace();
                    }
                    finished.set(true);
                });

        System.out.println("do other things...");

        while(!finished.get()){
            Thread.sleep(1);
        }
    }


}

改写之后， main线程发现如果finished没有变为true就会一直等1ms，直到whenComplete执行将finished变为true。

3.上面的改写很low，其实只要将守护线程变为前台进程，main结束后不会消失就行了。

package com.cy.java8;

import java.util.concurrent.*;
import static com.cy.java8.CompletableFutureInAction.get;

public class CompletableFutureInAction2 {

    public static void main(String[] args){
        ExecutorService executorService = Executors.newFixedThreadPool(2, r -> {
            Thread t = new Thread(r);
            t.setDaemon(false);     //非守护线程
            return t;
        });

        CompletableFuture.supplyAsync(() -> get(), executorService)
                .whenComplete((v, e) -> {
                    System.out.println("value = " + v);
                    if (e != null) {
                        e.printStackTrace();
                    }
                });

        System.out.println("do other things...");

        //main执行结束之后，executorService线程不会结束，需要手动shutdown
        executorService.shutdown();
    }


}

三、thenApply：           　　　　

package com.cy.java8;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class CompletableFutureInAction3 {

    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(2, r -> {
            Thread t = new Thread(r);
            t.setDaemon(false);
            return t;
        });

        /**
         * 将执行完的结果再*100
         */
        CompletableFuture.supplyAsync(CompletableFutureInAction::get, executor)
                            .thenApply(v -> multiply(v))
                            .whenComplete((v, e) -> System.out.println(v));
    }

    private static double multiply(double value){
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return value * 100;
    }

}

console打印：

43.15351824222534

四、CompletableFuture.join()　　

package com.cy.java8;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;

public class CompletableFutureInAction3 {

    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(2, r -> {
            Thread t = new Thread(r);
            t.setDaemon(false);
            return t;
        });

        /**
         * 需求：将一组商品列表里面的每个商品对应的价格查询出来，并将这个价格*100.
         * 5个商品同时并发去做这件事
         *
         * CompletableFuture.join():等到所有的结果都执行结束，会返回CompletableFuture自己本身
         * 执行完的结果，等于get()返回的结果。
         */
        List<Integer> productionIDs = Arrays.asList(1, 2, 3, 4, 5);     //待查的一组商品列表的ID
        List<Double> priceList = productionIDs.stream().map(id -> CompletableFuture.supplyAsync(() -> queryProduction(id), executor))
                                                        .map(future -> future.thenApply(price -> multiply(price)))
                                                        .map(multiplyFuture -> multiplyFuture.join())
                                                        .collect(Collectors.toList());
        System.out.println(priceList);

        /**
         * 按照以前，要5个分别for循环去查询
         * 或者分多个线程去查询，再将每个线程查询的结果汇总，等到全部线程都执行完了，结果也就出来了
         */
    }

    private static double multiply(double value) {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return value * 100;
    }

    /**
     * 模拟 根据商品id查询对应的价格
     * @param id
     * @return
     */
    private static double queryProduction(int id){
        return CompletableFutureInAction.get();
    }
}

console打印：

[90.93730009374265, 23.65282935900653, 17.415066430776815, 16.605197824452343, 60.143109082288206]

 说明：这里多个任务同时执行，最终把结果汇总到一起 ，这种都是并行去执行的，编写代码也比较简洁，不需要考虑多线程之间的一些交互、锁、多线程之间的通信、控制，都不需要去关心。

五、CompletableFuture的常用API介绍　

supplyAsync
thenApply
whenComplete
handle
thenRun
thenAccept
thenCompose
thenCombine
thenAcceptBoth

runAfterBoth
applyToEither
acceptEither
runAfterEither
anyOf
allOf

1）supplyAsync、thenApply、whenComplete前面的代码已经介绍了。

package com.cy.java8;

import java.util.concurrent.CompletableFuture;

public class CompletableFutureInAction4 {
    public static void main(String[] args) throws InterruptedException {
        CompletableFuture.supplyAsync(() -> 1)
                .thenApply(v -> Integer.sum(v,10))
                .whenComplete((v, e) -> System.out.println(v));

        Thread.sleep(1000);
    }
}

2）whenCompleteAsync:    whenComplete是同步的方式，如果对于结果的处理是比较占时间的，不想通过这种同步的方式去做，可以用whenCompleteAsync进行异步操作。

3）handle：和thenApply差不多，只是多了一个对于异常的考虑。

package com.cy.java8;

import java.util.concurrent.CompletableFuture;

public class CompletableFutureInAction4 {
    public static void main(String[] args) throws InterruptedException {
        CompletableFuture.supplyAsync(() -> 1)
                .handle((v, e) -> Integer.sum(v, 10))
                .whenComplete((v, e) -> System.out.println(v));

        Thread.sleep(1000);
    }
}

4）thenRun：如果想在completableFuture整个执行结束之后，还想进行一个操作，可以thenRun(Runnable r)

package com.cy.java8;

import java.util.concurrent.CompletableFuture;

public class CompletableFutureInAction4 {
    public static void main(String[] args) throws InterruptedException {
        CompletableFuture.supplyAsync(() -> 1)
                .handle((v, e) -> Integer.sum(v, 10))
                .whenComplete((v, e) -> System.out.println(v))
                .thenRun(()-> System.out.println("thenRunning..."));

        Thread.sleep(1000);
    }
}

11
thenRunning...

5）thenAccept:   thenAccept(Consumer c)里面传的是consumer，对执行结果进行消费，不会对执行结果进行任何操作。　　

package com.cy.java8;

import java.util.concurrent.CompletableFuture;

public class CompletableFutureInAction4 {
    public static void main(String[] args) throws InterruptedException {
        CompletableFuture.supplyAsync(() -> 1)
                        .thenApply(v -> Integer.sum(v, 10))
                        .thenAccept(System.out::println);

        Thread.sleep(1000);
    }
}

11

6）thenCompose： 对执行结果再交给另外一个CompletableFuture，它再去对这个执行结果进行另外的计算。compose：组合，组合设计模式。

package com.cy.java8;

import java.util.concurrent.CompletableFuture;

public class CompletableFutureInAction4 {
    public static void main(String[] args) throws InterruptedException {
        CompletableFuture.supplyAsync(() -> 1)
                        .thenCompose(v -> CompletableFuture.supplyAsync(() -> v * 10))
                        .thenAccept(System.out::println);
        
        Thread.sleep(1000);
    }
}

10

7）thenCombine： thenCombine(CompletableFuture extends CompletionStage, BiFuntion)

　　　　　　　　  CompletableFuture的计算结果v1作为BiFunction的第1个入参，thenCombine中的第一个参数CompletableFuture的计算结果v2作为BiFunction的第2个入参，biFunction进行操作然后返回结果。

package com.cy.java8;

import java.util.concurrent.CompletableFuture;

public class CompletableFutureInAction4 {
    public static void main(String[] args) throws InterruptedException {
        CompletableFuture.supplyAsync(() -> 1)
                        .thenCombine(CompletableFuture.supplyAsync(() -> 2.0), (v1, v2) -> v1 + v2)
                        .thenAccept(System.out::println);
        
        Thread.sleep(1000);
    }
}

3.0

8）thenAcceptBoth： 和thenCombine差不多，只不过它的第二个参数是BiConsumer　

package com.cy.java8;

import java.util.concurrent.CompletableFuture;

public class CompletableFutureInAction4 {
    public static void main(String[] args) throws InterruptedException {
        CompletableFuture.supplyAsync(() -> 1)
                        .thenAcceptBoth(CompletableFuture.supplyAsync(() -> 2.0), (v1, v2) -> {
                            System.out.println("value=" + (v1 + v2));
                        });

        Thread.sleep(1000);
    }
}

value=3.0

9）runAfterBoth：两个CompletableFuture都执行结束之后，run

package com.cy.java8;

import java.util.concurrent.CompletableFuture;

public class CompletableFutureInAction5 {
    public static void main(String[] args) throws InterruptedException {

        CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + " is running");
            return 1;
        }).runAfterBoth(CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + " is running too");
            return 2;
        }), () -> System.out.println("both done"));

        Thread.sleep(1000);
    }
}

ForkJoinPool.commonPool-worker-9 is running
ForkJoinPool.commonPool-worker-9 is running too
both done

　　

10）applyToEither

applyToEither：两个CompletableFuture只要有1个执行完了，就将这个CompletableFuture交给Function。谁先执行完就将谁交给Function去执行　　

package com.cy.java8;

import java.util.concurrent.CompletableFuture;

public class CompletableFutureInAction5 {
    public static void main(String[] args) throws InterruptedException {

        CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(900);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("I am future 1");
            return 1;
        }).applyToEither(CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(50);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("I am future 2");
            return 2;
        }), v -> {
            System.out.println("value = " + v);
            return v * 10;
        }).thenAccept(System.out::println);


        Thread.currentThread().join();
    }
}

I am future 2
value = 2
20
I am future 1

　　

11）acceptEither 

acceptEither：acceptEither(CompletableFuture extends CompletionStage, Consumer), 两个CompletableFuture谁先执行完成，就将谁的结果交给consumer执行。

package com.cy.java8;

import java.util.concurrent.CompletableFuture;

public class CompletableFutureInAction5 {
    public static void main(String[] args) throws InterruptedException {

        CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(900);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("I am future 1");
            return 1;
        }).acceptEither(CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(50);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("I am future 2");
            return 2;
        }), v -> System.out.println("value = " + v));

        Thread.currentThread().join();
    }
}

I am future 2
value = 2
I am future 1

　　

12）runAfterEither　

 runAfterEither: runAfterEither(CompletionStage, Runnable),只要有一个CompletableFuture执行完了，就执行run

package com.cy.java8;

import java.util.concurrent.CompletableFuture;

public class CompletableFutureInAction5 {
    public static void main(String[] args) throws InterruptedException {

        CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(900);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("I am future 1");
            return 1;
        }).runAfterEither(CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(50);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("I am future 2");
            return 2;
        }), () -> System.out.println("done."));

        Thread.currentThread().join();
    }
}

I am future 2
done.
I am future 1

　　

13）allOf

allOf(CompletableFuture<?>... cfs)，返回值是CompletableFuture<Void>。要等所有的CompletableFuture都执行完成，才能执行下一步动作。

package com.cy.java8;

import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;

public class CompletableFutureInAction5 {
    private final static Random RANDOM = new Random(System.currentTimeMillis());

    public static void main(String[] args) throws InterruptedException {
        List<CompletableFuture<Double>> list = Arrays.asList(1, 2, 3, 4).stream()
                .map(i -> CompletableFuture.supplyAsync(CompletableFutureInAction5::get))
                .collect(Collectors.toList());

        //要等所有的CompletableFuture这些task执行完了，才会打印done.
        CompletableFuture.allOf(list.toArray(new CompletableFuture[list.size()]))
                .thenRun(() -> System.out.println("done."));

        Thread.currentThread().join();
    }

    static double get() {
        try {
            Thread.sleep(RANDOM.nextInt(3000));
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        double result = RANDOM.nextDouble();
        System.out.println(result);
        return result;
    }
}

0.6446554001163166
0.24435437709196395
0.18251850071600362
0.5261702037394511
done.

　　

14）anyOf

和allOf相反，只要有一个CompletableFuture执行完成，就会执行下一步动作

package com.cy.java8;

import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;

public class CompletableFutureInAction5 {
    private final static Random RANDOM = new Random(System.currentTimeMillis());

    public static void main(String[] args) throws InterruptedException {
        List<CompletableFuture<Double>> list = Arrays.asList(1, 2, 3, 4).stream()
                .map(i -> CompletableFuture.supplyAsync(CompletableFutureInAction5::get))
                .collect(Collectors.toList());

        //只要有一个CompletableFuture执行完了，就会打印done.
        CompletableFuture.anyOf(list.toArray(new CompletableFuture[list.size()]))
                .thenRun(() -> System.out.println("done."));

        Thread.currentThread().join();
    }

    static double get() {
        try {
            Thread.sleep(RANDOM.nextInt(3000));
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        double result = RANDOM.nextDouble();
        System.out.println(result);
        return result;
    }
}

0.1334361442807943
done.
0.6715112881360222
0.12945359790698785
0.1307762755130788

　　

----