RxJava: الأساسيات، ReactiveX والعمل مع تدفقات البيانات

ما هو RxJava وReactiveX

RxJava هو تنفيذ Java لمواصفة ReactiveX، وهي مكتبة للبرمجة غير المتزامنة باستخدام التدفقات القابلة للملاحظة (Observable). تم إصدار RxJava 2 في عام 2016 مع دعم Reactive Streams (Flowable) والفصل إلى rx.Observable وio.reactivex.Observable. RxJava 3 (2019) هو الإصدار الرئيسي الحالي مع توافق عكسي مع RxJava 2.

الفكرة الأساسية لـ RxJava هي أن كل شيء هو تدفق: تدفق بيانات، تدفق أحداث، تدفق حالات. يمكن تمثيل أي عملية غير متزامنة كـ Observable يبعث بيانات أو خطأ أو إشارة اكتمال. يشترك Observer في Observable ويتلقى الإشعارات في الوقت الفعلي.

وفقًا لـ Badoo (2024)، قبل الانتقال إلى coroutines، كان 76% من تطبيقات Android في أفضل 200 على Google Play تستخدم RxJava للعمليات غير المتزامنة. النسبة الآن في انخفاض لصالح coroutines، لكن RxJava لا يزال موجودًا في كود الإنتاج لآلاف التطبيقات ويعتبر تقنية ناضجة ومُختبرة. ReactiveX هي مواصفة عبر المنصات تم تنفيذها أيضًا لـ JavaScript (RxJS) و.NET (Rx.NET) وSwift (RxSwift) ولغات أخرى.

نمط Observer في RxJava

يوسع ReactiveX نمط Observer الكلاسيكي بآليتين: سلسلة العوامل وإدارة الخيوط عبر Schedulers. لا يبدأ Observable في بعث البيانات حتى يشترك Observer (التقييم الكسول). يتيح ذلك بناء خط أنابيب بيانات يتم تنشيطه فقط عند وجود اشتراك.

أنواع Observable: Observable، Flowable، Single، Maybe، Completable

Observable — النوع الأساسي الذي يبعث 0..N عنصر مع onError أو onComplete. مناسب لتدفقات البيانات غير المحدودة — على سبيل المثال، أحداث النقر أو تحديثات الموقع الجغرافي. Observable لا يدعم الضغط العكسي (backpressure).

Flowable — إصدار Reactive Streams من Observable مع دعم الضغط العكسي. يُستخدم عندما قد يولد مصدر البيانات عناصر أسرع مما يستطيع Observer معالجته. Flowable يدعم استراتيجيات BACKPRESSURE_BUFFER وDROP وLATEST وERROR.

Single وMaybe وCompletable

Single يبعث عنصرًا واحدًا بالضبط أو خطأ — مثالي لطلبات الشبكة. Maybe يبعث 0 أو 1 عنصر، مناسب للذاكرة المؤقتة حيث قد تكون البيانات غائبة. Completable يبعث فقط onComplete أو onError، بدون بيانات، مناسب لعمليات الكتابة أو الحذف. تعمل هذه الأنواع على تبسيط API عن طريق تضييق العقد إلى حالة محددة. Retrofit (عميل HTTP شائع لـ Android) يدعم مباشرة جميع أنواع RxJava الخمسة، مما يسمح باختيار نوع الإرجاع الأكثر ملاءمة لكل نقطة نهاية بدون كود إضافي.

عوامل RxJava: تحويل وتصفية التدفقات

العوامل هي وظائف تحول Observable إلى آخر. تصف سلسلة العوامل خط أنابيب البيانات: كل عامل يأخذ التدفق من السابق، يحوله، ويمره إلى التالي. يحتوي RxJava على أكثر من 200 عامل مقسمة إلى فئات.

• map — يحول كل عنصر (Integer → String)
• flatMap — يحول عنصرًا إلى Observable ويدمج الكل في تدفق واحد
• filter — يمرر العناصر التي تحقق شرطًا
• zip — يجمع العناصر من N Observable حسب الفهرس
• merge — يدمج عدة Observable في واحد، مع الحفاظ على الترتيب الزمني
• debounce — يبعث العناصر فقط إذا مرت فترة زمنية محددة بدون انبعاث آخر

flatMap هو أحد أقوى عوامل RxJava. يتيح تنفيذ طلب غير متزامن لكل عنصر وجمع النتائج في تدفق مشترك. على سبيل المثال، يُستخدم flatMap لتحميل التفاصيل من قائمة معرفات: كل معرف → طلب شبكة → دمج النتائج. على عكس map الذي يحول العنصر ببساطة، يمكن لـ flatMap بعث عناصر متعددة أو التبديل إلى Observable آخر، مما يجعله الأساس لبناء خطوط الأنابيب غير المتزامنة.

معالجة الأخطاء عبر العوامل

onErrorResumeNext — يتحول إلى Observable احتياطي عند الخطأ. retry — يعيد الاشتراك N مرة عند الخطأ. onErrorReturn — يعيد قيمة افتراضية بدلاً من الخطأ. doOnError — ينفذ تأثيرًا جانبيًا عند الخطأ دون تغيير التدفق (تسجيل أو تحليلات). يتيح دمج هذه العوامل بناء خطوط أنابيب قوية باستراتيجية معالجة أخطاء واضحة بدون try/catch يدوي.

