RxJava: mahiyyəti, komponentləri və reaktiv proqramlaşdırma

RxJava nədir?

RxJava — Java virtual maşını üçün ReactiveX (Reactive Extensions) kitabxanasının tətbiqidir. RxJava-nın ilk versiyası Netflix şirkəti tərəfindən 2013-cü ildə server tətbiqlərində asinxron çağırışları idarə etmək üçün buraxılmışdır. Yaradılma anında Java-da əsas alternativ Future və Callback idi — hər iki yanaşma callback-hell və mürəkkəb axın idarəetməsinə gətirib çıxarırdı. RxJava funksional operator zəncirləri ilə Observable vasitəsilə asinxron əməliyyatların kompozisiyasını təklif etdi.

RxJava arxitekturası Reactive Streams spesifikasiyasına — bloklanmayan backpressure ilə asinxron axın emalı standartına əsaslanır. Spesifikasiya dörd interfeys müəyyən edir: Publisher, Subscriber, Subscription və Processor. RxJava 2+ Flowable tipi vasitəsilə Reactive Streams-i tam həyata keçirir, RxJava 1-dən fərqli olaraq backpressure kontraktlarına riayət edir. RxJava 2-də Observable backpressure dəstəkləmir — az sayda hadisə və ya UI hadisələri olan axınlar üçün nəzərdə tutulub.

JetBrains, 2025 sorğusuna görə, RxJava Android inkişafı üçün top-3 kitabxanalar sırasındadır. Əsas istifadə ssenariləri: Retrofit vasitəsilə şəbəkə sorğularının emalı (CallAdapter vasitəsilə RxJava ilə inteqrasiya olunub), Room ilə iş (reaktiv sorğular Flowable və ya Maybe qaytarır), RxBinding vasitəsilə animasiyalar və UI hadisələri və mətn daxil edərkən debounce-axtarış. Bütün bu ssenariləri eyni tipli zəncir birləşdirir: mənbə (Observable) → transformasiya (operatorlar) → abunə (subscribe).

RxJava versiya tarixi

RxJava 1 (2013) Observable və operator konsepsiyasını qoydu, lakin backpressure ilə bağlı problemlərdən əziyyət çəkirdi — sürətli axınlarda məlumatlar yaddaşda toplanaraq OutOfMemoryError-a səbəb olurdu. RxJava 2 (2016) arxitekturanı düzəltdi, Observable (backpressure olmadan) və Flowable (backpressure ilə) böldü. RxJava 3 (2020) Java 8 Stream API dəstəyi, əlavə operatorlar və abunə zamanı təkmilləşdirilmiş performans əlavə etdi. Hal-hazırda RxJava 3 — yeni layihələr üçün tövsiyə olunan versiyadır.

RxJava-da reaktiv axın növləri

RxJava beş əsas reaktiv mənbə tipi təmin edir, hər biri müəyyən ssenariyə yönəlmişdir. Observable və Flowable çoxsaylı dəyərlər buraxır, Single — bir dəyər və ya xəta, Completable — yalnız məlumatsız tamamlanma faktı, Maybe — bir dəyər, sıfır və ya xəta. Doğru tipin seçilməsi kod miqdarını azaldır və zənciri özünü-sənədləşdirən edir.

Flowable — böyük məlumat axınları ilə iş üçün ən çevik tip. Backpressure dəstəyi ilə Reactive Streams Publisher həyata keçirir: consumer Subscription.request(n) vasitəsilə müəyyən sayda element tələb edə bilər. Bu, producer və consumer sürətləri uyğun gəlmədikdə buferin daşmasının qarşısını alır. Backpressure kritik deyilsə — Observable istifadə edin, request mexanizminin olmaması səbəbindən daha az yükə malikdir.

Single — HTTP sorğuları üçün optimal seçimdir. Retrofit 2 RxJava CallAdapter ilə hər sorğu üçün Single<ResponseBody> qaytarır. Single dəqiq bir onSuccess və ya onError çağırışına zəmanət verir ki, bu da HTTP sorğusunun semantikasına uyğundur — bir cavab və ya bir xəta. Completable məlumat qaytarmayan yazma əməliyyatları üçün istifadə olunur: insert, update, delete. Maybe keşi yoxlayarkən rahatdır — dəyər qaytara bilər, qaytarmaya da bilər.

