RxJava: mohiyati, komponentlari va reaktiv dasturlash

RxJava nima?

RxJava — Java virtual mashinasi uchun ReactiveX (Reactive Extensions) kutubxonasining tatbiqidir. RxJava-ning birinchi versiyasi Netflix kompaniyasi tomonidan 2013-yilda server ilovalarida asinxron chaqiriqlarni boshqarish uchun chiqarilgan. Yaratilish vaqtida Java-da asosiy muqobil Future va Callback edi — ikkala yondashuv ham callback-hell va murakkab oqim boshqaruviga olib kelardi. RxJava funksional operator zanjirlari bilan Observable orqali asinxron operatsiyalar kompozitsiyasini taklif qildi.

RxJava arxitekturasi Reactive Streams spetsifikatsiyasiga — bloklanmaydigan backpressure bilan asinxron oqimlarni qayta ishlash standartiga asoslanadi. Spetsifikatsiya to'rtta interfeysni belgilaydi: Publisher, Subscriber, Subscription va Processor. RxJava 2+ Flowable tipi orqali Reactive Streams-ni to'liq amalga oshiradi, RxJava 1 dan farqli ravishda backpressure kontraktlariga rioya qiladi. RxJava 2-dagi Observable backpressure-ni qo'llab-quvvatlamaydi — u kam sonli hodisalar yoki UI hodisalari bo'lgan oqimlar uchun mo'ljallangan.

JetBrains, 2025 so'roviga ko'ra, RxJava Android ishlanmasi uchun top-3 kutubxonalar qatoriga kiradi. Asosiy foydalanish stsenariylari: Retrofit orqali tarmoq so'rovlarini qayta ishlash (CallAdapter orqali RxJava bilan integratsiyalangan), Room bilan ishlash (reaktiv so'rovlar Flowable yoki Maybe qaytaradi), RxBinding orqali animatsiyalar va UI hodisalari va matn kiritishda debounce-qidirish. Barcha bu stsenariylarni bir xil turdagi zanjir birlashtiradi: manba (Observable) → transformatsiya (operatorlar) → obuna (subscribe).

RxJava versiya tarixi

RxJava 1 (2013) Observable va operator kontseptsiyasini yaratdi, lekin backpressure bilan bog'liq muammolardan aziyat chekdi — tez oqimlarda ma'lumotlar xotirada to'planib, OutOfMemoryError-ga sabab bo'lardi. RxJava 2 (2016) arxitekturani tuzatdi, Observable (backpressure-siz) va Flowable (backpressure bilan) ni ajratdi. RxJava 3 (2020) Java 8 Stream API qo'llab-quvvatlashi, qo'shimcha operatorlar va obuna vaqtida yaxshilangan samaradorlikni qo'shdi. Hozirda RxJava 3 — yangi loyihalar uchun tavsiya etilgan versiyadir.

RxJava-dagi reaktiv oqim turlari

RxJava besh asosiy reaktiv manba turini taqdim etadi, ularning har biri ma'lum stsenariyga yo'naltirilgan. Observable va Flowable ko'p qiymatlarni chiqaradi, Single — bitta qiymat yoki xato, Completable — faqat ma'lumotsiz tugallanish fakti, Maybe — bitta qiymat, nol yoki xato. To'g'ri turni tanlash kod miqdorini kamaytiradi va zanjirni o'zini-o'zi hujjatlashtiradigan qiladi.

Flowable — katta ma'lumot oqimlari bilan ishlash uchun eng moslashuvchan tur. Backpressure qo'llab-quvvatlashi bilan Reactive Streams Publisher-ni amalga oshiradi: consumer Subscription.request(n) orqali ma'lum miqdordagi elementlarni so'rashi mumkin. Bu producer va consumer tezliklari mos kelmaganda buferning to'lib ketishini oldini oladi. Agar backpressure muhim bo'lmasa — Observable dan foydalaning, u request mexanizmining yo'qligi sababli kamroq yukga ega.

Single — HTTP so'rovlari uchun optimal tanlovdir. Retrofit 2 RxJava CallAdapter bilan har bir so'rov uchun Single<ResponseBody> qaytaradi. Single aniq bitta onSuccess yoki onError chaqiruvini kafolatlaydi, bu HTTP so'rov semantikasiga mos keladi — bitta javob yoki bitta xato. Completable ma'lumot qaytarmaydigan yozish operatsiyalari uchun ishlatiladi: insert, update, delete. Maybe keshni tekshirishda qulay — qiymat qaytarishi mumkin, qaytarmasligi ham mumkin.

