أدوات التزامن المتقدّمة

تركيب CompletableFuture

15 دقيقة الدرس 6 من 13

تركيب CompletableFuture

في الدرس السابق تعرّفت على أساسيات CompletableFuture: كيف تبدأ عملية حسابية غير متزامنة بـ supplyAsync وتحوّل نتيجتها بـ thenApply. غير أن البرامج الحقيقية نادرًا ما تحتوي على خطوة غير متزامنة واحدة معزولة. في أغلب الأحيان ستحتاج إلى ربط استدعاء غير متزامن بآخر، أو دمج نتيجتين من استدعاءين مستقلّين، أو انتظار مجموعة من الاستدعاءات قبل المتابعة. هذا هو ما تعنيه عملية التركيب — وإتقانها هو ما يميّز الكود غير المتزامن النظيف والصحيح عن الكود الذي يُعلِق الخيوط بصمت أو يبتلع الاستثناءات.

thenCompose — ربط الخطوات غير المتزامنة

thenCompose هو المقابل غير المتزامن لـ flatMap في التدفقات: يأخذ نتيجة CompletableFuture ويمرّرها إلى دالة تُعيد نسخة أخرى من CompletableFuture، ثم يُسطّح المرحلتين في مرحلة واحدة.

لماذا لا نستخدم thenApply؟ إذا أعادت دالة التحويل نفسها CompletableFuture فإن thenApply ستُعطيك CompletableFuture<CompletableFuture<T>> — مستقبل متداخل عديم الفائدة حتى تفكّه يدويًا. يقوم thenCompose بهذا الفكّ تلقائيًا.

import java.util.concurrent.CompletableFuture;

// محاكاة جلب معرّف المستخدم من خدمة بعيدة
CompletableFuture<String> fetchUserId() {
    return CompletableFuture.supplyAsync(() -> {
        // مثال: استدعاء HTTP
        return "user-42";
    });
}

// محاكاة جلب الملف الشخصي بمعرّف المستخدم
CompletableFuture<String> fetchProfile(String userId) {
    return CompletableFuture.supplyAsync(() -> {
        return "Profile of " + userId;
    });
}

// خطأ: يُعطي CompletableFuture<CompletableFuture<String>>
// var nested = fetchUserId().thenApply(id -> fetchProfile(id));

// صحيح: خطّ أنابيب مُسطَّح
CompletableFuture<String> profile = fetchUserId()
    .thenCompose(id -> fetchProfile(id));

System.out.println(profile.join()); // Profile of user-42

سياق التنفيذ مهم. عند استخدام thenCompose (بدون لاحقة Async) يعمل الكود التالي على نفس خيط المرحلة السابقة — غالبًا خيط من مجموعة fork-join المشتركة. استخدم thenComposeAsync مع Executor صريح حين تريد التحكم في مجموعة الخيوط التالية، مثلًا لإبقاء I/O المانع بعيدًا عن المجموعة المشتركة.

thenCombine — دمج مستقبلَين مستقلَّين

حين لا تعتمد عمليتان غير متزامنتان على بعضهما يجب تشغيلهما معًا ودمج نتيجتيهما فقط بعد اكتمالهما. هذا بالضبط ما يفعله thenCombine: يأخذ CompletableFuture ثانيًا وBiFunction، وينتظر اكتمال كليهما، ثم يدمج النتيجتين.

import java.util.concurrent.CompletableFuture;

CompletableFuture<Double> fetchUsdRate = CompletableFuture.supplyAsync(() -> {
    // محاكاة استدعاء صرف العملة
    return 3.75; // 1 دولار = 3.75 ريال (مثال)
});

CompletableFuture<Integer> fetchQuantity = CompletableFuture.supplyAsync(() -> {
    // محاكاة استعلام قاعدة بيانات
    return 120; // وحدات في المخزون
});

// كلا الجلبَين يعملان بالتوازي؛ thenCombine ينتظر كليهما
CompletableFuture<String> summary = fetchUsdRate
    .thenCombine(fetchQuantity, (rate, qty) ->
        "Total SAR value: " + (rate * qty));

System.out.println(summary.join()); // Total SAR value: 450.0

لا تسلسل ما يمكن تشغيله بالتوازي. لو كتبت fetchUsdRate().thenCompose(_ -> fetchQuantity()) لتسلسلت استدعاءين مستقلّين وأهدرت وقت المعالج. استخدم thenCombine أو allOf ليبدأ كلاهما فورًا وتدفع فقط تكلفة الأبطأ منهما.

allOf — انتظار عدّة مستقبلات

يُعيد CompletableFuture.allOf(futures...) نسخة من CompletableFuture<Void> تكتمل حين تكتمل كل المستقبلات المُمرَّرة. ولأن نوع النتيجة Void، عادةً ما تجلب النتائج الفردية بنفسك بعد اكتمال allOf.

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

List<String> urls = List.of(
    "https://api.example.com/a",
    "https://api.example.com/b",
    "https://api.example.com/c"
);