// Single-in HTTP sorğusu üçün istifadə nümunəsi
interface ApiService {
    @GET("users/{id}")
    fun getUser(@Path("id") userId: Int): Single<User>
}

// Əsas axında emal ilə abunə
apiService.getUser(42)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe({ user ->
        textView.text = user.name
    }, { error ->
        Log.e("API", "Error: ${error.message}")
    })
    .addTo(compositeDisposable)

Transformasiya və axın idarəetmə operatorları

Operatorlar RxJava — bunlar bir reaktiv mənbə qəbul edib digərini qaytaran, məlumat axınını transformasiya edən yüksək səviyyəli funksiyalardır. RxJava 3 kateqoriyalara bölünmüş 400-dən çox operator ehtiva edir: transformasiya, filtrasiya, birləşdirmə, xəta idarəetməsi və vaxt idarəetməsi. Hər bir operator tənbəldir — zəncir deklarasiya zamanı qurulur, abunə zamanı icra olunur.

Transformasiya operatorları

map — hər bir dəyəri funksiya vasitəsilə çevirən əsas operatordur. flatMap hər bir element üçün Observable qaytaran funksiya qəbul edir və nəticəni vahid axına açır. switchMap flatMap-a bənzəyir, lakin yeni element gəldikdə əvvəlki Observable-dan abunəni ləğv edir. concatMap elementlərin sırasını qoruyur — flatMap-dən fərqli olaraq, hər bir iç-içə Observable-a ardıcıl olaraq abunə olur.

// JSON-un transformasiya və filtrasiya ilə pars edilməsi
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("Xəta", it.message) })

Axınların birləşdirilməsi — RxJava-nın xüsusilə güclü olduğu sahədir. zip bir neçə Observable-dan elementləri indeks üzrə cüt-cüt birləşdirir: birinci birinci ilə, ikinci ikinci ilə. combineLatest axınların hər hansı biri dəyişdikdə yeni dəyər buraxır, bütün axınların son dəyərlərini birləşdirir. merge bir neçə Observable-ı birinə birləşdirir, hadisələrin gəliş sırasını qoruyur. concat hər Observable-a ardıcıl abunə olur və növbətiyə keçməzdən əvvəl onun bütün hadisələrini ötürür.

Vaxt idarəetməsi debounce (göndərmədən əvvəl axında fasilə gözləmə), throttleFirst (ilk hadisəni buraxmaq, qalanlarını pəncərə ərzində görməməzlik), timeout (hadisə interval ərzində gəlməzsə xəta) daxildir. Mətn daxil edərkən debounce-axtarış — ən geniş yayılmış ssenaridir: searchObservable.debounce(300, MILLISECONDS).distinctUntilChanged() sürətli yazma zamanı lazımsız sorğuların qarşısını alır.

Schedulers və çoxaxınlılıq

Scheduler RxJava-da — axın hovuzu üzərində abstraksiyadır. Kitabxana beş daxili Scheduler təmin edir: Schedulers.io() I/O əməliyyatları üçün (şəbəkə, fayllar), Schedulers.computation() CPU intensiv tapşırıqlar üçün, Schedulers.newThread() hər yeni axın üçün, Schedulers.single() təkaxınlı icra üçün və Schedulers.trampoline() cari axında dərhal icra üçün.

subscribeOn və observeOn

subscribeOn Observable mənbəyinin hansı Scheduler-də icra olunduğunu müəyyən edir. Zəncirdə bir neçə subscribeOn varsa — mənbəyə ən yaxın olan prioritetə malikdir. observeOn downstream-i göstərilən Scheduler-ə keçirir — hər observeOn istifadəsi sonrakı operatorlar üçün axını dəyişir. Tipik Android nümunəsi: subscribeOn(Schedulers.io()) şəbəkə ilə iş üçün, observeOn(AndroidSchedulers.mainThread()) UI yeniləmək üçün.