// Single ni HTTP so'rovi uchun ishlatish namunasi
interface ApiService {
    @GET("users/{id}")
    fun getUser(@Path("id") userId: Int): Single<User>
}

// Asosiy oqimda qayta ishlash bilan obuna
apiService.getUser(42)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe({ user ->
        textView.text = user.name
    }, { error ->
        Log.e("API", "Error: ${error.message}")
    })
    .addTo(compositeDisposable)

Transformatsiya va oqim boshqaruv operatorlari

Operatorlar RxJava — bu bir reaktiv manbani qabul qilib, boshqasini qaytaradigan, ma'lumot oqimini transformatsiya qiladigan yuqori darajali funksiyalardir. RxJava 3 kategoriyalarga bo'lingan 400 dan ortiq operatorni o'z ichiga oladi: transformatsiya, filtrlash, birlashtirish, xato boshqaruvi va vaqt boshqaruvi. Har bir operator dangasa — zanjir deklaratsiya vaqtida quriladi, obuna vaqtida bajariladi.

Transformatsiya operatorlari

map — har bir qiymatni funksiya orqali o'zgartiradigan asosiy operator. flatMap har bir element uchun Observable qaytaradigan funksiyani qabul qiladi va natijani yagona oqimga ochadi. switchMap flatMap-ga o'xshaydi, lekin yangi element kelganda avvalgi Observable-dan obunani bekor qiladi. concatMap elementlar tartibini saqlaydi — flatMap-dan farqli o'laroq, har bir ichki Observable-ga ketma-ket obuna bo'ladi.

// JSON ni transformatsiya va filtrlash bilan pars qilish
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("Xato", it.message) })

Oqimlarni birlashtirish — RxJava ayniqsa kuchli bo'lgan sohadir. zip bir nechta Observable-dan elementlarni indeks bo'yicha juft-juft birlashtiradi: birinchi birinchi bilan, ikkinchi ikkinchi bilan. combineLatest oqimlarning istalgan biri o'zgarganda yangi qiymat chiqaradi, barcha oqimlarning oxirgi qiymatlarini birlashtiradi. merge bir nechta Observable-ni bittaga birlashtiradi, hodisalarning kelish tartibini saqlaydi. concat har bir Observable-ga ketma-ket obuna bo'ladi va keyingisiga o'tishdan oldin uning barcha hodisalarini uzatadi.

