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)، قبل از انتقال به کرتین‌ها، 76% از برنامه‌های Android از 200 برنامه برتر Google Play از RxJava برای عملیات ناهمگام استفاده می‌کردند. اکنون سهم به نفع کرتین‌ها کاهش می‌یابد، اما RxJava در کد تولید هزاران برنامه باقی می‌ماند و یک فناوری بالغ و آزمایش‌شده محسوب می‌شود. ReactiveX — یک مشخصات چندسکویی است که برای JavaScript (RxJS)، .NET (Rx.NET)، Swift (RxSwift) و زبان‌های دیگر نیز پیاده‌سازی شده است.

الگوی Observer در RxJava

ReactiveX الگوی کلاسیک Observer را با دو مکانیزم گسترش می‌دهد: زنجیره عملگرها (operator chaining) و مدیریت رشته‌ها (schedulers). Observable تا زمانی که Observer مشترک نشود ارسال داده را آغاز نمی‌کند (ارزیابی تنبل). این امکان ساخت خط لوله داده را فراهم می‌کند که فقط در صورت وجود اشتراک فعال می‌شود.

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

Observable — نوع پایه که 0..N عنصر را با onError یا onComplete ارسال می‌کند. مناسب برای جریان‌های داده با طول نامحدود — مثلاً رویدادهای کلیک یا به‌روزرسانی موقعیت جغرافیایی. Observable از backpressure پشتیبانی نمی‌کند.

Flowable — نسخه Reactive Streams از Observable با پشتیبانی از backpressure. زمانی استفاده می‌شود که منبع داده می‌تواند عناصر را سریع‌تر از توانایی Observer در پردازش تولید کند. Flowable از استراتژی‌های BACKPRESSURE_BUFFER، DROP، LATEST و ERROR پشتیبانی می‌کند.

Single، Maybe و Completable

Single دقیقاً یک عنصر یا خطا ارسال می‌کند — ایده‌آل برای درخواست‌های شبکه. Maybe — 0 یا 1 عنصر، مناسب برای کش که داده ممکن است وجود نداشته باشد. Completable — فقط onComplete یا onError، بدون داده، مناسب برای عملیات نوشتن یا حذف. این انواع API را ساده‌تر می‌کنند و قرارداد را به حالت خاص محدود می‌کنند. Retrofit (کلاینت HTTP محبوب برای Android) از هر پنج نوع RxJava مستقیماً پشتیبانی می‌کند و امکان انتخاب مناسب‌ترین نوع بازگشت را برای هر endpoint بدون پوشش اضافی فراهم می‌کند.

عملگرهای RxJava: تبدیل و فیلتر جریان‌ها

عملگرها توابعی هستند که یک Observable را به Observable دیگر تبدیل می‌کنند. زنجیره عملگرها (operator chain) خط لوله داده را توصیف می‌کند: هر عملگر جریان را از قبلی می‌گیرد، آن را تبدیل می‌کند و به بعدی منتقل می‌کند. RxJava شامل بیش از 200 عملگر تقسیم‌شده به دسته‌ها است.

• map — هر عنصر را تبدیل می‌کند (Integer → String)
• flatMap — عنصر را به Observable تبدیل کرده و همه را در یک جریان ترکیب می‌کند
• filter — عناصر را بر اساس شرط عبور می‌دهد
• zip — عناصر N Observable را بر اساس ایندکس ترکیب می‌کند
• merge — چند Observable را در یکی ادغام می‌کند و ترتیب زمانی را حفظ می‌کند
• debounce — عناصر را اگر فاصله زمانی بین آنها کمتر از بازه مشخص شده باشد عبور می‌دهد

flatMap — یکی از قدرتمندترین عملگرهای RxJava. امکان اجرای درخواست ناهمگام برای هر عنصر و جمع‌آوری نتایج در یک جریان مشترک را فراهم می‌کند. به عنوان مثال، flatMap برای بارگذاری جزئیات بر اساس لیست ID استفاده می‌شود: هر ID → درخواست شبکه → ترکیب نتایج. برخلاف map که صرفاً عنصر را تبدیل می‌کند، flatMap می‌تواند چندین عنصر ارسال کند یا به Observable دیگر سوئیچ کند که آن را به پایه‌ای برای ساخت خطوط لوله ناهمگام تبدیل می‌کند.

مدیریت خطا از طریق عملگرها