// Kontekst dəyişdirmə ilə çoxaxınlı emal
Observable.fromCallable(() -> database.getItems())
    .subscribeOn(Schedulers.io())            // BD io-da
    .map(items -> processItems(items))     // transformasiya io-da
    .observeOn(Schedulers.computation())    // computation-a keçirik
    .map(processed -> compressImages(processed))
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(result -> ui.showResult(result))

AndroidSchedulers.mainThread() — RxAndroid kitabxanasından Scheduler, Android-in əsas axınında kod icra edir. Reaktiv zəncirdə hər hansı UI yeniləmələri üçün məcburidir. Kitabxana daxildə Handler istifadə edir və yüksək yük altında belə UI axınında icraya zəmanət verir. Fon əməliyyatları üçün Schedulers.io() limitsiz axın hovuzunu dəstəkləyir və istənilən bloklayan əməliyyatlar üçün uyğundur. Schedulers.computation() prosessor nüvələrinin sayına bərabər sabit hovuzdan istifadə edir.

Android-də RxJava: praktik tətbiq

RxJava Android-də üç əsas ssenari üçün istifadə olunur: Room-a reaktiv sorğular, Retrofit ilə inteqrasiya və RxBinding vasitəsilə UI-nin reaktiv bağlanması. Hər ssenari üçün öz tip dəsti xarakterikdir: Room müşahidə olunan sorğular üçün Flowable, Retrofit — HTTP sorğuları üçün Single, RxBinding — UI hadisələri üçün Observable qaytarır.

Room + RxJava

Room — Google-dan məlumatların davamlılıq kitabxanası. Room 2.1-dən başlayaraq, verilənlər bazası reaktiv qaytarılan tipləri dəstəkləyir: Flowable və Observable. Cədvəldə hər hansı qeyd dəyişdikdə Room avtomatik olaraq axına yeni dəyər göndərir. Developer ViewModel-də Flowable-a abunə olur və hər dəyişiklikdə əl ilə sorğu olmadan aktual məlumatları alır.

// Reaktiv sorğu ilə 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 kompozisiyası
class UserViewModel(private val dao: UserDao) : ViewModel() {
    val users: Flowable<List<User>> = dao.getAllUsers()
        .subscribeOn(Schedulers.io())
}

MVVM + RxJava nümunəsi ViewModel-in View-ə istinadlarının olmamasına əsaslanır. ViewModel reaktiv mənbələr (Flowable, Transformations vasitəsilə LiveData) dərc edir, Activity və ya Fragment isə onlara abunə olur. Bu, test edilə bilənliyi təmin edir: ViewModel UI olmadan, RxJavaPlugins.setComputationScheduler vasitəsilə Scheduler-i əvəz edərək test edilir. ViewModel-də CompositeDisposable abunəliklərin həyat dövrünü idarə edir — onCleared() zamanı bütün abunəliklər ləğv olunur.

RxJava vs Kotlin Flow

Kotlin Flow — Kotlin-də soyuq axınların yerli tətbiqi, korutinlərə daxil edilmiş və Kotlin 1.3-də təqdim edilmişdir. Flow RxJava ilə eyni vəzifələri həll edir, lakin əsas fərqlərlə: korutinlərin daxili dəstəyi (suspend-funksiyalar), coroutine cancellation vasitəsilə ləğv etmə və backpressure ilə bağlı problemlərin olmaması — Flow buferləşdirmə əvəzinə suspend istifadə edir. Flow əlavə asılılıqlar tələb etməyən Kotlin standart kitabxanasının bir hissəsidir.

RxJava Java layihələri, Java 7-8 dəstəyi olan layihələr və RxJava-da mövcud kod bazaları üçün üstünlük verilən seçim olaraq qalır. RxJava ekosistemi əhəmiyyətli dərəcədə zəngindir: Flow-da ~50 operatora qarşı >400 operator, daxili CallAdapter vasitəsilə Retrofit ilə inteqrasiya, Flowable vasitəsilə backpressure dəstəyi və Android üçün RxBinding, RxPermissions, RxLocation-un mövcudluğu. Kotlin Flow sürətlə yetişir, lakin RxJava-nın mürəkkəb axın birləşdirmə ssenarilərində çevikliyi hələ də yüksəkdir.