Meningkatkan allOf dan anyOf di CompletableFuture

Halo lagi. Untuk mengantisipasi dimulainya kursus, "Pengembang Java" menyiapkan terjemahan materi yang bermanfaat untuk Anda.

---

Ada dua metode di CompletableFuture yang desainnya mengejutkan saya:

• CompletableFuture # allOf
• CompletableFuture # anyOf

Pada artikel ini kita akan melihat apa yang salah dengan mereka dan bagaimana mereka dapat dibuat lebih nyaman.

CompletableFuture # allOf

Mari kita lihat tanda tangan metode:

 public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs) { // ... } 

Setidaknya ada dua masalah yang diperdebatkan di sini:

1. Metode ini menerima beberapa objek CompletableFuture yang mengembalikan objek dari tipe yang berbeda.
2. Metode mengembalikan CompletableFuture , yang mengembalikan Void

Juga, beberapa mungkin tidak suka jumlah variabel parameter, jadi mari kita lihat bagian ini juga.

CompletableFuture<Void> sering digunakan sebagai sinyal untuk menyelesaikan operasi, namun, dengan membuat perubahan kecil pada API, metode ini dapat digunakan baik sebagai perangkat pensinyalan dan sebagai pembawa hasil semua operasi selesai . Mari kita coba melakukannya.

Asynchronous CompletableFuture # allOf

Pertama, mari kita datang dengan tanda tangan yang tepat.

Adalah wajar untuk mengasumsikan bahwa dalam kebanyakan kasus memproses daftar CompletableFuture homogen dan mengembalikan CompletableFuture berisi daftar hasil akan diperlukan:

 public static <T> CompletableFuture<List<T>> allOf( Collection<CompletableFuture<T>> futures) { // ... } 

Jeroan metode asli kemungkinan besar lebih kompleks daripada yang Anda harapkan:

 static CompletableFuture<Void> andTree( CompletableFuture<?>[] cfs, int lo, int hi) { CompletableFuture<Void> d = new CompletableFuture<Void>(); if (lo > hi) // empty d.result = NIL; else { CompletableFuture<?> a, b; int mid = (lo + hi) >>> 1; if ((a = (lo == mid ? cfs[lo] : andTree(cfs, lo, mid))) == null || (b = (lo == hi ? a : (hi == mid+1) ? cfs[hi] : andTree(cfs, mid+1, hi))) == null) throw new NullPointerException(); if (!d.biRelay(a, b)) { BiRelay<?,?> c = new BiRelay<>(d, a, b); a.bipush(b, c); c.tryFire(SYNC); } } return d; } 

Karena itu, alih-alih membuatnya dari awal, kami akan mencoba menggunakan kembali apa yang sudah ada dalam metode asli seolah-olah itu dimaksudkan untuk digunakan sebagai perangkat pensinyalan penyelesaian ... dan kemudian hanya mengubah hasil yang kosong ke daftar di masa mendatang:

 CompletableFuture<List<CompletableFuture<T>>> i = futures.stream() .collect(collectingAndThen( toList(), l -> CompletableFuture.allOf(l.toArray(new CompletableFuture[0])) .thenApply(__ -> l))); 

Sejauh ini bagus. Kami berhasil mendapatkannya
CompletableFuture<List<CompletableFuture<T> > > CompletableFuture<Void> bukan CompletableFuture<Void> , yang sudah baik. Tetapi kita tidak perlu daftar masa depan dengan hasil, kita perlu daftar hasil.

Sekarang kita bisa memproses daftar dan menghapus masa depan yang tidak diinginkan dari itu. Sangatlah normal untuk memanggil metode join CompletableFuture #, karena kita tahu bahwa mereka tidak akan pernah diblokir (pada titik ini, semua masa depan sudah selesai):

 CompletableFuture<List<T>> result = intermediate .thenApply(list -> list.stream() .map(CompletableFuture::join) .collect(toList())); 

Sekarang mari kita gabungkan semua ini menjadi solusi final:

 public static <T> CompletableFuture<List<T>> allOf( Collection<CompletableFuture<T>> futures) { return futures.stream() .collect(collectingAndThen( toList(), l -> CompletableFuture.allOf(l.toArray(new CompletableFuture[0])) .thenApply(__ -> l.stream() .map(CompletableFuture::join) .collect(Collectors.toList())))); } 

Asynchronous dan Jatuh CompletableFuture # allOf

Jika ada pengecualian, CompletableFuture # allOf yang asli menunggu semua operasi yang tersisa untuk diselesaikan.

Dan jika kita ingin melaporkan penyelesaian operasi ketika ada pengecualian di dalamnya, maka kita harus mengubah implementasinya.

Untuk melakukan ini, buat instance baru CompletableFuture dan hentikan secara manual setelah salah satu operasi menimbulkan pengecualian:

 CompletableFuture<List<T>> result = new CompletableFuture<>(); futures.forEach(f -> f .handle((__, ex) -> ex == null || result.completeExceptionally(ex))); 

... tapi kemudian kita harus berurusan dengan skenario ketika semua masa depan akan selesai dengan sukses. Ini dapat dengan mudah dilakukan dengan menggunakan metode allOf () yang ditingkatkan, dan kemudian cukup hentikan masa depan secara manual:

 allOf(futures).thenAccept(result::complete); 

Sekarang kita bisa menggabungkan semuanya untuk membentuk solusi akhir:

 public static <T> CompletableFuture<List<T>> allOfShortcircuiting(Collection<CompletableFuture<T>> futures) { CompletableFuture<List<T>> result = new CompletableFuture<>(); for (CompletableFuture<?> f : futures) { f.handle((__, ex) -> ex == null || result.completeExceptionally(ex)); } allOf(futures).thenAccept(result::complete); return result; } 

CompletableFuture # anyOf

Mari kita mulai dengan tanda tangan metode:

 public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs) { // ... } 

Kami dapat segera mendeteksi masalah yang sama dengan metode yang dibahas di atas:

1. Metode ini menerima beberapa objek CompletableFuture berisi objek dari tipe yang berbeda.
2. Metode mengembalikan CompletableFuture berisi objek bertipe Object .

Sejauh yang saya mengerti, metode CompletableFuture#allOf dirancang untuk digunakan sebagai perangkat pensinyalan. Tetapi CompletableFuture#anyOf tidak mengikuti filosofi ini, mengembalikan CompletableFuture<Object> , yang bahkan lebih membingungkan.

Lihatlah contoh berikut di mana saya mencoba memproses CompletableFuture yang berisi data dari berbagai jenis:

 CompletableFuture<Integer> f1 = CompletableFuture.completedFuture(1); CompletableFuture<String> f2 = CompletableFuture.completedFuture("2"); Integer result = CompletableFuture.anyOf(f1, f2) .thenApply(r -> { if (r instanceof Integer) { return (Integer) r; } else if (r instanceof String) { return Integer.valueOf((String) r); } throw new IllegalStateException("unexpected object type!"); }).join(); 

Cukup tidak nyaman, bukan?

Untungnya, ini cukup mudah untuk beradaptasi dengan skenario yang lebih umum (menunggu salah satu dari banyak nilai di masa depan dengan jenis yang sama) dengan mengubah tanda tangan dan memperkenalkan casting tipe langsung.

Dengan demikian, dengan peningkatan kami, kami dapat menggunakan kembali metode yang ada dan membawa hasilnya dengan aman:

 public static <T> CompletableFuture<T> anyOf(List<CompletableFuture<T>> cfs) { return CompletableFuture.anyOf(cfs.toArray(new CompletableFuture[0])) .thenApply(o -> (T) o); } public static <T> CompletableFuture<T> anyOf(CompletableFuture<T>... cfs) { return CompletableFuture.anyOf(cfs).thenApply(o -> (T) o); } 

Kode sumber

Kode sumber dapat ditemukan di Github .

Itu saja. Sampai jumpa di lapangan .