RxJava: ماهیت، مؤلفه‌ها و برنامه‌نویسی واکنش‌گرا

RxJava چیست؟

RxJava — پیاده‌سازی کتابخانه ReactiveX (Reactive Extensions) برای ماشین مجازی Java است. نسخه اول RxJava توسط شرکت Netflix در سال 2013 برای مدیریت فراخوانی‌های ناهمگام در برنامه‌های سرور منتشر شد. در زمان ایجاد، جایگزین اصلی در Java Future و Callback بودند — هر دو رویکرد به callback-hell و مدیریت پیچیده نخ‌ها منجر می‌شدند. RxJava ترکیب عملیات ناهمگام را از طریق Observable با زنجیره‌هایی از عملگرهای تابعی پیشنهاد کرد.

معماری RxJava بر اساس مشخصات Reactive Streams — استانداردی برای پردازش ناهمگام جریان‌ها با backpressure غیرمسدودکننده — استوار است. این مشخصات چهار رابط را تعریف می‌کند: Publisher, Subscriber, Subscription و Processor. RxJava 2+ به طور کامل Reactive Streams را از طریق نوع Flowable پیاده‌سازی می‌کند و برخلاف RxJava 1 از قراردادهای backpressure پیروی می‌کند. Observable در RxJava 2 از backpressure پشتیبانی نمی‌کند — برای جریان‌هایی با تعداد کم رویداد یا رویدادهای UI طراحی شده است.

بر اساس نظرسنجی JetBrains, 2025، RxJava در بین سه کتابخانه برتر برای توسعه Android قرار دارد. سناریوهای اصلی استفاده: پردازش درخواست‌های شبکه از طریق Retrofit (ادغام شده با RxJava از طریق CallAdapter)، کار با Room (پرس‌وجوهای واکنش‌گرا Flowable یا Maybe برمی‌گردانند)، انیمیشن‌ها و رویدادهای UI از طریق RxBinding و جستجوی debounce هنگام ورود متن. همه این سناریوها با یک زنجیره مشابه متحد می‌شوند: منبع (Observable) → تبدیل (عملگرها) → اشتراک (subscribe).

تاریخچه نسخه‌های RxJava

RxJava 1 (2013) مفهوم Observable و عملگرها را پایه‌گذاری کرد، اما از مشکلات backpressure رنج می‌برد — در جریان‌های سریع، داده‌ها در حافظه جمع شده و باعث OutOfMemoryError می‌شدند. RxJava 2 (2016) معماری را اصلاح کرد و Observable (بدون backpressure) و Flowable (با backpressure) را جدا کرد. RxJava 3 (2020) پشتیبانی از Java 8 Stream API، عملگرهای اضافی و عملکرد بهبود یافته در اشتراک را اضافه کرد. در حال حاضر RxJava 3 — نسخه توصیه شده برای پروژه‌های جدید است.

انواع جریان‌های واکنش‌گرا در RxJava

RxJava پنج نوع اصلی از منابع واکنش‌گرا را فراهم می‌کند که هر کدام برای سناریوی خاصی طراحی شده‌اند. Observable و Flowable مقادیر متعددی منتشر می‌کنند، Single — یک مقدار یا خطا، Completable — فقط واقعیت تکمیل بدون داده، Maybe — یک مقدار، صفر یا خطا. انتخاب نوع صحیح مقدار کد را کاهش می‌دهد و زنجیره را خودمستند می‌کند.

Flowable — انعطاف‌پذیرترین نوع برای کار با جریان‌های داده بزرگ است. Publisher Reactive Streams را با پشتیبانی از backpressure پیاده‌سازی می‌کند: consumer می‌تواند تعداد مشخصی از عناصر را از طریق Subscription.request(n) درخواست کند. این کار از سرریز بافر هنگام عدم تطابق سرعت producer و consumer جلوگیری می‌کند. اگر backpressure حیاتی نیست — از Observable استفاده کنید، به دلیل عدم وجود مکانیزم request سربار کمتری دارد.

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 عناصر را از چندین Observable به صورت جفتی بر اساس شاخص ترکیب می‌کند: اولی با اولی، دومی با دومی. combineLatest با تغییر هر یک از جریان‌ها مقدار جدیدی منتشر می‌کند و آخرین مقادیر همه جریان‌ها را ترکیب می‌کند. merge چندین Observable را در یکی ترکیب می‌کند و ترتیب ورود رویدادها را حفظ می‌کند. concat به صورت ترتیبی در هر Observable اشتراک می‌کند و تمام رویدادهای آن را قبل از رفتن به بعدی منتقل می‌کند.

مدیریت زمان شامل debounce (انتظار برای مکث در جریان قبل از ارسال)، throttleFirst (عبور اولین رویداد، نادیده گرفتن بقیه در طول پنجره)، timeout (خطا اگر رویداد در بازه زمانی نیامد) است. جستجوی debounce هنگام ورود متن — رایج‌ترین سناریو: searchObservable.debounce(300, MILLISECONDS).distinctUntilChanged() از درخواست‌های اضافی هنگام تایپ سریع جلوگیری می‌کند.

