RxJava: الجوهر والمكونات والبرمجة التفاعلية

ما هو RxJava؟

RxJava هو تطبيق لمكتبة ReactiveX (Reactive Extensions) للآلة الافتراضية Java. تم إصدار النسخة الأولى من RxJava بواسطة Netflix في عام 2013 لإدارة الاستدعاءات غير المتزامنة في تطبيقات الخادم. في وقت إنشائه، كانت البدائل الرئيسية في Java هي Future و Callback — كلا النهجين أدى إلى تعقيد الاستدعاءات وإدارة معقدة للخيوط. قدم RxJava تركيب العمليات غير المتزامنة عبر Observable مع سلاسل من عوامل التشغيل الوظيفية.

تعتمد بنية RxJava على مواصفات Reactive Streams — معيار لمعالجة التدفقات غير المتزامنة مع ضغط عكسي غير محظور. تحدد المواصفة أربع واجهات: Publisher و Subscriber و Subscription و Processor. ينفذ RxJava 2+ مواصفات Reactive Streams بالكامل من خلال نوع Flowable، مع الالتزام بعقود الضغط العكسي على عكس RxJava 1. لا يدعم Observable في RxJava 2 الضغط العكسي — فهو مخصص للتدفقات ذات عدد صغير من الأحداث أو أحداث واجهة المستخدم.

وفقًا لاستطلاع JetBrains، 2025، يعد RxJava من بين أفضل 3 مكتبات لتطوير Android. حالات الاستخدام الرئيسية تشمل: معالجة طلبات الشبكة عبر Retrofit (متكامل مع RxJava من خلال CallAdapter)، العمل مع Room (الاستعلامات التفاعلية تعيد Flowable أو Maybe)، الرسوم المتحركة وأحداث واجهة المستخدم عبر RxBinding، والبحث المزيل للارتداد عند إدخال النص. تشترك جميع هذه السيناريوهات في نمط سلسلة مشترك: مصدر (Observable) ← تحويل (عوامل تشغيل) ← اشتراك (subscribe).

تاريخ إصدارات RxJava

RxJava 1 (2013) وضع الأساس مع Observable وعوامل التشغيل، لكنه عانى من مشاكل الضغط العكسي — في التدفقات السريعة، تراكمت البيانات في الذاكرة، مما تسبب في OutOfMemoryError. RxJava 2 (2016) أصلح البنية بفصل Observable (بدون ضغط عكسي) و Flowable (مع ضغط عكسي). RxJava 3 (2020) أضاف دعم Java 8 Stream API وعوامل تشغيل إضافية وأداء محسن للاشتراك. حاليًا، RxJava 3 هو الإصدار الموصى به للمشاريع الجديدة.

أنواع التدفقات التفاعلية في RxJava

RxJava يوفر خمسة أنواع رئيسية من المصادر التفاعلية، كل منها مصمم لسيناريو محدد. Observable و Flowable يصدران قيمًا متعددة، Single يصدر قيمة واحدة أو خطأ، Completable يصدر فقط الإكتمال بدون بيانات، و Maybe يصدر قيمة واحدة أو صفر أو خطأ. اختيار النوع الصحيح يقلل من حجم الكود ويجعل السلسلة موثقة ذاتيًا.

Flowable هو النوع الأكثر مرونة للعمل مع تدفقات البيانات الكبيرة. ينفذ Publisher الخاص بـ Reactive Streams مع دعم الضغط العكسي: يمكن للمستهلك طلب عدد محدد من العناصر عبر Subscription.request(n). هذا يمنع تجاوز سعة المخزن المؤقت عندما لا تتطابق سرعات المنتج والمستهلك. إذا لم يكن الضغط العكسي حرجًا، استخدم Observable — له تكلفة إضافية أقل بسبب عدم وجود آلية الطلب.

Single هو الخيار الأمثل لطلبات HTTP. Retrofit 2 مع RxJava CallAdapter يعيد Single<ResponseBody> لكل طلب. يضمن Single استدعاء واحد بالضبط لـ onSuccess أو onError، وهو ما يتوافق مع دلالات طلب HTTP — استجابة واحدة أو خطأ واحد. Completable يستخدم لعمليات الكتابة التي لا تعيد بيانات: insert، update، delete. Maybe مناسب للتحقق من ذاكرة التخزين المؤقت — قد يعيد قيمة أو لا.

