Cara menggunakan alat Java multi-threading CompletableFuture

Prakata

Masalah dengan Masa Depan

Apabila menulis program berbilang benang, anda boleh menggunakan Masa Depan untuk mendapatkan hasil daripada tak segerak Walau bagaimanapun, anda akan menemui beberapa masalah semasa penggunaan:

• Jika anda ingin melakukan operasi selanjutnya pada hasil Masa Depan, anda perlu menyekat utas semasa

• Pelbagai Niaga Hadapan tidak boleh dilaksanakan dalam rantaian Hasil daripada setiap Masa Depan adalah bebas Ia dijangka melakukan satu lagi perkara yang tidak segerak kepada hasil daripada Masa Depan Tiada strategi pengendalian pengecualian , jika pelaksanaan Masa Depan gagal, anda perlu menangkap secara manual

• CompletableFuture wujud

Untuk menyelesaikan masalah Masa Depan, JDK menyediakan kami dengan satu yang berguna dalam kelas 1.8 Alat CompletableFuture;

Ia melaksanakan antara muka Masa Depan dan Tahap Selesai dan menyediakan kaedah pemprosesan yang sepadan untuk kekurangan Masa Depan.

Logik pemprosesan baharu kami boleh dipanggil semula secara automatik selepas urutan tak segerak dilaksanakan tanpa menyekat

• Boleh mengatur berbilang tugasan tak segerak pengisihan

• Pengendalian pengecualian

• Idea teras CompletableFuture ialah setiap tugas tak segerak boleh dianggap sebagai langkah (CompletionStage), dan maka tugas tak segerak lain boleh melakukan perkara yang mereka mahu lakukan berdasarkan langkah ini.

CompletionStage mentakrifkan banyak kaedah pemprosesan langkah, yang sangat berkuasa Berikut adalah beberapa kaedah yang biasa digunakan dalam kehidupan seharian untuk rujukan anda.

Penggunaan

Penggunaan asas - Serahkan tugas tak segerak

Penggunaan mudah

Pelaksanaan tak segerak, tiada keputusan diperlukan:

rreee

Segerak dan Segerak kembalikan hasil pada masa yang sama:

// 可以执行Executors异步执行，如果不指定，默认使用ForkJoinPool
CompletableFuture.runAsync(() -> System.out.println("Hello CompletableFuture!"));

Proses hasil tugas tak segerak sebelumnya

kemudianJalankan: Tidak perlu hasil daripada langkah sebelumnya, lakukan terus operasi baharu

• kemudianTerima: Dapatkan kandungan pemprosesan tak segerak sebelumnya dan lakukan operasi baharu

• kemudianGunakan: Dapatkan kandungan langkah sebelumnya dan kemudian jana kandungan baharu

• Apa-apa sahaja dengan akhiran Async bermakna operasi pemprosesan baharu masih tidak segerak. Operasi Async boleh menentukan Pelaksana untuk pemprosesan

// 同样可以指定线程池
CompletableFuture<String> stringCompletableFuture = CompletableFuture.supplyAsync(() -> "Hello CompletableFuture!");
System.out.println(stringCompletableFuture.get());

Pilih dua hasil -terima Sama ada

Apabila kami mempunyai dua pemprosesan panggilan balik Apabila , sebarang penyelesaian boleh digunakan, dan tiada hubungan antara kedua-dua hasil, kemudian gunakan acceptEither.

Sesiapa yang melengkapkan pelaksanaan dua utas tak segerak dahulu akan menggunakan hasil

Perkara yang sama berlaku untuk jenis kaedah lain.

// Demo
       CompletableFuture
                .supplyAsync(() -> "Hello CompletableFuture!")
                // 针对上一步的结果做处理，产生新的结果
                .thenApplyAsync(s -> s.toUpperCase())
                // 针对上一步的结果做处理，不返回结果
                .thenAcceptAsync(s -> System.out.println(s))
                // 不需要上一步返回的结果，直接进行操作
                .thenRunAsync(() -> System.out.println("end"));
        ;

Gabungkan kedua-dua keputusan -kemudianGabungkan, kemudianTerimaKedua-duanya

kemudianGabungkan

Apabila kita mempunyai dua Tahap Penyiapan, kita perlu menyepadukan keputusan kedua-duanya dan kemudian mengira hasil baharu.

kemudianKarang memproses hasil Tahap Penyusunan sebelumnya dan mengembalikan hasilnya, dan jenis pemulangan mestilah Tahap Penyusunan.

• thenCombine mendapat keputusan CompletionStage pertama, kemudian mendapat CompletionStage semasa, dan memproses keputusan kedua-duanya.