Schedulers و چندنخی

Scheduler در RxJava — انتزاعی بر روی استخر نخ‌ها است. کتابخانه پنج Scheduler داخلی فراهم می‌کند: Schedulers.io() برای عملیات I/O (شبکه، فایل‌ها)، Schedulers.computation() برای وظایف سنگین CPU، Schedulers.newThread() برای هر نخ جدید، Schedulers.single() برای اجرای تک‌نخی و Schedulers.trampoline() برای اجرای فوری در نخ فعلی.

subscribeOn و observeOn

subscribeOn تعیین می‌کند که منبع Observable در کدام Scheduler اجرا شود. اگر چندین subscribeOn در زنجیره وجود داشته باشد — نزدیک‌ترین به منبع اولویت دارد. observeOn downstream را به Scheduler مشخص شده تغییر می‌دهد — هر استفاده از observeOn نخ را برای عملگرهای بعدی تغییر می‌دهد. الگوی معمول Android: subscribeOn(Schedulers.io()) برای کار با شبکه، observeOn(AndroidSchedulers.mainThread()) برای به‌روزرسانی UI.

// پردازش چندنخی با تغییر زمینه
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 اجرا می‌کند. این برای هرگونه به‌روزرسانی UI در زنجیره واکنش‌گرا اجباری است. کتابخانه در داخل از Handler استفاده می‌کند و اجرا در نخ UI را حتی تحت بار بالا تضمین می‌کند. برای عملیات پس‌زمینه، Schedulers.io() از استخر نخ نامحدود پشتیبانی می‌کند و برای هرگونه عملیات مسدودکننده مناسب است. Schedulers.computation() از استخر ثابتی برابر با تعداد هسته‌های پردازنده استفاده می‌کند.

RxJava در Android: کاربرد عملی

RxJava در Android برای سه سناریوی اصلی استفاده می‌شود: پرس‌وجوهای واکنش‌گرا به Room، ادغام با Retrofit و اتصال واکنش‌گرای UI از طریق RxBinding. برای هر سناریو مجموعه نوع خاص خود مشخص است: Room برای پرس‌وجوهای قابل مشاهده Flowable برمی‌گرداند، Retrofit — Single برای درخواست‌های HTTP، RxBinding — Observable برای رویدادهای UI.

Room + RxJava

Room — کتابخانه ماندگاری داده از Google. از Room 2.1 به بعد، پایگاه داده از انواع برگشتی واکنش‌گرا پشتیبانی می‌کند: Flowable و Observable. هنگام تغییر هر رکورد در جدول، Room به طور خودکار مقدار جدیدی را به جریان ارسال می‌کند. توسعه‌دهنده در ViewModel در Flowable اشتراک می‌کند و بدون پرس‌وجوی دستی در هر تغییر، داده‌های به‌روز را دریافت می‌کند.

// 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 به View ارجاعی ندارد. ViewModel منابع واکنش‌گرا (Flowable، LiveData از طریق Transformations) منتشر می‌کند و Activity یا Fragment در آنها اشتراک می‌کنند. این امر تست‌پذیری را فراهم می‌کند: ViewModel بدون UI و با جایگزینی Scheduler از طریق RxJavaPlugins.setComputationScheduler تست می‌شود. CompositeDisposable در ViewModel چرخه حیات اشتراک‌ها را مدیریت می‌کند — در onCleared() همه اشتراک‌ها لغو می‌شوند.

RxJava در مقابل Kotlin Flow

Kotlin Flow — پیاده‌سازی بومی جریان‌های سرد در Kotlin است که در کوروتین‌ها تعبیه شده و در Kotlin 1.3 معرفی شده است. Flow همان وظایف RxJava را حل می‌کند، اما با تفاوت‌های اساسی: پشتیبانی داخلی از کوروتین‌ها (توابع suspend)، لغو از طریق coroutine cancellation و عدم وجود مشکلات backpressure — Flow به جای بافر کردن از suspend استفاده می‌کند. Flow بخشی از کتابخانه استاندارد Kotlin است و به وابستگی‌های اضافی نیاز ندارد.

RxJava انتخاب ارجح برای پروژه‌های Java، پروژه‌های با پشتیبانی Java 7-8 و پایگاه‌های کد موجود RxJava باقی می‌ماند. اکوسیستم RxJava به طور قابل توجهی غنی‌تر است: >400 عملگر در مقابل ~50 در Flow، ادغام با Retrofit از طریق CallAdapter داخلی، پشتیبانی از backpressure از طریق Flowable و وجود RxBinding، RxPermissions، RxLocation برای Android. Kotlin Flow به سرعت در حال پیشرفت است، اما انعطاف‌پذیری RxJava در سناریوهای پیچیده ترکیب جریان همچنان بالاتر است.