Bagaimanakah Java8 CompletableFuture melaksanakan pengaturcaraan berbilang benang tak segerak?-javaTutorial-php.cn

1. Contoh semakan Masa Depan

Dalam sesetengah senario perniagaan, kita perlu menggunakan berbilang benang untuk melaksanakan tugas secara tidak segerak untuk mempercepatkan pelaksanaan tugas.

JDK5 menambah antara muka Masa Depan baharu, yang digunakan untuk menerangkan hasil pengiraan tak segerak.

Walaupun kaedah penggunaan Masa Depan dan yang berkaitan menyediakan keupayaan untuk melaksanakan tugasan secara tidak segerak, adalah sangat menyusahkan untuk mendapatkan hasil Tugas Future.isDone selesai, dan kemudian dapatkan hasilnya.

Kedua-dua kaedah pemprosesan ini tidak begitu elegan Kod yang berkaitan adalah seperti berikut:

    @Test
    public void testFuture() throws ExecutionException, InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(5);
        Future<String> future = executorService.submit(() -> {
            Thread.sleep(2000);
            return "hello";
        });
        System.out.println(future.get());
        System.out.println("end");
    }

Salin selepas log masuk

Pada masa yang sama, Masa Depan tidak dapat menyelesaikan senario di mana pelbagai tugas tak segerak perlu bergantung antara satu sama lain. Secara ringkasnya, utas utama perlu menunggu tugasan sub-utas selesai sebelum melaksanakannya .

Dua Niaga Hadapan ditakrifkan di sini yang pertama mendapatkan maklumat pengguna melalui id pengguna, dan yang kedua memperoleh maklumat produk melalui id produk.

    @Test
    public void testCountDownLatch() throws InterruptedException, ExecutionException {
        ExecutorService executorService = Executors.newFixedThreadPool(5);
        CountDownLatch downLatch = new CountDownLatch(2);
        long startTime = System.currentTimeMillis();
        Future<String> userFuture = executorService.submit(() -> {
            //模拟查询商品耗时500毫秒
            Thread.sleep(500);
            downLatch.countDown();
            return "用户A";
        });
 
        Future<String> goodsFuture = executorService.submit(() -> {
            //模拟查询商品耗时500毫秒
            Thread.sleep(400);
            downLatch.countDown();
            return "商品A";
        });
 
        downLatch.await();
        //模拟主程序耗时时间
        Thread.sleep(600);
        System.out.println("获取用户信息:" + userFuture.get());
        System.out.println("获取商品信息:" + goodsFuture.get());
        System.out.println("总共用时" + (System.currentTimeMillis() - startTime) + "ms");
 
    }

Salin selepas log masuk

「Hasil jalankan」

Dapatkan maklumat pengguna: Pengguna A
Dapatkan maklumat produk: Produk A
Jumlah masa 1110ms

Ia boleh dilihat daripada keputusan berjalan bahawa keputusan telah diperolehi, dan jika kita tidak menggunakan operasi tak segerak, masa pelaksanaan hendaklah: 500+400+600 = 1500. Selepas menggunakan operasi tak segerak, ia sebenarnya hanya mengambil masa 1110.

Tetapi selepas Java8, saya tidak lagi fikir ini adalah penyelesaian yang elegan Seterusnya, mari belajar tentang penggunaan CompletableFuture.