Schedulers: إدارة الخيوط في RxJava

Schedulers تحدد الخيط الذي يتم فيه تنفيذ Observable وObserver. يحدد subscribeOn خيط المصدر، ويحدد observeOn خيط Observer والعوامل اللاحقة. هذا الفصل هو ميزة رئيسية لـ RxJava: المصدر على خيط IO، المعالجة على computation، واجهة المستخدم على الخيط الرئيسي.

أهم Schedulers: Schedulers.io() — لعمليات الإدخال/الإخراج (شبكة، قرص)، pool غير محدود. Schedulers.computation() — للحسابات، pool ثابت حسب عدد النوى. Schedulers.newThread() — خيط جديد لكل مهمة. AndroidSchedulers.mainThread() — الخيط الرئيسي لـ Android (RxAndroid). يوجد أيضًا Schedulers.trampoline() لتنفيذ المهام في الخيط الحالي مع طابور FIFO، مفيد للاختبارات.

وفقًا لـ Google (2025)، الاستخدام الصحيح لـ Schedulers هو أصعب جزء في RxJava للمبتدئين. الخطأ النموذجي هو استدعاء subscribeOn بعد observeOn، مما لا يؤثر على المصدر. يجب أن يكون subscribeOn أولًا في السلسلة للمصدر، وobserveOn قبل اشتراك واجهة المستخدم. قاعدة: subscribeOn يؤثر فقط على المنبع (المصدر)، observeOn يغير المصب (المشترك وجميع العوامل بعده).

أمثلة كود RxJava في Android

لننظر في ثلاثة سيناريوهات: طلب شبكة باستخدام Single، طلبات متوازية باستخدام zip، وdebounce لحقل بحث باستخدام debounce.

طلب شبكة باستخدام Single

Single مثالي لطلبات Retrofit: طلب واحد — استجابة واحدة. اشترك في الخيط الرئيسي لتحديث واجهة المستخدم.

api.getUser(id)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new SingleObserver<User>() {
        @Override
        public void onSuccess(User user) { showUser(user); }
        @Override
        public void onError(Throwable e) { showError(e); }
    })

طلبات متوازية باستخدام zip

zip يجمع نتائج اثنين من Single المستقلين في واحد. يتم تنفيذهما بالتوازي، وتظهر النتيجة بعد اكتمال كليهما.

Single.zip(
    api.getProfile(),
    api.getSettings(),
    (profile, settings) -> new Dashboard(profile, settings)
)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(dashboard -> showDashboard(dashboard), e -> logError(e))

Debounce لحقل البحث

debounce يتجاهل التغييرات السريعة في النص ويرسل طلبًا فقط بعد توقف 400 مللي ثانية. distinctUntilChanged يلغي الطلب إذا لم يتغير النص.

RxTextView.textChanges(searchView)
    .debounce(400, TimeUnit.MILLISECONDS)
    .filter(text -> text.length() >= 3)
    .distinctUntilChanged()
    .switchMap(query -> api.search(query))
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(results -> showResults(results))

RxJava مقابل Kotlin Coroutines: مقارنة المناهج

RxJava وKotlin Coroutines يحلان نفس المشكلة — البرمجة غير المتزامنة — لكن بمناهج مختلفة جوهريًا. RxJava مبني على نمط Observer وهو قائم على الدفع: المصدر يرسل البيانات، Observer يتفاعل. Coroutines قائمة على السحب: الكود يطلب البيانات بشكل تسلسلي عبر await.

• RxJava — تفاعلي، تدفق بيانات، >200 عامل، قائم على الدفع، منحنى تعلم حاد
• Coroutines — تسلسلي، suspend/await، ~40 دالة، قائم على السحب، بناء جملة بسيط
• RxJava — ناضج (2016)، نظام بيئي ضخم، لكن منحنى تعلم حاد
• Coroutines — حديث (2018)، الخيار المفضل لـ Google للكود الجديد
• RxJava — ضغط عكسي مدمج عبر Flowable، استراتيجيات تخزين مؤقت مُختبرة
• Coroutines — Flow مع ضغط عكسي حديث، لكن قيد التطوير النشط من JetBrains

وفقًا لـ Google I/O 2024، Kotlin Coroutines هي النهج الموصى به للكود غير المتزامن الجديد في Android. يظل RxJava مدعومًا للمشاريع القائمة. توفر Google مكتبات جسر (kotlinx-coroutines-rx3) للترحيل التدريجي. AndroidX (LiveData، Room، Paging 3) يدعم كلا النهجين، مما يسمح باستخدام RxJava في الوحدات القديمة وCoroutines في الجديدة بدون تعارض التبعيات.

استراتيجية الترحيل من RxJava إلى Coroutines

انتقال تدريجي: كل مكون جديد يُكتب بـ Coroutines، وكود RxJava القديم لا يُلمس. RxJava ← Coroutines عبر awaitSingle() أو awaitFirst(). Coroutines ← RxJava عبر future() أو asFlowable(). يستغرق الترحيل الكامل 6–18 شهرًا للمشاريع الكبيرة.