الشبكات وHTTP

HTTP غير المتزامن

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

HTTP غير المتزامن

في الدرس السابق استخدمتَ HttpClient.send() — استدعاء حاجب يُثبّت الخيط الحالي حتى يستجيب الخادم. لعدد محدود من الطلبات المتسلسلة هذا مقبول تمامًا. لكن حين تحتاج إلى إرسال عشرات الطلبات بالتوازي، أو ربط نتائج بعضها ببعض، أو إبقاء خيط واجهة المستخدم أو الخادم غير محجوب، فإن النمط الحاجب هو الخيار الخاطئ. HttpClient.sendAsync() هو الحل: يعود فورًا بـ CompletableFuture<HttpResponse<T>> يمكنك تركيبه بصورة غير حاجبة باستخدام الإمكانات الكاملة لواجهة CompletableFuture التي درستها في دروس التزامن.

sendAsync: الأساسيات

توقيع sendAsync يطابق send تمامًا — يأخذ نفس HttpRequest وBodyHandler — لكنه يُعيد مستقبلًا بدلًا من الحجب:

import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.net.http.HttpResponse.BodyHandlers;
import java.util.concurrent.CompletableFuture;

HttpClient client = HttpClient.newHttpClient();

HttpRequest request = HttpRequest.newBuilder()
    .uri(URI.create("https://api.example.com/v1/products/42"))
    .header("Accept", "application/json")
    .build();

CompletableFuture<HttpResponse<String>> future =
    client.sendAsync(request, BodyHandlers.ofString());

// الخيط الحالي حر للقيام بعمل آخر.
// حين تصل الاستجابة يكتمل المستقبل على منفّذ العميل.
HttpResponse<String> response = future.join(); // احجب فقط حين لا بد
System.out.println(response.statusCode());
System.out.println(response.body());

join() مقابل get() — كلاهما يحجب حتى اكتمال المستقبل. يُفضَّل join() في خطوط الأنابيب غير الحاجبة لأنه يُعيد رمي الاستثناءات كـ CompletionException غير محقَّقة بدلًا من ExecutionException المحقَّقة، مما يُبقي جسم اللامبدا نظيفًا.

تركيب المستقبل: thenApply وthenAccept وthenCompose

القوة الحقيقية تكمن في أنك لا تحتاج إلى الحجب أبدًا. بدلًا من ذلك، تُسلسل ردود النداء التي تعمل فور اكتمال كل مرحلة. أهم ثلاث دوال تركيب هي:

thenApply(Function) — تحويل النتيجة بصورة متزامنة؛ تُعيد CompletableFuture جديدًا من النوع المُحوَّل.
thenAccept(Consumer) — استهلاك النتيجة (أثر جانبي)؛ تُعيد CompletableFuture<Void>.
thenCompose(Function) — خريطة مُسطَّحة (flat-map): حين يُعيد رد النداء بنفسه CompletableFuture، هذه الدالة تتجنب التداخل (مكافئة flatMap في التدفقات غير المتزامنة).

مثال عملي — جلب منتج، استخراج اسمه، ثم تسجيله دون حجب صريح قط:

client.sendAsync(request, BodyHandlers.ofString())
    .thenApply(HttpResponse::body)               // استخراج الجسم: CF<String>
    .thenApply(body -> parseProductName(body))   // تحويل: CF<String>
    .thenAccept(name -> System.out.println("المنتج: " + name))  // استهلاك
    .exceptionally(ex -> {
        System.err.println("فشل الطلب: " + ex.getMessage());
        return null;
    });

إرسال طلبات متعددة بالتوازي

هنا يتألّق sendAsync حقًا. إرسال N طلبًا بالتسلسل يستغرق N * RTT. إرسالها بالتوازي يستغرق RTT واحدة تقريبًا. استخدم Stream لبناء المستقبلات، ثم CompletableFuture.allOf() للانتظار على جميعها:

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

List<String> productIds = List.of("42", "99", "107", "221");

// 1. إطلاق جميع الطلبات بالتوازي
List<CompletableFuture<String>> futures = productIds.stream()
    .map(id -> HttpRequest.newBuilder()
            .uri(URI.create("https://api.example.com/v1/products/" + id))
            .build())
    .map(req -> client.sendAsync(req, BodyHandlers.ofString())
            .thenApply(HttpResponse::body))
    .collect(Collectors.toList());

// 2. الدمج: انتظر اكتمال الجميع ثم اجمع النتائج
CompletableFuture<Void> allDone = CompletableFuture.allOf(
    futures.toArray(new CompletableFuture[0])
);

List<String> bodies = allDone
    .thenApply(v -> futures.stream()
            .map(CompletableFuture::join)   // join آمن هنا — الجميع اكتملوا
            .collect(Collectors.toList()))
    .join();

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

allOf + join() هو الأسلوب الاصطلاحي. بعد عودة allOf(...).join() يكون كل مستقبل فردي قد اكتمل بالفعل، لذا استدعاء .join() على كل منها لا يحجب — يقرأ النتيجة المخزّنة فحسب.

anyOf: أول استجابة ناجحة تفوز

أحيانًا تريد أسرع استجابة من بين عدة نقاط نهاية متكافئة — نمط الطلب المُحوَّط (hedged request) المستخدم للمرونة:

HttpRequest mirror1 = HttpRequest.newBuilder()
    .uri(URI.create("https://cdn-eu.example.com/data.json")).build();