2. Laksanakan contoh di atas melalui CompletableFuture

    @Test
    public void testCompletableInfo() throws InterruptedException, ExecutionException {
        long startTime = System.currentTimeMillis();
 
        //调用用户服务获取用户基本信息
        CompletableFuture<String> userFuture = CompletableFuture.supplyAsync(() ->
                //模拟查询商品耗时500毫秒
        {
            try {
                Thread.sleep(500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "用户A";
        });
 
        //调用商品服务获取商品基本信息
        CompletableFuture<String> goodsFuture = CompletableFuture.supplyAsync(() ->
                //模拟查询商品耗时500毫秒
        {
            try {
                Thread.sleep(400);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "商品A";
        });
 
        System.out.println("获取用户信息:" + userFuture.get());
        System.out.println("获取商品信息:" + goodsFuture.get());
 
        //模拟主程序耗时时间
        Thread.sleep(600);
        System.out.println("总共用时" + (System.currentTimeMillis() - startTime) + "ms");
    }

Salin selepas log masuk

Jalankan hasil

Dapatkan maklumat pengguna: Pengguna A
Dapatkan maklumat produk: Produk A
Jumlah Ia mengambil masa 1112ms

Fungsi CountDownLatch boleh dilaksanakan dengan mudah melalui CompletableFuture Anda fikir ini adalah penamat, tetapi ia jauh lebih baik daripada ini.

Sebagai contoh, ia boleh dilaksanakan: selepas tugasan 1 dilaksanakan, tugasan 2 dilaksanakan, malah hasil pelaksanaan tugasan 1 boleh digunakan sebagai parameter input tugas 2 dan fungsi berkuasa yang lain API CompletableFuture.

3. Kaedah penciptaan CompletableFuture

3.1. 4 kaedah penciptaan yang biasa digunakan

Terdapat empat kaedah statik dalam kod sumber CompletableFuture untuk melaksanakan tugas tak segerak

rreee

Secara amnya kami menggunakan kaedah statik di atas untuk mencipta CompletableFuture Di sini kami juga menerangkan perbezaannya:

「supplyAsync」Laksanakan tugas dan sokongan nilai pulangan.
「runAsync」Laksanakan tugas dan tiada nilai pulangan.

3.1.1, "supplyAsync method"

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){..}

Salin selepas log masuk

3.1.2, "runAsync method"

//使用默认内置线程池ForkJoinPool.commonPool()，根据supplier构建执行任务
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
//自定义线程，根据supplier构建执行任务
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)

Salin selepas log masuk

3.2, 4 cara untuk mendapatkan hasil < . )"

=> telah disediakan dalam Masa Depan, yang menyediakan pemprosesan tamat masa. Jika keputusan tidak diperoleh dalam masa yang ditentukan, pengecualian tamat masa akan dibuang

=> Dapatkan keputusan dengan segera tanpa menyekat Jika pengiraan hasil selesai, keputusan akan dikembalikan atau pengecualian berlaku semasa proses pengiraan tidak selesai, nilai set IfAbsent akan dikembalikan
=> Tiada pengecualian akan dilemparkan dalam kaedah

//使用默认内置线程池ForkJoinPool.commonPool()，根据runnable构建执行任务
public static CompletableFuture<Void> runAsync(Runnable runnable) 
//自定义线程，根据runnable构建执行任务
public static CompletableFuture<Void> runAsync(Runnable runnable,  Executor executor)

Salin selepas log masuk

「Run result ":

"Produk B"

, kerana hasilnya tidak dapat diperolehi serta merta kerana terpaksa tidur selama 1 saat 示例

Kaedah gabungan tidak akan membuang pengecualian dalam kaedah hasil, tetapi hasil pelaksanaan akan membuang pengecualian Pengecualian yang dilemparkan ialah CompletionException

Kaedah get akan membuang pengecualian dalam kaedah hasil , pengecualian yang dilemparkan oleh hasil pelaksanaan ialah ExecutionException
4 kaedah panggil balik tak segerak 🎜>
4.1, thenRun/thenRunAsync
Ringkasnya,
"Selepas menyelesaikan tugasan pertama, lakukan tugas kedua, dan tugas kedua tidak akan mengembalikan nilai"
.

Contoh

//方式一
public T get()
//方式二
public T get(long timeout, TimeUnit unit)
//方式三
public T getNow(T valueIfAbsent)
//方式四
public T join()

Salin selepas log masuk

Bagaimanakah Java8 CompletableFuture melaksanakan pengaturcaraan berbilang benang tak segerak?

"Apakah perbezaan antara thenRun dan thenRunAsync?"

Jika anda melaksanakan tugas pertama Bila kaedah thenRun dipanggil untuk melaksanakan tugas kedua, tugas kedua dan tugas pertama berkongsi kumpulan benang yang sama.

Apabila anda memanggil thenRunAsync untuk melaksanakan tugas kedua, tugas pertama menggunakan kumpulan benang yang anda lalui dan tugas kedua menggunakan kumpulan benang ForkJoin.

: ThenAccept and thenAcceptAsync, thenApply and thenApplyAsync, dsb. diperkenalkan kemudian, perbezaan antara mereka juga ini.

4.2, thenAccept/thenAcceptAsync

Selepas tugasan pertama dilaksanakan, tugas kaedah panggil balik kedua dilaksanakan

