RxJava: मूल बातें, ReactiveX और डेटा स्ट्रीम के साथ काम करना

RxJava और ReactiveX क्या है

RxJava ReactiveX विनिर्देश का जावा कार्यान्वयन है, जो ऑब्ज़र्वेबल स्ट्रीम (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 Observables से एलिमेंट को इंडेक्स द्वारा जोड़ता है
• merge — कई Observables को एक में मर्ज करता है, कालानुक्रमिक क्रम बनाए रखता है
• 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 पैटर्न पर बना है और push-आधारित है: स्रोत डेटा भेजता है, Observer प्रतिक्रिया करता है। Coroutines pull-आधारित हैं: कोड await के माध्यम से क्रमिक रूप से डेटा का अनुरोध करता है।

• RxJava — रिएक्टिव, डेटा स्ट्रीम, >200 ऑपरेटर, push-आधारित, कठिन सीखने की अवस्था
• Coroutines — अनुक्रमिक, suspend/await, ~40 फ़ंक्शन, pull-आधारित, सरल सिंटैक्स
• 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 महीने लगते हैं।