HttpRequest mirror2 = HttpRequest.newBuilder()
    .uri(URI.create("https://cdn-us.example.com/data.json")).build();

CompletableFuture<Object> fastest = CompletableFuture.anyOf(
    client.sendAsync(mirror1, BodyHandlers.ofString()).thenApply(HttpResponse::body),
    client.sendAsync(mirror2, BodyHandlers.ofString()).thenApply(HttpResponse::body)
);

String result = (String) fastest.join();

لاحظ أن anyOf تُعيد CompletableFuture<Object> (بلا نوع عام محدد) لأن نظام الأنواع في Java لا يستطيع توحيد أنواع المستقبلات المختلفة. تُلقي النتيجة يدويًا. الطلب الآخر الجاري يستمر في الخلفية وتُجمَّع ذاكرته حين لا تبقى إشارة إلى مستقبله.

معالجة الأخطاء في خطوط الأنابيب غير المتزامنة

يجب معالجة كل من أخطاء الشبكة (رفض الاتصال، انتهاء المهلة) ورموز الحالة غير 2xx. sendAsync نفسها تُفشل المستقبل لأخطاء النقل فقط — 404 أو 500 من الخادم تُعدّ مستقبلًا ناجحًا برمز حالة سيئ. تحتاج إلى التحقق من كليهما:

client.sendAsync(request, BodyHandlers.ofString())
    .thenApply(response -> {
        if (response.statusCode() < 200 || response.statusCode() >= 300) {
            throw new RuntimeException(
                "خطأ HTTP: " + response.statusCode() + " لـ " + request.uri());
        }
        return response.body();
    })
    .exceptionally(ex -> {
        // تعالج كلًا من أخطاء النقل وRuntimeException المُرمى
        System.err.println("فشل: " + ex.getCause().getMessage());
        return "{}"; // قيمة افتراضية آمنة للإبقاء على خط الأنابيب
    })
    .thenAccept(body -> processBody(body));

exceptionally() تبتلع الخطأ وتستعيد. إن أردتَ إعادة رمي الخطأ بدلًا من الاستعادة، استخدم whenComplete() أو handle() — يتيحان لك فحص كل من النتيجة والاستثناء وإعادة الرمي عند الحاجة. استدعاء exceptionally() بإعادة قيمة غير null يُحوّل المرحلة الفاشلة إلى ناجحة.

التحكم في المنفّذ

افتراضيًا، تعمل ردود نداء اكتمال sendAsync على المنفّذ الداخلي للعميل (مجموعة fork-join صغيرة). في تطبيق خادم كثيرًا ما تريد تشغيل ردود النداء على مجموعة خيوط محددة — مثلًا نفس منفّذ الخيوط الافتراضية الذي ضبطته على العميل:

import java.util.concurrent.Executors;

HttpClient client = HttpClient.newBuilder()
    .executor(Executors.newVirtualThreadPerTaskExecutor()) // Java 21+
    .build();

// تعمل ردود النداء الآن على خيوط افتراضية — خفيفة وغير حاجبة وقابلة للتوسع
client.sendAsync(request, BodyHandlers.ofString())
    .thenApply(HttpResponse::body)
    .thenAccept(body -> processBody(body));

بديلًا، أضف اللاحقة Async لأي مرحلة — thenApplyAsync(fn, executor) — لنقل تلك المرحلة تحديدًا إلى مجموعة خيوط مختارة دون تغيير العميل بأكمله.

الجمع معًا: نمط واقعي

نمط إنتاجي نموذجي: جلب قائمة مُعرِّفات من نقطة نهاية، ثم إثراء كل مُعرِّف بالتوازي من نقطة نهاية التفاصيل، وجمع النتائج الناجحة فقط:

import java.util.Optional;

// الخطوة 1: جلب المُعرِّفات (الحجب مقبول هنا — استدعاء واحد فقط)
List<String> ids = fetchIdList(client); // مساعد متزامن

// الخطوة 2: إثراء بالتوازي
List<CompletableFuture<Optional<Product>>> enrichFutures = ids.stream()
    .map(id -> client
        .sendAsync(detailRequest(id), BodyHandlers.ofString())
        .thenApply(r -> r.statusCode() == 200
            ? Optional.of(parseProduct(r.body()))
            : Optional.<Product>empty())
        .exceptionally(ex -> Optional.empty()))  // فشل الشبكة → فارغ
    .collect(Collectors.toList());

// الخطوة 3: انتظر الجميع، صفِّ الناجحين
List<Product> products = CompletableFuture
    .allOf(enrichFutures.toArray(new CompletableFuture[0]))
    .thenApply(v -> enrichFutures.stream()
        .map(CompletableFuture::join)
        .filter(Optional::isPresent)
        .map(Optional::get)
        .collect(Collectors.toList()))
    .join();

الخلاصة

تُعيد sendAsync() فورًا CompletableFuture مُحرِّرةً الخيط الحالي. ركِّبه بـ thenApply وthenCompose وthenAccept للتحويلات؛ استخدم allOf للتوزيع والتجميع، وanyOf للطلبات المُحوَّطة. عالج دائمًا أخطاء النقل ورموز الحالة غير 2xx بصورة صريحة. في Java 21+ اقرن ذلك بمنفّذ خيوط افتراضية لأقصى إنتاجية بأدنى تهيئة.