// تشغيل كل الجلبات بالتوازي
List<CompletableFuture<String>> fetches = urls.stream()
    .map(url -> CompletableFuture.supplyAsync(() -> fetch(url)))
    .collect(Collectors.toList());

// انتظار اكتمالها جميعًا
CompletableFuture<Void> allDone =
    CompletableFuture.allOf(fetches.toArray(new CompletableFuture[0]));

// بعد اكتمال allOf، اجمع النتائج (join() آمن هنا — قد اكتملت بالفعل)
List<String> results = allDone.thenApply(__ ->
    fetches.stream()
           .map(CompletableFuture::join)
           .collect(Collectors.toList())
).join();

results.forEach(System.out::println);

allOf لا تُوصّل استثناءات المهام الفردية تلقائيًا. إذا فشل أحد المستقبلات، يكتمل allOf نفسه بشكل استثنائي، لكن استدعاء join() على المستقبل الفاشل هو الطريقة للحصول على السبب الفعلي. عالج الاستثناءات على كل مستقبل فردي إذا احتجت تقريرًا مفصّلًا بالأخطاء.

معالجة الاستثناءات في خطوط الأنابيب المركّبة

تسري الاستثناءات عبر سلسلة CompletableFuture كـمرحلة فاشلة. كل مرحلة تالية لم تُسجّل معالجًا للأخطاء تُتجاوز، ويظهر الاستثناء عند استدعاء get() أو join() النهائي مُغلَّفًا في CompletionException. لديك ثلاث استراتيجيات للتعافي:

exceptionally(fn) — يعمل فقط عند فشل المرحلة؛ يستبدل الاستثناء بقيمة افتراضية.
handle(BiFunction) — يعمل سواء نجحت المرحلة أم فشلت؛ يتيح فحص النتيجة والاستثناء معًا في مكان واحد.
whenComplete(BiConsumer) — مثل handle لكنه لا يحوّل النتيجة؛ مفيد للتسجيل أو التنظيف.

CompletableFuture<String> riskyFetch = CompletableFuture
    .supplyAsync(() -> {
        if (Math.random() < 0.5) throw new RuntimeException("Network timeout");
        return "fresh data";
    });

// الاستراتيجية 1 — التعافي بقيمة احتياطية
String result1 = riskyFetch
    .exceptionally(ex -> "cached fallback")
    .join();

// الاستراتيجية 2 — فحص الجانبين معًا
String result2 = riskyFetch
    .handle((data, ex) -> {
        if (ex != null) {
            System.err.println("Failed: " + ex.getMessage());
            return "default";
        }
        return data;
    })
    .join();

System.out.println(result1);
System.out.println(result2);

التركيب ومعالجة الاستثناءات معًا. حين تمتلك سلسلة thenCompose وإحدى الخطوات قد تفشل، ألصق exceptionally أو handle عند النقطة التي يمكنك فيها التعافي بمعنى — عادةً مباشرةً بعد المرحلة التي قد تفشل، وليس فقط في النهاية. المعالج في النهاية يلتقط كل شيء لكنه يمنحك سياقًا أقل عن أي خطوة فشلت.

الجمع في نموذج واقعي

المثال التالي يجمع thenCompose وthenCombine وexceptionally في خط أنابيب واحد يُحضر معرّف المنتج بشكل غير متزامن، ثم يحمّل التفاصيل والسعر الحي بالتوازي، ويتعافى بأناقة عند أي فشل:

import java.util.concurrent.CompletableFuture;

CompletableFuture<String> pipeline = CompletableFuture
    // الخطوة 1 — تحويل اسم المنتج إلى معرّف رقمي
    .supplyAsync(() -> resolveProductId("java-book"))

    // الخطوة 2 — جلب التفاصيل (اعتماد تسلسلي على الخطوة 1)
    .thenCompose(id ->
        CompletableFuture.supplyAsync(() -> fetchDetails(id))
            // الدمج مع جلب السعر بالتوازي
            .thenCombine(
                CompletableFuture.supplyAsync(() -> fetchPrice(id)),
                (details, price) -> details + " | price: $" + price
            )
    )

    // الخطوة 3 — التعافي في حالة أي فشل
    .exceptionally(ex -> "Product unavailable: " + ex.getMessage());

System.out.println(pipeline.join());

الخلاصة

استخدم thenCompose حين تكون الخطوة التالية بحد ذاتها غير متزامنة (لتجنّب التداخل المزدوج).
استخدم thenCombine لدمج مستقبلَين مستقلَّين دون تسلسلهما.
استخدم allOf للتوزيع على مجموعة ديناميكية وجمع كل النتائج بعد ذلك.
ألصق exceptionally أو handle قريبًا من المرحلة التي قد تفشل لتعافٍ دقيق.
اختر دومًا الصيغة المتزامنة (Async) مع executor صريح حين يجب أن يبقى I/O المانع بعيدًا عن المجموعة المشتركة.