, dihantar kepada kaedah panggil balik, tetapi kaedah panggil balik tidak mempunyai nilai pulangan.

    @Test
    public void testCompletableThenAccept() throws ExecutionException, InterruptedException {
        long startTime = System.currentTimeMillis();
        CompletableFuture<String> cp1 = CompletableFuture.supplyAsync(() -> {
            return "dev";
 
        });
        CompletableFuture<Void> cp2 = cp1.thenAccept((a) -> {
            System.out.println("上一个任务的返回结果为: " + a);
        });
 
        cp2.get();
    }

Salin selepas log masuk

4.3、 thenApply/thenApplyAsync

表示第一个任务执行完成后，执行第二个回调方法任务，会将该任务的执行结果，作为入参，传递到回调方法中，并且回调方法是有返回值的。

示例

    @Test
    public void testCompletableThenApply() throws ExecutionException, InterruptedException {
        CompletableFuture<String> cp1 = CompletableFuture.supplyAsync(() -> {
            return "dev";
 
        }).thenApply((a) -> {
            if (Objects.equals(a, "dev")) {
                return "dev";
            }
            return "prod";
        });
 
        System.out.println("当前环境为:" + cp1.get());
 
        //输出: 当前环境为:dev
    }

Salin selepas log masuk

5、异常回调

当CompletableFuture的任务不论是正常完成还是出现异常它都会调用「whenComplete」这回调函数。

「正常完成」：whenComplete返回结果和上级任务一致，异常为null；
「出现异常」：whenComplete返回结果为null，异常为上级任务的异常；

即调用get()时，正常完成时就获取到结果，出现异常时就会抛出异常，需要你处理该异常。

下面来看看示例

5.1、只用whenComplete

    @Test
    public void testCompletableWhenComplete() throws ExecutionException, InterruptedException {
        CompletableFuture<Double> future = CompletableFuture.supplyAsync(() -> {
 
            if (Math.random() < 0.5) {
                throw new RuntimeException("出错了");
            }
            System.out.println("正常结束");
            return 0.11;
 
        }).whenComplete((aDouble, throwable) -> {
            if (aDouble == null) {
                System.out.println("whenComplete aDouble is null");
            } else {
                System.out.println("whenComplete aDouble is " + aDouble);
            }
            if (throwable == null) {
                System.out.println("whenComplete throwable is null");
            } else {
                System.out.println("whenComplete throwable is " + throwable.getMessage());
            }
        });
        System.out.println("最终返回的结果 = " + future.get());
    }

Salin selepas log masuk

正常完成，没有异常时：

正常结束
whenComplete aDouble is 0.11
whenComplete throwable is null
最终返回的结果 = 0.11

出现异常时：get()会抛出异常

whenComplete aDouble is null
whenComplete throwable is java.lang.RuntimeException: 出错了

java.util.concurrent.ExecutionException: java.lang.RuntimeException: 出错了
at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357)
at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1895)

5.2、whenComplete + exceptionally示例

    @Test
    public void testWhenCompleteExceptionally() throws ExecutionException, InterruptedException {
        CompletableFuture<Double> future = CompletableFuture.supplyAsync(() -> {
            if (Math.random() < 0.5) {
                throw new RuntimeException("出错了");
            }
            System.out.println("正常结束");
            return 0.11;
 
        }).whenComplete((aDouble, throwable) -> {
            if (aDouble == null) {
                System.out.println("whenComplete aDouble is null");
            } else {
                System.out.println("whenComplete aDouble is " + aDouble);
            }
            if (throwable == null) {
                System.out.println("whenComplete throwable is null");
            } else {
                System.out.println("whenComplete throwable is " + throwable.getMessage());
            }
        }).exceptionally((throwable) -> {
            System.out.println("exceptionally中异常：" + throwable.getMessage());
            return 0.0;
        });
 
        System.out.println("最终返回的结果 = " + future.get());
    }

Salin selepas log masuk

当出现异常时，exceptionally中会捕获该异常，给出默认返回值0.0。

whenComplete aDouble is null
whenComplete throwable is java.lang.RuntimeException: 出错了
exceptionally中异常：java.lang.RuntimeException: 出错了
最终返回的结果 = 0.0

6、多任务组合回调

Bagaimanakah Java8 CompletableFuture melaksanakan pengaturcaraan berbilang benang tak segerak?