Vaqt boshqaruvi debounce (jo'natishdan oldin oqimda pauza kutish), throttleFirst (birinchi hodisani o'tkazish, qolganlarini oyna davomida e'tiborsiz qoldirish), timeout (hodisa interval davomida kelmasa xato) ni o'z ichiga oladi. Matn kiritishda debounce-qidirish — eng keng tarqalgan stsenariy: searchObservable.debounce(300, MILLISECONDS).distinctUntilChanged() tez yozishda keraksiz so'rovlarning oldini oladi.

Schedulers va ko'p oqimlilik

Scheduler RxJava-da — oqimlar to'plami ustidagi abstraktsiyadir. Kutubxona beshta ichki Scheduler-ni taqdim etadi: Schedulers.io() I/O operatsiyalari uchun (tarmoq, fayllar), Schedulers.computation() CPU intensiv vazifalar uchun, Schedulers.newThread() har bir yangi oqim uchun, Schedulers.single() yagona oqimli bajarish uchun va Schedulers.trampoline() joriy oqimda darhol bajarish uchun.

subscribeOn va observeOn

subscribeOn Observable manbasining qaysi Scheduler-da bajarilishini aniqlaydi. Agar zanjirda bir nechta subscribeOn bo'lsa — manbaga eng yaqini ustunlikka ega. observeOn downstream-ni ko'rsatilgan Scheduler-ga o'tkazadi — har bir observeOn ishlatilishi keyingi operatorlar uchun oqimni o'zgartiradi. Odatdagi Android namunasi: subscribeOn(Schedulers.io()) tarmoq bilan ishlash uchun, observeOn(AndroidSchedulers.mainThread()) UI yangilash uchun.

// Kontekst almashtirish bilan ko'p oqimli qayta ishlash
Observable.fromCallable(() -> database.getItems())
    .subscribeOn(Schedulers.io())            // MB io da
    .map(items -> processItems(items))     // transformatsiya io da
    .observeOn(Schedulers.computation())    // computation ga o'tkazamiz
    .map(processed -> compressImages(processed))
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(result -> ui.showResult(result))

AndroidSchedulers.mainThread() — RxAndroid kutubxonasidan Scheduler, Android-ning asosiy oqimida kod bajaradi. Reaktiv zanjirda har qanday UI yangilanishlari uchun majburiydir. Kutubxona ichki Handler-dan foydalanadi va yuqori yuk ostida ham UI oqimida bajarilishni kafolatlaydi. Fon operatsiyalari uchun Schedulers.io() cheksiz oqimlar to'plamini qo'llab-quvvatlaydi va har qanday bloklovchi operatsiyalar uchun mos keladi. Schedulers.computation() protsessor yadrolari soniga teng sobit to'plamdan foydalanadi.

Android-da RxJava: amaliy qo'llanish

RxJava Android-da uchta asosiy stsenariy uchun ishlatiladi: Room-ga reaktiv so'rovlar, Retrofit bilan integratsiya va RxBinding orqali UI-ni reaktiv bog'lash. Har bir stsenariy uchun o'ziga xos turlar to'plami xarakterlidir: Room kuzatiladigan so'rovlar uchun Flowable, Retrofit — HTTP so'rovlari uchun Single, RxBinding — UI hodisalari uchun Observable qaytaradi.

Room + RxJava

Room — Google dan ma'lumotlarning doimiylik kutubxonasi. Room 2.1 dan boshlab, ma'lumotlar bazasi reaktiv qaytariladigan turlarni qo'llab-quvvatlaydi: Flowable va Observable. Jadvaldagi istalgan yozuv o'zgarganda, Room avtomatik ravishda oqimga yangi qiymat yuboradi. Dasturchi ViewModel-da Flowable-ga obuna bo'ladi va har bir o'zgarishda qo'lda so'rovlarsiz dolzarb ma'lumotlarni oladi.

// Reaktiv so'rov bilan 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 kompozitsiyasi
class UserViewModel(private val dao: UserDao) : ViewModel() {
    val users: Flowable<List<User>> = dao.getAllUsers()
        .subscribeOn(Schedulers.io())
}

MVVM + RxJava namunasi ViewModel-ning View-ga havolalari yo'qligiga asoslanadi. ViewModel reaktiv manbalarni (Flowable, Transformations orqali LiveData) nashr etadi, Activity yoki Fragment esa ularga obuna bo'ladi. Bu test qilish imkoniyatini ta'minlaydi: ViewModel UI-siz, RxJavaPlugins.setComputationScheduler orqali Scheduler-ni almashtirib test qilinadi. ViewModel-dagi CompositeDisposable obunalarning hayot aylanishini boshqaradi — onCleared() da barcha obunalar bekor qilinadi.

RxJava vs Kotlin Flow

Kotlin Flow — Kotlin-dagi sovuq oqimlarning mahalliy tatbiqi, korutinlarga o'rnatilgan va Kotlin 1.3 da taqdim etilgan. Flow RxJava bilan bir xil vazifalarni hal qiladi, lekin asosiy farqlar bilan: korutinlarning ichki qo'llab-quvvatlanishi (suspend-funksiyalar), coroutine cancellation orqali bekor qilish va backpressure bilan bog'liq muammolarning yo'qligi — Flow buferlash o'rniga suspend dan foydalanadi. Flow qo'shimcha bog'liqliklarni talab qilmaydigan Kotlin standart kutubxonasining bir qismidir.

RxJava Java loyihalari, Java 7-8 qo'llab-quvvatlovchi loyihalar va RxJava-dagi mavjud kod bazalari uchun afzal tanlov bo'lib qolmoqda. RxJava ekotizimi sezilarli darajada boy: Flow-dagi ~50 operatorga qarshi >400 operator, o'rnatilgan CallAdapter orqali Retrofit bilan integratsiya, Flowable orqali backpressure qo'llab-quvvatlashi va Android uchun RxBinding, RxPermissions, RxLocation mavjudligi. Kotlin Flow tezda yetib kelmoqda, lekin RxJava-ning murakkab oqim birlashtirish stsenariylarida moslashuvchanligi hali ham yuqori.