• // 返回abc
  CompletableFuture
                  .supplyAsync(() -> {
                      SleepUtils.sleep(100);
                      return "Hello CompletableFuture!";
                  })
                  .acceptEither(CompletableFuture.supplyAsync(() -> "abc"), new Consumer<String>() {
                      @Override
                      public void accept(String s) {
                          System.out.println(s);
                      }
                  });
  // 返回Hello CompletableFuture!       
  CompletableFuture
                  .supplyAsync(() -> "Hello CompletableFuture!")
                  .acceptEither(CompletableFuture.supplyAsync(() -> {
                      SleepUtils.sleep(100);
                      return "abc";
                  }), new Consumer<String>() {
                      @Override
                      public void accept(String s) {
                          System.out.println(s);
                      }
                  });

  thenAcceptBoth

memerlukan hasil dua asynchronous CompletableFutures Apabila kedua-duanya selesai, thenAcceptBoth callback dimasukkan.

        CompletableFuture<Integer> heightAsync = CompletableFuture.supplyAsync(() -> 172);

        CompletableFuture<Double> weightAsync = CompletableFuture.supplyAsync(() -> 65)
                .thenCombine(heightAsync, new BiFunction<Integer, Integer, Double>() {
                    @Override
                    public Double apply(Integer wight, Integer height) {
                        return wight * 10000.0 / (height * height);
                    }
                })
                ;

Pengendalian pengecualian

Apabila kami menggunakan CompleteFuture untuk membuat panggilan berantai, dalam berbilang panggilan balik tak segerak, Jika terdapat masalah dengan satu pelaksanaan, semua panggilan balik berikutnya akan berhenti, jadi strategi pengendalian pengecualian diperlukan.

sangat luar biasa

secara luar biasa memberi kita peluang untuk memulihkan dan menyesuaikan kandungan pemulangan apabila ralat berlaku.

// thenAcceptBoth案例：
        CompletableFuture
                .supplyAsync(() -> "Hello CompletableFuture!")
                .thenAcceptBoth(CompletableFuture.supplyAsync(() -> "abc"), new BiConsumer<String, String>() {
                		// 参数一为我们刚开始运行时的CompletableStage，新传入的作为第二个参数
                    @Override
                    public void accept(String s, String s2) {
                        System.out.println("param1=" + s + ", param2=" + s2);
                    }
                });
// 结果：param1=Hello CompletableFuture!, param2=abc

handle

secara luar biasa hanya akan dilaksanakan apabila pengecualian berlaku, manakala pemegang akan dilaksanakan tanpa mengira sama ada ralat berlaku.

        CompletableFuture.supplyAsync(() -> {
            throw new RuntimeException("发生错误");
        }).exceptionally(throwable -> {
            log.error("调用错误 {}", throwable.getMessage(), throwable);
            return "异常处理内容";
        });

Kes

Sebilangan besar pengguna menghantar mesej teks|Mesej

Keperluan adalah untuk memberitahu pengguna dengan syarat tertentu dalam jadual melalui mesej teks, tetapi terdapat berjuta-juta daripada pengguna mesej teks , jika bacaan berbenang tunggal digunakan, kecekapan membaca akan menjadi sangat perlahan. Pada masa ini, anda boleh mempertimbangkan untuk menggunakan pelbagai benang untuk membaca; threading untuk bacaan kelompok

CompletableFuture.supplyAsync(() -> {
    return "abc";
})
.handle((r,err) -> {
    log.error("调用错误 {}", err.getMessage(), err);
    // 对结果做额外的处理
    return r;
})
;

3. Lakukan pemprosesan logik perniagaan, atau anda juga boleh melakukan pemprosesan logik perniagaan secara langsung selepas membaca;

并发获取商品不同信息

在系统拆分比较细的时候，价格，优惠券，库存，商品详情等信息分散在不同的系统中，有时候需要同时获取商品的所有信息， 有时候可能只需要获取商品的部分信息。

当然问题点在于要调用多个不同的系统，需要将RT降低下来，那么需要进行并发调用；

     List<Task> taskList = new ArrayList<>();
        List<Object> result = taskList.stream()
                .map(task -> CompletableFuture.supplyAsync(()->{
//                    handlerMap.get(task).query();
                    return "";
                }, executorService))
                .map(c -> c.join())
                .collect(Collectors.toList());

问题

thenRun和thenRunAsync有什么区别

• 如果不使用传入的线程池，大家用默认的线程池ForkJoinPool

• thenRun用的默认和上一个任务使用相同的线程池

• thenRunAsync在执行新的任务的时候可以接受传入一个新的线程池，使用新的线程池执行任务；

handle和exceptional有什么区别

exceptionally是只有发生异常时才会执行，而handle则是不管是否发生错误都会执行。

Atas ialah kandungan terperinci Cara menggunakan alat Java multi-threading CompletableFuture. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!