البرمجة التفاعلية في Java: مكتبة Reactor

البرمجة التفاعلية في Java باستخدام Project Reactor

\n\n

البرمجة التفاعلية هي نموذج تطوير يركز على تدفقات البيانات غير المتزامنة ونشر التغييرات. في Java، يتم تنفيذ هذا المفهوم من خلال مواصفة Reactive Streams وتطبيقها الأكثر شيوعًا — مكتبة Project Reactor (جزء من نظام Spring WebFlux البيئي).

\n\n

على عكس الكود الأمرّي التقليدي، حيث ينفذ البرنامج الأوامر بشكل تسلسلي، يسمح النهج التفاعلي بمعالجة البيانات فور وصولها، دون حظر تيارات التنفيذ. هذا مهم بشكل خاص للأنظمة عالية التحميل، والخدمات المصغرة، والتطبيقات التي تتعامل مع عدد كبير من الطلبات المتزامنة.

\n\n

المكونات الأساسية لـ Reactor

\n\n

توفر المكتبة نوعين رئيسيين من التسلسلات التفاعلية:

\n\n

\n
• Mono<T> — يمثل تدفقًا يمكن أن يحتوي على 0 أو 1 عنصر. يُستخدم للعمليات غير المتزامنة الفردية (مثل طلب HTTP، أو جلب سجل من قاعدة البيانات).
\n
• Flux<T> — يمثل تدفقًا يمكن أن يحتوي على من 0 إلى N عنصر. مناسب للمجموعات، وتدفقات الأحداث، أو البيانات التي تصل على أجزاء.
\n

\n\n

كلا النوعين يطبقان واجهة Publisher من مواصفة Reactive Streams ويدعمان آلية الضغط العكسي (backpressure) — وهي قدرة المشترك على التحكم في سرعة استقبال البيانات.

\n\n

مثال: إنشاء ومعالجة Flux

\n\n

لنلقِ نظرة على مثال بسيط لإنشاء تدفق تفاعلي من الأرقام ومعالجته:

\n\n

import reactor.core.publisher.Flux;\n\npublic class ReactorExample {\n    public static void main(String[] args) {\n        Flux<Integer> numbers = Flux.just(1, 2, 3, 4, 5);\n\n        numbers\n            .map(n -> n * 2)          // تحويل\n            .filter(n -> n % 3 == 0)   // تصفية\n            .subscribe(\n                data -> System.out.println("تم الاستلام: " + data),\n                error -> System.err.println("خطأ: " + error),\n                () -> System.out.println("اكتمل!")\n            );\n    }\n}\n

\n\n

في هذا المثال:

\n

\n
• Flux.just() ينشئ تدفقًا من خمسة أرقام.
\n
• map() يضرب كل عنصر في 2.
\n
• filter() يبقي فقط الأرقام القابلة للقسمة على 3.
\n
• subscribe() يبدأ المعالجة — يطبع النتيجة، ويعالج الأخطاء، ويشير إلى الاكتمال.
\n

\n\n

عدم التزامن والعمل مع Mono

\n\n

يسمح Reactor بتنفيذ عمليات غير متزامنة دون حظر التيارات. على سبيل المثال، محاكاة طلب إلى خدمة بعيدة:

\n\n

import reactor.core.publisher.Mono;\nimport java.time.Duration;\n\npublic class AsyncExample {\n    public static void main(String[] args) throws InterruptedException {\n        Mono<String> asyncResult = Mono.fromSupplier(() -> {\n            try {\n                Thread.sleep(1000); // محاكاة تأخير\n            } catch (InterruptedException e) {\n                e.printStackTrace();\n            }\n            return "تم استلام البيانات";\n        }).subscribeOn(Schedulers.boundedElastic()); // التنفيذ في تيار آخر\n\n        System.out.println("تم إرسال الطلب...");\n        \n        asyncResult.subscribe(System.out::println);\n        \n        Thread.sleep(2000); // انتظار الاكتمال\n    }\n}\n

\n\n

النقطة الأساسية: استدعاء .subscribeOn() ينقل تنفيذ العملية المكلفة إلى تيار منفصل، دون حظر التيار الرئيسي. سيحصل المشترك على النتيجة بشكل غير متزامن.

\n\n

معالجة الأخطاء في التدفقات التفاعلية

\n\n

على عكس try-catch، في البرمجة التفاعلية تتم معالجة الأخطاء من خلال عوامل التشغيل:

\n\n

Flux<String> fluxWithError = Flux.just("A", "B", "C")\n    .map(s -> {\n        if ("B".equals(s)) t