RxJava: asoslar, ReactiveX va ma’lumot oqimlari bilan ishlash

RxJava va ReactiveX nima

RxJava — ReactiveX spetsifikatsiyasining Java tatbiqi, kuzatiladigan oqimlar (Observable) orqali asinxron dasturlash uchun kutubxona. RxJava 2 2016-yilda Reactive Streams (Flowable) qo’llab-quvvatlashi va rx.Observable hamda io.reactivex.Observable ga bo’linishi bilan chiqarilgan. RxJava 3 (2019) — RxJava 2 bilan orqaga moslikka ega joriy asosiy versiya.

RxJava ning asosiy g’oyasi — hamma narsa oqim: ma’lumot oqimi, hodisa oqimi, holat oqimi. Har qanday asinxron operatsiyani ma’lumot, xato yoki tugash signalini uzatuvchi Observable sifatida taqdim etish mumkin. Observer Observable’ga obuna bo’ladi va real vaqtda bildirishnomalar oladi.

Badoo ma’lumotlariga ko’ra (2024), korutinlarga o‘tishdan oldin top-200 Google Play’dagi 76% Android ilovalari asinxron operatsiyalar uchun RxJava dan foydalangan. Hozirda ulush korutinlar foydasiga kamaymoqda, ammo RxJava minglab ilovalarning ishlab chiqarish kodida qolmoqda va etuk, sinovdan o‘tgan texnologiya hisoblanadi. ReactiveX — JavaScript (RxJS), .NET (Rx.NET), Swift (RxSwift) va boshqa tillar uchun ham tatbiq qilingan ko‘p platformali spetsifikatsiya.

RxJava’dagi Observer naqshi

ReactiveX klassik Observer naqshini ikki mexanizm bilan kengaytiradi: operator zanjiri (operator chaining) va oqimlarni boshqarish (schedulers). Observable Observer obuna bo‘lgunga qadar ma’lumotlarni uzatishni boshlamaydi (lazy evaluation). Bu faqat obuna mavjud bo‘lganda faollashadigan ma’lumot pipeline’ini qurish imkonini beradi.

Observable turlari: Observable, Flowable, Single, Maybe, Completable

Observable — onError yoki onComplete bilan 0..N elementni uzatuvchi asosiy tur. Cheklanmagan uzunlikdagi ma’lumot oqimlari uchun mos — masalan, bosish hodisalari yoki geolokatsiya yangilanishlari. Observable backpressure’ni qo’llab-quvvatlamaydi.

Flowable — backpressure qo’llab-quvvatlovchi Reactive Streams versiyasi. Ma’lumot manbai elementlarni Observer qayta ishlashidan tezroq yaratishi mumkin bo‘lganda ishlatiladi. Flowable BACKPRESSURE_BUFFER, DROP, LATEST va ERROR strategiyalarini qo‘llab-quvvatlaydi.

Single, Maybe va Completable

Single aynan bitta element yoki xatoni uzatadi — tarmoq so‘rovlari uchun ideal. Maybe — 0 yoki 1 element, ma’lumot bo‘lmasligi mumkin bo‘lgan kesh uchun mos. Completable — faqat onComplete yoki onError, ma’lumotsiz, yozish yoki o‘chirish operatsiyalari uchun qulay. Bu turlar API’ni soddalashtiradi, kontraktni aniq holatga toraytiradi. Retrofit (Android uchun mashhur HTTP klienti) barcha besh RxJava turini to‘g‘ridan-to‘g‘ri qo‘llab-quvvatlaydi, har bir endpoint uchun keraksiz o‘ramlarsiz eng mos qaytarish turini tanlash imkonini beradi.

RxJava operatorlari: oqimlarni transformatsiya qilish va filtrlash

Operatorlar — bir Observable’ni boshqasiga aylantiruvchi funksiyalar. Operator zanjiri (operator chain) ma’lumot pipeline’ini tasvirlaydi: har bir operator oldingisidan oqimni qabul qiladi, uni o‘zgartiradi va keyingisiga uzatadi. RxJava toifalarga bo‘lingan 200 dan ortiq operatorni o‘z ichiga oladi.

• map — har bir elementni o‘zgartiradi (Integer → String)
• flatMap — elementni Observable’ga aylantiradi va hammasini bitta oqimda birlashtiradi
• filter — elementlarni shart bo‘yicha o‘tkazadi
• zip — N Observable elementlarini indeks bo‘yicha birlashtiradi
• merge — bir nechta Observable’ni bittasiga birlashtiradi, vaqt tartibini saqlaydi
• debounce — elementlar orasida belgilangan intervaldan kam bo‘lsa, o‘tkazadi

flatMap — RxJava’ning eng kuchli operatorlaridan biri. Har bir element uchun asinxron so‘rovni bajarish va natijalarni umumiy oqimda to‘plash imkonini beradi. Masalan, flatMap ID ro‘yxati bo‘yicha tafsilotlarni yuklash uchun ishlatiladi: har bir ID → tarmoq so‘rovi → natijalarni birlashtirish. Faqat elementni o‘zgartiruvchi map’dan farqli o‘laroq, flatMap bir nechta elementni uzatishi yoki boshqa Observable’ga o‘tishi mumkin, bu esa uni asinxron pipeline’lar qurish uchun asos qiladi.