6.1、AND组合关系

thenCombine / thenAcceptBoth / runAfterBoth都表示：「当任务一和任务二都完成再执行任务三」。

区别在于：

「runAfterBoth」 不会把执行结果当做方法入参，且没有返回值
「thenAcceptBoth」: 会将两个任务的执行结果作为方法入参，传递到指定方法中，且无返回值
「thenCombine」：会将两个任务的执行结果作为方法入参，传递到指定方法中，且有返回值

示例

    @Test
    public void testCompletableThenCombine() throws ExecutionException, InterruptedException {
        //创建线程池
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        //开启异步任务1
        CompletableFuture<Integer> task = CompletableFuture.supplyAsync(() -> {
            System.out.println("异步任务1，当前线程是：" + Thread.currentThread().getId());
            int result = 1 + 1;
            System.out.println("异步任务1结束");
            return result;
        }, executorService);
 
        //开启异步任务2
        CompletableFuture<Integer> task2 = CompletableFuture.supplyAsync(() -> {
            System.out.println("异步任务2，当前线程是：" + Thread.currentThread().getId());
            int result = 1 + 1;
            System.out.println("异步任务2结束");
            return result;
        }, executorService);
 
        //任务组合
        CompletableFuture<Integer> task3 = task.thenCombineAsync(task2, (f1, f2) -> {
            System.out.println("执行任务3，当前线程是：" + Thread.currentThread().getId());
            System.out.println("任务1返回值：" + f1);
            System.out.println("任务2返回值：" + f2);
            return f1 + f2;
        }, executorService);
 
        Integer res = task3.get();
        System.out.println("最终结果：" + res);
    }

Salin selepas log masuk

「运行结果」

异步任务1，当前线程是：17
异步任务1结束
异步任务2，当前线程是：18
异步任务2结束
执行任务3，当前线程是：19
任务1返回值：2
任务2返回值：2
最终结果：4

6.2、OR组合关系

applyToEither / acceptEither / runAfterEither 都表示：「两个任务，只要有一个任务完成，就执行任务三」。

区别在于：

「runAfterEither」：不会把执行结果当做方法入参，且没有返回值
「acceptEither」: 会将已经执行完成的任务，作为方法入参，传递到指定方法中，且无返回值
「applyToEither」：会将已经执行完成的任务，作为方法入参，传递到指定方法中，且有返回值

示例

    @Test
    public void testCompletableEitherAsync() {
        //创建线程池
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        //开启异步任务1
        CompletableFuture<Integer> task = CompletableFuture.supplyAsync(() -> {
            System.out.println("异步任务1，当前线程是：" + Thread.currentThread().getId());
 
            int result = 1 + 1;
            System.out.println("异步任务1结束");
            return result;
        }, executorService);
 
        //开启异步任务2
        CompletableFuture<Integer> task2 = CompletableFuture.supplyAsync(() -> {
            System.out.println("异步任务2，当前线程是：" + Thread.currentThread().getId());
            int result = 1 + 2;
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("异步任务2结束");
            return result;
        }, executorService);
 
        //任务组合
        task.acceptEitherAsync(task2, (res) -> {
            System.out.println("执行任务3，当前线程是：" + Thread.currentThread().getId());
            System.out.println("上一个任务的结果为：" + res);
        }, executorService);
    }

Salin selepas log masuk

运行结果

//通过结果可以看出，异步任务2都没有执行结束，任务3获取的也是1的执行结果
异步任务1，当前线程是：17
异步任务1结束
异步任务2，当前线程是：18
执行任务3，当前线程是：19
上一个任务的结果为：2

注意

如果把上面的核心线程数改为1也就是

ExecutorService executorService = Executors.newFixedThreadPool(1);

运行结果就是下面的了，会发现根本没有执行任务3，显然是任务3直接被丢弃了。

异步任务1，当前线程是：17
异步任务1结束
异步任务2，当前线程是：17

6.3、多任务组合

「allOf」：等待所有任务完成
「anyOf」：只要有一个任务完成

示例