// مثال على استخدام Single لطلب HTTP
interface ApiService {
    @GET("users/{id}")
    fun getUser(@Path("id") userId: Int): Single<User>
}

// الاشتراك مع المعالجة في الخيط الرئيسي
apiService.getUser(42)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe({ user ->
        textView.text = user.name
    }, { error ->
        Log.e("API", "Error: ${error.message}")
    })
    .addTo(compositeDisposable)

عوامل تحويل وإدارة التدفقات

عوامل التشغيل في RxJava هي دوال عالية الترتيب تأخذ مصدرًا تفاعليًا وتعيد آخر، محولة تدفق البيانات. يحتوي RxJava 3 على أكثر من 400 عامل تشغيل، مقسمة إلى فئات: التحويل والتصفية والدمج ومعالجة الأخطاء وإدارة الوقت. كل عامل تشغيل كسول — يتم بناء السلسلة عند التصريح وتنفيذها عند الاشتراك.

عوامل التحويل

map هو العامل الأساسي الذي يحول كل قيمة عبر دالة. flatMap يأخذ دالة تعيد Observable لكل عنصر ويفرد النتيجة في تدفق واحد. switchMap مشابه لـ flatMap، لكن عند وصول عنصر جديد، يلغي الاشتراك من Observable السابق. concatMap يحافظ على ترتيب العناصر — على عكس flatMap، يشترك بالتسلسل في كل Observable متداخل.

// تحليل JSON مع التحويل والتصفية
apiService.getUsers()
    .flatMap { users ->
        Observable.fromIterable(users)
    }
    .filter { user ->
        user.age >= 18
    }
    .map { user ->
        UserDto(user.name, user.age)
    }
    .toList()
    .subscribeOn(Schedulers.computation())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe({ adapter.submitList(it) },
               { Log.e("خطأ", it.message) })

دمج التدفقات هو مجال يتفوق فيه RxJava بشكل خاص. zip يدمج العناصر من Observables متعددة أزواجًا حسب الفهرس: الأول مع الأول، الثاني مع الثاني. combineLatest يصدر قيمة جديدة عندما يتغير أي تدفق، مدمجًا أحدث القيم من جميع التدفقات. merge يدمج Observables متعددة في تدفق واحد، محافظًا على ترتيب وصول الأحداث. concat يشترك بالتسلسل في كل Observable ويمرر جميع أحداثه قبل الانتقال إلى التالي.

إدارة الوقت تشمل debounce (انتظار توقف في التدفق قبل الإصدار)، throttleFirst (إصدار أول حدث، تجاهل الباقي خلال نافذة زمنية)، timeout (خطأ إذا لم يصل حدث خلال الفاصل الزمني). البحث المزيل للارتداد عند إدخال النص هو السيناريو الأكثر شيوعًا: searchObservable.debounce(300, MILLISECONDS).distinctUntilChanged() يمنع الطلبات غير الضرورية أثناء الكتابة السريعة.

Schedulers وتعدد الخيوط

Scheduler في RxJava هو تجريد فوق مجموعة خيوط. توفر المكتبة خمسة Schedulers مدمجة: Schedulers.io() لعمليات الإدخال/الإخراج (الشبكة، الملفات)، Schedulers.computation() للمهام كثيفة الاستخدام لوحدة المعالجة المركزية، Schedulers.newThread() لخيط جديد في كل مرة، Schedulers.single() للتنفيذ أحادي الخيط، و Schedulers.trampoline() للتنفيذ الفوري في الخيط الحالي.

subscribeOn و observeOn

subscribeOn يحدد أي Scheduler ينفذ Observable المصدر. إذا كان هناك عدة subscribeOn في السلسلة، الأولوية للأقرب إلى المصدر. observeOn يحول المصب إلى Scheduler المحدد — كل استخدام لـ observeOn يغير الخيط لعوامل التشغيل التالية. نمط Android النموذجي: subscribeOn(Schedulers.io()) للعمل مع الشبكة، observeOn(AndroidSchedulers.mainThread()) لتحديث واجهة المستخدم.

