RxJava: جوہر، اجزاء اور ری ایکٹو پروگرامنگ

RxJava کیا ہے؟

RxJava جاوا ورچوئل مشین کے لیے ReactiveX (Reactive Extensions) لائبریری کا ایک نفاذ ہے۔ RxJava کا پہلا ورژن Netflix نے 2013 میں سرور سائیڈ ایپلیکیشنز میں غیر مطابقت پذیر کالز کے انتظام کے لیے جاری کیا تھا۔ اس کی تخلیق کے وقت، Java میں اہم متبادل Future اور Callback تھے — دونوں طریقے callback-hell اور پیچیدہ تھریڈ مینجمنٹ کی طرف لے جاتے تھے۔ RxJava نے فنکشنل آپریٹرز کی زنجیروں کے ساتھ Observable کے ذریعے غیر مطابقت پذیر کارروائیوں کی ترکیب متعارف کرائی۔

RxJava کا فن تعمیر Reactive Streams تصریح پر مبنی ہے — غیر مسدود بیک پریشر کے ساتھ غیر مطابقت پذیر اسٹریم پروسیسنگ کا ایک معیار۔ تصریح چار انٹرفیس کی وضاحت کرتی ہے: Publisher، Subscriber، Subscription اور Processor۔ RxJava 2+ RxJava 1 کے برعکس بیک پریشر کے معاہدوں کی پابندی کرتے ہوئے Flowable قسم کے ذریعے Reactive Streams کو مکمل طور پر نافذ کرتا ہے۔ RxJava 2 میں Observable بیک پریشر کو سپورٹ نہیں کرتا — یہ کم تعداد میں ایونٹس یا UI ایونٹس والی اسٹریمز کے لیے ہے۔