allOf：等待所有任务完成

    @Test
    public void testCompletableAallOf() throws ExecutionException, InterruptedException {
        //创建线程池
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        //开启异步任务1
        CompletableFuture<Integer> task = CompletableFuture.supplyAsync(() -> {
            System.out.println("异步任务1，当前线程是：" + Thread.currentThread().getId());
            int result = 1 + 1;
            System.out.println("异步任务1结束");
            return result;
        }, executorService);
 
        //开启异步任务2
        CompletableFuture<Integer> task2 = CompletableFuture.supplyAsync(() -> {
            System.out.println("异步任务2，当前线程是：" + Thread.currentThread().getId());
            int result = 1 + 2;
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("异步任务2结束");
            return result;
        }, executorService);
 
        //开启异步任务3
        CompletableFuture<Integer> task3 = CompletableFuture.supplyAsync(() -> {
            System.out.println("异步任务3，当前线程是：" + Thread.currentThread().getId());
            int result = 1 + 3;
            try {
                Thread.sleep(4000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("异步任务3结束");
            return result;
        }, executorService);
 
        //任务组合
        CompletableFuture<Void> allOf = CompletableFuture.allOf(task, task2, task3);
 
        //等待所有任务完成
        allOf.get();
        //获取任务的返回结果
        System.out.println("task结果为：" + task.get());
        System.out.println("task2结果为：" + task2.get());
        System.out.println("task3结果为：" + task3.get());
    }

Salin selepas log masuk

anyOf: 只要有一个任务完成

    @Test
    public void testCompletableAnyOf() throws ExecutionException, InterruptedException {
        //创建线程池
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        //开启异步任务1
        CompletableFuture<Integer> task = CompletableFuture.supplyAsync(() -> {
            int result = 1 + 1;
            return result;
        }, executorService);
 
        //开启异步任务2
        CompletableFuture<Integer> task2 = CompletableFuture.supplyAsync(() -> {
            int result = 1 + 2;
            return result;
        }, executorService);
 
        //开启异步任务3
        CompletableFuture<Integer> task3 = CompletableFuture.supplyAsync(() -> {
            int result = 1 + 3;
            return result;
        }, executorService);
 
        //任务组合
        CompletableFuture<Object> anyOf = CompletableFuture.anyOf(task, task2, task3);
        //只要有一个有任务完成
        Object o = anyOf.get();
        System.out.println("完成的任务的结果：" + o);
    }

Salin selepas log masuk

7、CompletableFuture使用有哪些注意点

Bagaimanakah Java8 CompletableFuture melaksanakan pengaturcaraan berbilang benang tak segerak?

CompletableFuture 使我们的异步编程更加便利的、代码更加优雅的同时，我们也要关注下它，使用的一些注意点。

7.1、Future需要获取返回值，才能获取异常信息

    @Test
    public void testWhenCompleteExceptionally() {
        CompletableFuture<Double> future = CompletableFuture.supplyAsync(() -> {
            if (1 == 1) {
                throw new RuntimeException("出错了");
            }
            return 0.11;
        });
 
        //如果不加 get()方法这一行，看不到异常信息
        //future.get();
    }

Salin selepas log masuk

Future需要获取返回值，才能获取到异常信息。如果不加 get()/join()方法，看不到异常信息。

小伙伴们使用的时候，注意一下哈,考虑是否加try...catch...或者使用exceptionally方法。

7.2、CompletableFuture的get()方法是阻塞的

CompletableFuture的get()方法是阻塞的，如果使用它来获取异步调用的返回值，需要添加超时时间。

//反例
 CompletableFuture.get();
//正例
CompletableFuture.get(5, TimeUnit.SECONDS);

Salin selepas log masuk

7.3、不建议使用默认线程池

CompletableFuture代码中又使用了默认的「ForkJoin线程池」，处理的线程个数是电脑「CPU核数-1」。在大量请求过来的时候，处理逻辑复杂的话，响应会很慢。一般建议使用自定义线程池，优化线程池配置参数。

7.4、自定义线程池时，注意饱和策略

CompletableFuture的get()方法是阻塞的，我们一般建议使用future.get(5, TimeUnit.SECONDS)。并且一般建议使用自定义线程池。

但是如果线程池拒绝策略是DiscardPolicy或者DiscardOldestPolicy，当线程池饱和时，会直接丢弃任务，不会抛弃异常。因此建议，CompletableFuture线程池策略最好使用AbortPolicy，然后耗时的异步线程，做好线程池隔离哈。

Atas ialah kandungan terperinci Bagaimanakah Java8 CompletableFuture melaksanakan pengaturcaraan berbilang benang tak segerak?. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!