// معالجة متعددة الخيوط مع تبديل السياق
Observable.fromCallable(() -> database.getItems())
    .subscribeOn(Schedulers.io())            // قاعدة البيانات على io
    .map(items -> processItems(items))     // التحويل على io
    .observeOn(Schedulers.computation())    // التبديل إلى computation
    .map(processed -> compressImages(processed))
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(result -> ui.showResult(result))

AndroidSchedulers.mainThread() هو Scheduler من مكتبة RxAndroid ينفذ الكود على الخيط الرئيسي لنظام Android. وهو إلزامي لأي تحديثات لواجهة المستخدم في سلسلة تفاعلية. تستخدم المكتبة Handler داخليًا وتضمن التنفيذ في خيط واجهة المستخدم حتى تحت الأحمال العالية. للعمليات الخلفية، يدعم Schedulers.io() مجموعة خيوط غير محدودة ومناسب لأي عمليات حظر. يستخدم Schedulers.computation() مجموعة ثابتة تساوي عدد أنوية وحدة المعالجة المركزية.

RxJava في Android: التطبيقات العملية

RxJava في Android يُستخدم لثلاثة سيناريوهات رئيسية: الاستعلامات التفاعلية لـ Room، التكامل مع Retrofit، والربط التفاعلي لواجهة المستخدم عبر RxBinding. لكل سيناريو مجموعة خاصة به من الأنواع: Room يعيد Flowable للاستعلامات القابلة للملاحظة، Retrofit يعيد Single لطلبات HTTP، RxBinding يعيد Observable لأحداث واجهة المستخدم.

Room + RxJava

Room هي مكتبة استمرارية بيانات من Google. بدءًا من Room 2.1، تدعم قاعدة البيانات أنواع الإرجاع التفاعلية: Flowable و Observable. عندما يتغير أي سجل في الجدول، يرسل Room تلقائيًا قيمة جديدة إلى التدفق. يشترك المطور في Flowable في ViewModel ويحصل على بيانات محدثة بدون استعلامات يدوية عند كل تغيير.

// Room DAO مع استعلام تفاعلي
@Dao
interface UserDao {
    @Query("SELECT * FROM users WHERE id = :id")
    fun getUserById(@Param("id") userId: Int): Flowable<User>

    @Insert
    fun insertUser(user: User): Completable
}

// ViewModel — تركيب Room + Network
class UserViewModel(private val dao: UserDao) : ViewModel() {
    val users: Flowable<List<User>> = dao.getAllUsers()
        .subscribeOn(Schedulers.io())
}

نمط MVVM + RxJava مبني على أن ViewModel لا يحتوي على مراجع للعرض. ينشر ViewModel مصادر تفاعلية (Flowable، LiveData عبر Transformations)، ويشترك فيها Activity أو Fragment. هذا يوفر قابلية الاختبار: يتم اختبار ViewModel بدون واجهة المستخدم، عن طريق استبدال Schedulers عبر RxJavaPlugins.setComputationScheduler. يدير CompositeDisposable في ViewModel دورة حياة الاشتراكات — عند onCleared()، يتم إلغاء جميع الاشتراكات.

RxJava مقابل Kotlin Flow

Kotlin Flow هو تطبيق أصلي للتدفقات الباردة في Kotlin، مدمج في coroutines ومقدم في Kotlin 1.3. يحل Flow نفس المشاكل التي يحلها RxJava ولكن مع اختلافات جوهرية: دعم مدمج لـ coroutines (دوال suspend)، الإلغاء عبر coroutine cancellation، وغياب مشاكل الضغط العكسي — يستخدم Flow suspend بدلاً من التخزين المؤقت. Flow جزء من المكتبة القياسية لـ Kotlin، ولا يتطلب تبعيات إضافية.

RxJava يظل الخيار المفضل لمشاريع Java، والمشاريع التي تدعم Java 7-8، وقواعد الكود الحالية على RxJava. نظام RxJava البيئي أغنى بكثير: أكثر من 400 عامل تشغيل مقابل حوالي 50 في Flow، والتكامل مع Retrofit عبر CallAdapter مدمج، ودعم الضغط العكسي عبر Flowable، و RxBinding و RxPermissions و RxLocation لنظام Android. Kotlin Flow يلحق بسرعة، لكن مرونة RxJava في سيناريوهات دمج التدفقات المعقدة لا تزال أعلى.