Operatorlar orqali xato boshqaruvi

onErrorResumeNext — xatoda zaxira Observable’ga o‘tadi. retry — xatoda obunani N marta takrorlaydi. onErrorReturn — xato o‘rniga standart qiymatni qaytaradi. doOnError — oqimni o‘zgartirmasdan xatoda yon ta’sirni bajaradi (loglash yoki analitika). Bu operatorlarning kombinatsiyasi qo‘l bilan try/catchsiz aniq nosozliklarni boshqarish strategiyasi bilan ishonchli pipeline’lar qurish imkonini beradi.

Schedulers: RxJava’da oqimlarni boshqarish

Schedulers Observable va Observer qaysi oqimda bajarilishini aniqlaydi. subscribeOn manba uchun oqimni belgilaydi, observeOn — Observer va keyingi operatorlar uchun oqimni. Bu ajratish — RxJava ning asosiy ustunligi: manba IO oqimida, qayta ishlash computation’da, UI — asosiy oqimda.

Asosiy Schedulers: Schedulers.io() — I/O operatsiyalari (tarmoq, disk) uchun, cheksiz hovuz. Schedulers.computation() — hisob-kitoblar uchun, yadrolar soniga qarab sobit hovuz. Schedulers.newThread() — har bir vazifa uchun yangi oqim. AndroidSchedulers.mainThread() — Android asosiy oqimi (RxAndroid). Shuningdek testlar uchun foydali FIFO navbati bilan joriy oqimda vazifalarni bajarish uchun Schedulers.trampoline() mavjud.

Google ma’lumotlariga ko‘ra (2025), Schedulers’dan to‘g‘ri foydalanish RxJava’da yangi boshlanuvchilar uchun eng qiyin qismidir. Odatiy xato — observeOn’dan keyin subscribeOn’ni chaqirish, bu manbaga ta’sir qilmaydi. subscribeOn manba uchun zanjirda birinchi bo‘lishi kerak, observeOn — UI obunasidan oldin. Qoida: subscribeOn faqat upstream’ga (manba) ta’sir qiladi, observeOn downstream’ni (obunachi va undan keyingi barcha operatorlar) o‘zgartiradi.

Android’da RxJava bilan kod namunalari

Uch stsenariyni ko‘rib chiqamiz: Single bilan tarmoq so‘rovi, zip bilan parallel so‘rovlar va debounce bilan qidiruv maydoni uchun debounce.

Single bilan tarmoq so‘rovi

Single Retrofit so‘rovlari uchun idealdir: bitta so‘rov — bitta javob. UI yangilash uchun asosiy oqimda obuna.

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 bilan parallel so‘rovlar

zip ikkita mustaqil Single natijalarini bittasiga birlashtiradi. Parallel bajariladi, natija — ikkalasi tugagandan so‘ng.

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))

Qidiruv maydoni uchun Debounce

debounce matnning tez o‘zgarishlarini e’tiborsiz qoldiradi va so‘rovni faqat 400 ms pauzadan keyin yuboradi. distinctUntilChanged matn o‘zgarmagan bo‘lsa so‘rovni bekor qiladi.

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 vs Kotlin Coroutines: yondashuvlarni solishtirish

RxJava va Kotlin Coroutines bir xil vazifani — asinxron dasturlashni — ammo tubdan farqli yondashuvlar bilan hal qiladi. RxJava Observer naqshiga asoslangan va push-based: manba ma’lumot yuboradi, Observer reaksiya beradi. Korutinlar — pull-based: kod ketma-ket await orqali ma’lumot oladi.

• RxJava — reaktiv, ma’lumot oqimi, >200 operator, push-based, qiyin o‘rganish egri chizig‘i
• Coroutines — ketma-ket, suspend/await, ~40 funksiya, pull-based, sodda sintaksis
• RxJava — etuk (2016), ulkan ekotizim, ammo qiyin o‘rganish egri chizig‘i
• Coroutines — zamonaviy (2018), Google tomonidan yangi kod uchun afzal tanlov
• RxJava — Flowable orqali qutidan chiqqan backpressure, ishlangan bufer strategiyalari
• Coroutines — Flow backpressure bilan yaqinda, ammo JetBrains tomonidan faol rivojlantirilmoqda

Google I/O 2024 ma’lumotlariga ko‘ra, Kotlin Coroutines Android’da yangi asinxron kod uchun tavsiya etilgan yondashuvdir. RxJava mavjud loyihalar uchun qo‘llab-quvvatlanadi. Google bosqichma-bosqich migratsiya uchun o‘tish kutubxonalarini (kotlinx-coroutines-rx3) taqdim etadi. AndroidX (LiveData, Room, Paging 3) ikkala yondashuvni qo‘llab-quvvatlaydi, eski modullarda RxJava va yangi modullarda korutinlardan bog‘liqlik mojarolarisiz foydalanish imkonini beradi.

RxJava’dan korutinlarga migratsiya strategiyasi

Bosqichma-bosqich o‘tish: har bir yangi komponent korutinlarda yoziladi, eski RxJava kodi o‘zgartirilmaydi. RxJava → korutinlar awaitSingle() yoki awaitFirst() orqali. Korutinlar → RxJava future() yoki asFlowable() orqali. Katta loyihalar uchun to‘liq migratsiya 6–18 oy davom etadi.