JetBrains، 2025 سروے کے مطابق، RxJava Android ڈویلپمنٹ کے لیے سرفہرست 3 لائبریریوں میں شامل ہے۔ اہم استعمال کے معاملات میں شامل ہیں: Retrofit کے ذریعے نیٹ ورک کی درخواستوں کو ہینڈل کرنا (CallAdapter کے ذریعے RxJava کے ساتھ مربوط)، Room کے ساتھ کام کرنا (ری ایکٹو سوالات Flowable یا Maybe لوٹاتے ہیں)، RxBinding کے ذریعے اینیمیشنز اور UI ایونٹس، اور ٹیکسٹ ان پٹ پر ڈیباؤنس تلاش۔ یہ تمام منظرنامے ایک مشترکہ زنجیر کے پیٹرن کا اشتراک کرتے ہیں: ماخذ (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 بڑے ڈیٹا اسٹریمز کے ساتھ کام کرنے کے لیے سب سے لچکدار قسم ہے۔ یہ بیک پریشر سپورٹ کے ساتھ Reactive Streams Publisher کو نافذ کرتا ہے: صارف Subscription.request(n) کے ذریعے عناصر کی ایک مخصوص تعداد کی درخواست کر سکتا ہے۔ یہ بفر اوور فلو کو روکتا ہے جب پروڈیوسر اور صارف کی رفتار مطابقت نہیں رکھتی۔ اگر بیک پریشر اہم نہیں ہے تو Observable استعمال کریں — درخواست کے طریقہ کار کی عدم موجودگی کی وجہ سے اس کا اوور ہیڈ کم ہے۔

Single HTTP درخواستوں کے لیے بہترین انتخاب ہے۔ RxJava CallAdapter کے ساتھ Retrofit 2 ہر درخواست کے لیے Single<ResponseBody> لوٹاتا ہے۔ Single بالکل ایک onSuccess یا onError کال کی ضمانت دیتا ہے، جو HTTP درخواست کی معنویات سے مطابقت رکھتا ہے — ایک جواب یا ایک خرابی۔ Completable تحریری کارروائیوں کے لیے استعمال ہوتا ہے جو ڈیٹا واپس نہیں کرتی: insert، update، delete۔ Maybe کیشے کی جانچ کے لیے آسان ہے — یہ قدر واپس کر سکتا ہے یا نہیں۔

// HTTP درخواست کے لیے Single کے استعمال کی مثال
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() تیز ٹائپنگ کے دوران غیر ضروری درخواستوں کو روکتا ہے۔

Scheduler اور ملٹی تھریڈنگ

Scheduler RxJava میں تھریڈ پول پر ایک تجرید ہے۔ لائبریری پانچ بلٹ ان Scheduler فراہم کرتی ہے: I/O آپریشنز (نیٹ ورک، فائلز) کے لیے Schedulers.io()، CPU انٹینسیو کاموں کے لیے Schedulers.computation()، ہر بار ایک نیا تھریڈ بنانے کے لیے Schedulers.newThread()، سنگل تھریڈ پر عملدرآمد کے لیے Schedulers.single() اور موجودہ تھریڈ میں فوری عملدرآمد کے لیے Schedulers.trampoline()۔

subscribeOn اور observeOn

subscribeOn طے کرتا ہے کہ کون سا Scheduler ماخذ Observable کو چلاتا ہے۔ اگر زنجیر میں متعدد subscribeOn ہوں تو ماخذ کے قریب ترین کو ترجیح دی جاتی ہے۔ observeOn ڈاؤن اسٹریم کو مخصوص Scheduler پر منتقل کرتا ہے — observeOn کا ہر استعمال بعد کے آپریٹرز کے لیے تھریڈ تبدیل کرتا ہے۔ ایک عام Android پیٹرن: نیٹ ورک آپریشنز کے لیے subscribeOn(Schedulers.io())، UI اپ ڈیٹس کے لیے observeOn(AndroidSchedulers.mainThread())۔

// سیاق و سباق کی تبدیلی کے ساتھ ملٹی تھریڈڈ پروسیسنگ
Observable.fromCallable(() -> database.getItems())
    .subscribeOn(Schedulers.io())            // io پر DB
    .map(items -> processItems(items))     // io پر تبدیلی
    .observeOn(Schedulers.computation())    // computation پر جائیں
    .map(processed -> compressImages(processed))
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(result -> ui.showResult(result))

AndroidSchedulers.mainThread() RxAndroid لائبریری سے ایک Scheduler ہے جو Android مین تھریڈ پر کوڈ چلاتا ہے۔ یہ ری ایکٹو زنجیر میں کسی بھی UI اپ ڈیٹ کے لیے لازمی ہے۔ لائبریری اندرونی طور پر Handler استعمال کرتی ہے اور زیادہ لوڈ کے تحت بھی UI تھریڈ پر عملدرآمد کی ضمانت دیتی ہے۔ بیک گراؤنڈ آپریشنز کے لیے، Schedulers.io() لامحدود تھریڈ پول کو سپورٹ کرتا ہے اور کسی بھی بلاک کرنے والی کارروائی کے لیے موزوں ہے۔ Schedulers.computation() CPU کورز کی تعداد کے برابر ایک فکسڈ پول استعمال کرتا ہے۔

Android میں RxJava: عملی اطلاقات

RxJava Android میں تین اہم منظرناموں کے لیے استعمال ہوتا ہے: Room کو ری ایکٹو سوالات، Retrofit کے ساتھ انضمام، اور RxBinding کے ذریعے ری ایکٹو UI بائنڈنگ۔ ہر منظرنامے کی اپنی قسم کا مجموعہ ہے: Room قابل مشاہدہ سوالات کے لیے Flowable لوٹاتا ہے، Retrofit HTTP درخواستوں کے لیے Single لوٹاتا ہے، RxBinding UI ایونٹس کے لیے Observable لوٹاتا ہے۔

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، Transformations کے ذریعے LiveData) شائع کرتا ہے، اور Activity یا Fragment ان میں سبسکرائب کرتا ہے۔ یہ جانچ پڑتال فراہم کرتا ہے: ViewModel کا UI کے بغیر تجربہ کیا جاتا ہے، RxJavaPlugins.setComputationScheduler کے ذریعے Schedulers کو تبدیل کرکے۔ ViewModel میں CompositeDisposable سبسکرپشن لائف سائیکل کا انتظام کرتا ہے — onCleared() پر، تمام سبسکرپشنز منسوخ ہو جاتی ہیں۔

RxJava بمقابلہ Kotlin Flow

Kotlin Flow Kotlin میں کولڈ اسٹریمز کا ایک مقامی نفاذ ہے، جو کوروٹینز میں بنایا گیا ہے اور Kotlin 1.3 میں متعارف کرایا گیا ہے۔ Flow RxJava جیسے مسائل حل کرتا ہے لیکن بنیادی فرقوں کے ساتھ: بلٹ ان کوروٹین سپورٹ (suspend فنکشنز)، coroutine cancellation کے ذریعے منسوخی، اور بیک پریشر کے مسائل کی عدم موجودگی — Flow بفرنگ کے بجائے suspend استعمال کرتا ہے۔ Flow Kotlin معیاری لائبریری کا حصہ ہے، جس کے لیے اضافی انحصار کی ضرورت نہیں ہے۔

RxJava Java پروجیکٹس، Java 7-8 کو سپورٹ کرنے والے پروجیکٹس اور موجودہ RxJava کوڈ بیسز کے لیے ترجیحی انتخاب ہے۔ RxJava ایکو سسٹم نمایاں طور پر زیادہ امیر ہے: Flow میں تقریباً 50 کے مقابلے میں 400 سے زیادہ آپریٹرز، بلٹ ان CallAdapter کے ذریعے Retrofit کے ساتھ انضمام، Flowable کے ذریعے بیک پریشر سپورٹ، اور Android کے لیے RxBinding، RxPermissions، RxLocation۔ Kotlin Flow تیزی سے پکڑ رہا ہے، لیکن پیچیدہ اسٹریم یکجا کرنے کے منظرناموں میں RxJava کی لچک اب بھی زیادہ ہے۔