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() तेज़ टाइपिंग के दौरान अनावश्यक अनुरोधों को रोकता है।

Schedulers और मल्टीथ्रेडिंग

Scheduler RxJava में थ्रेड पूल पर एक अमूर्तता है। लाइब्रेरी पाँच अंतर्निर्मित Schedulers प्रदान करती है: I/O संचालन (नेटवर्क, फ़ाइलें) के लिए Schedulers.io(), CPU-गहन कार्यों के लिए Schedulers.computation(), हर बार एक नए थ्रेड के लिए Schedulers.newThread(), एकल-थ्रेड निष्पादन के लिए Schedulers.single(), और वर्तमान थ्रेड में तत्काल निष्पादन के लिए Schedulers.trampoline()।

subscribeOn और observeOn

subscribeOn यह निर्धारित करता है कि कौन सा Scheduler स्रोत Observable को निष्पादित करता है। यदि श्रृंखला में कई subscribeOn हैं, तो स्रोत के सबसे निकट वाले को प्राथमिकता दी जाती है। observeOn downstream को निर्दिष्ट 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 का लचीलापन अभी भी अधिक है।