onErrorResumeNext — در صورت خطا به Observable پشتیبان سوئیچ می‌کند. retry — اشتراک را در صورت خطا N بار تکرار می‌کند. onErrorReturn — مقدار پیش‌فرض را به جای خطا برمی‌گرداند. doOnError — عمل جانبی در زمان خطا بدون تغییر جریان انجام می‌دهد (ثبت یا تحلیل). ترکیب این عملگرها امکان ساخت خطوط لوله قابل اعتماد با استراتژی روشن مدیریت خطا بدون try/catch دستی را فراهم می‌کند.

Schedulers: مدیریت رشته‌ها در RxJava

Schedulers تعیین می‌کند که Observable و Observer روی کدام رشته اجرا شوند. subscribeOn رشته را برای منبع تعیین می‌کند، observeOn — رشته را برای Observer و عملگرهای بعدی. این جداسازی — مزیت کلیدی RxJava است: منبع روی رشته IO، پردازش روی computation، UI — روی رشته اصلی.

Schedulers اصلی: Schedulers.io() — برای عملیات I/O (شبکه، دیسک)، استخر نامحدود. Schedulers.computation() — برای محاسبات، استخر ثابت بر اساس تعداد هسته‌ها. Schedulers.newThread() — رشته جدید برای هر وظیفه. AndroidSchedulers.mainThread() — رشته اصلی Android (RxAndroid). همچنین Schedulers.trampoline() برای اجرای وظایف در رشته فعلی با صف FIFO وجود دارد که برای تست‌ها مفید است.

طبق داده‌های Google (2025)، استفاده صحیح از Schedulers سخت‌ترین بخش RxJava برای مبتدیان است. خطای معمول — فراخوانی subscribeOn بعد از observeOn که روی منبع تأثیر نمی‌گذارد. subscribeOn باید اولین در زنجیره برای منبع باشد، observeOn — قبل از اشتراک UI. قاعده: subscribeOn فقط روی upstream (منبع) تأثیر می‌گذارد، observeOn downstream (مشترک و همه عملگرهای بعد از آن) را تغییر می‌دهد.

نمونه کدهای RxJava در Android

سه سناریو را بررسی می‌کنیم: درخواست شبکه با Single، درخواست‌های موازی با zip و debounce برای فیلد جستجو با debounce.

درخواست شبکه با Single

Single برای درخواست‌های Retrofit ایده‌آل است: یک درخواست — یک پاسخ. اشتراک روی رشته اصلی برای به‌روزرسانی UI.

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 ساخته شده و push-based است: منبع داده ارسال می‌کند، Observer واکنش نشان می‌دهد. کرتین‌ها — pull-based: کد به صورت ترتیبی داده را از طریق await دریافت می‌کند.

• RxJava — واکنشی، جریان داده، >200 عملگر، push-based، منحنی یادگیری دشوار
• Coroutines — ترتیبی، suspend/await، ~40 تابع، pull-based، نحو ساده
• RxJava — بالغ (2016)، اکوسیستم عظیم، اما منحنی یادگیری دشوار
• Coroutines — مدرن (2018)، انتخاب ترجیحی Google برای کد جدید
• RxJava — backpressure داخلی از طریق Flowable، استراتژی‌های بافر تکامل‌یافته
• Coroutines — Flow با backpressure اخیراً، اما فعالانه توسط JetBrains توسعه می‌یابد

طبق Google I/O 2024، Kotlin Coroutines رویکرد توصیه‌شده برای کد ناهمگام جدید در Android است. RxJava برای پروژه‌های موجود پشتیبانی می‌شود. Google کتابخانه‌های انتقالی (kotlinx-coroutines-rx3) برای مهاجرت تدریجی ارائه می‌دهد. AndroidX (LiveData، Room، Paging 3) از هر دو رویکرد پشتیبانی می‌کند و امکان استفاده از RxJava در ماژول‌های قدیمی و کرتین‌ها در ماژول‌های جدید بدون تضاد وابستگی را فراهم می‌کند.

استراتژی مهاجرت از RxJava به کرتین‌ها

انتقال تدریجی: هر کامپوننت جدید با کرتین‌ها نوشته می‌شود، کد RxJava قدیمی تغییر نمی‌کند. RxJava → کرتین‌ها از طریق awaitSingle() یا awaitFirst(). کرتین‌ها → RxJava از طریق future() یا asFlowable(). مهاجرت کامل برای پروژه‌های بزرگ 6–18 ماه طول می‌کشد.