RxJava: بنیادی باتیں، ReactiveX اور ڈیٹا سٹریم کے ساتھ کام کرنا

RxJava اور ReactiveX کیا ہے

RxJava ReactiveX تصریح کا Java نفاذ ہے، جو قابل مشاہدہ سٹریمز (Observable) کا استعمال کرتے ہوئے غیر متزامن پروگرامنگ کے لیے ایک لائبریری ہے۔ RxJava 2 2016 میں Reactive Streams (Flowable) سپورٹ اور rx.Observable اور io.reactivex.Observable میں تقسیم کے ساتھ جاری کیا گیا تھا۔ RxJava 3 (2019) RxJava 2 کے ساتھ پسماندہ مطابقت رکھنے والا موجودہ بڑا ورژن ہے۔

RxJava کا بنیادی خیال یہ ہے کہ ہر چیز ایک سٹریم ہے: ڈیٹا سٹریم، ایونٹ سٹریم، اسٹیٹ سٹریم۔ کسی بھی غیر متزامن آپریشن کو Observable کے طور پر پیش کیا جا سکتا ہے جو ڈیٹا، خرابی یا تکمیل کا سگنل خارج کرتا ہے۔ ایک Observer Observable کو سبسکرائب کرتا ہے اور ریئل ٹائم میں نوٹیفیکیشن وصول کرتا ہے۔

Badoo (2024) کے مطابق، coroutines میں منتقلی سے پہلے، Google Play کے ٹاپ 200 میں سے 76% Android ایپس غیر متزامن آپریشنز کے لیے RxJava استعمال کرتی تھیں۔ اب یہ حصہ coroutines کے حق میں کم ہو رہا ہے، لیکن RxJava ہزاروں ایپس کے پروڈکشن کوڈ میں موجود ہے اور اسے ایک پختہ، آزمودہ ٹیکنالوجی سمجھا جاتا ہے۔ ReactiveX ایک کراس پلیٹ فارم تصریح ہے جسے JavaScript (RxJS)، .NET (Rx.NET)، Swift (RxSwift) اور دیگر زبانوں کے لیے بھی نافذ کیا گیا ہے۔

RxJava میں Observer پیٹرن

ReactiveX کلاسک Observer پیٹرن کو دو میکانزم کے ساتھ بڑھاتا ہے: آپریٹر چیننگ اور Scheduler پر مبنی تھریڈنگ۔ Observable اس وقت تک ڈیٹا خارج کرنا شروع نہیں کرتا جب تک کوئی Observer سبسکرائب نہ کرے (سست تشخیص)۔ یہ ڈیٹا پائپ لائن بنانے کی اجازت دیتا ہے جو صرف سبسکرپشن موجود ہونے پر فعال ہوتی ہے۔

Observable کی اقسام: Observable، Flowable، Single، Maybe، Completable

Observable — بنیادی قسم جو onError یا onComplete کے ساتھ 0..N عناصر خارج کرتی ہے۔ لامحدود ڈیٹا سٹریمز کے لیے موزوں — مثال کے طور پر، کلک ایونٹس یا جغرافیائی محل وقوع کی تازہ کاریاں۔ Observable بیک پریشر کو سپورٹ نہیں کرتا۔

Flowable — بیک پریشر سپورٹ کے ساتھ Observable کا Reactive Streams ورژن۔ اس وقت استعمال ہوتا ہے جب ڈیٹا کا ذریعہ Observer کی پروسیسنگ رفتار سے تیز عناصر پیدا کر سکتا ہو۔ Flowable BACKPRESSURE_BUFFER، DROP، LATEST اور ERROR کی حکمت عملیوں کو سپورٹ کرتا ہے۔

Single، Maybe اور Completable

Single بالکل ایک عنصر یا خرابی خارج کرتا ہے — نیٹ ورک کی درخواستوں کے لیے مثالی۔ Maybe 0 یا 1 عنصر خارج کرتا ہے، کیش کے لیے موزوں جہاں ڈیٹا موجود نہ ہو سکے۔ Completable ڈیٹا کے بغیر صرف onComplete یا onError خارج کرتا ہے، تحریر یا حذف کرنے کے آپریشنز کے لیے آسان۔ یہ اقسام معاہدے کو مخصوص صورت تک محدود کرکے API کو آسان بناتی ہیں۔ Retrofit (Android کے لیے ایک مقبول HTTP کلائنٹ) پانچوں RxJava اقسام کو براہ راست سپورٹ کرتا ہے، جس سے اضافی کوڈ کے بغیر ہر اینڈ پوائنٹ کے لیے سب سے موزوں ریٹرن قسم منتخب کی جا سکتی ہے۔

RxJava آپریٹرز: سٹریم ٹرانسفارمیشن اور فلٹرنگ

آپریٹرز وہ فنکشنز ہیں جو ایک Observable کو دوسرے میں تبدیل کرتے ہیں۔ آپریٹر چین ڈیٹا پائپ لائن کو بیان کرتی ہے: ہر آپریٹر پچھلے سے سٹریم لیتا ہے، اسے تبدیل کرتا ہے اور اگلے کو منتقل کرتا ہے۔ 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 کا سب سے مشکل حصہ ہے۔ ایک عام غلطی observeOn کے بعد subscribeOn کو کال کرنا ہے، جو ماخذ کو متاثر نہیں کرتا۔ subscribeOn ماخذ کے لیے چین میں پہلا ہونا چاہیے، observeOn UI سبسکرپشن سے پہلے۔ قاعدہ: subscribeOn صرف اوپر کی طرف (ماخذ) کو متاثر کرتا ہے، observeOn نیچے کی طرف (سبسکرائبر اور اس کے بعد کے تمام آپریٹرز) کو سوئچ کرتا ہے۔

Android میں RxJava کوڈ کی مثالیں

تین منظرناموں پر غور کریں: Single کے ساتھ نیٹ ورک کی درخواست، zip کے ساتھ متوازی درخواستیں، اور 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 تیز رفتار متن کی تبدیلیوں کو نظر انداز کرتا ہے اور صرف 400 ms کے توقف کے بعد درخواست بھیجتا ہے۔ 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 مہینے لگتے ہیں۔