RxJava: əsaslar, ReactiveX və məlumat axınları ilə iş

RxJava və ReactiveX nədir

RxJava — ReactiveX spesifikasiyasının Java tətbiqi, müşahidə olunan axınlar (Observable) vasitəsilə asinxron proqramlaşdırma üçün kitabxanadır. RxJava 2 2016-cı ildə Reactive Streams (Flowable) dəstəyi və rx.Observable ilə io.reactivex.Observable bölgüsü ilə buraxılmışdır. RxJava 3 (2019) — RxJava 2 ilə geriyə uyğunluğu olan cari əsas versiyadır.

RxJava-nın əsas ideyası — hər şey axındır: məlumat axını, hadisə axını, vəziyyət axını. İstənilən asinxron əməliyyatı məlumat, xəta və ya tamamlanma siqnalı göndərən Observable kimi təqdim etmək olar. Observer Observable-a abunə olur və real vaxtda bildirişlər alır.

Badoo məlumatlarına görə (2024), korutinlərə keçiddən əvvəl top-200 Google Play-dən 76% Android tətbiqi asinxron əməliyyatlar üçün RxJava istifadə edirdi. Hal-hazırda pay korutinlərin lehinə azalır, lakin RxJava minlərlə tətbiqin istehsalat kodunda qalır və yetkin, sınaqdan keçmiş texnologiya hesab olunur. ReactiveX — JavaScript (RxJS), .NET (Rx.NET), Swift (RxSwift) və digər dillər üçün də tətbiq edilmiş çoxplatformalı spesifikasiyadır.

RxJava-da Observer naxışı

ReactiveX klassik Observer naxışını iki mexanizmlə genişləndirir: operator zənciri (operator chaining) və axın idarəetməsi (schedulers). Observable Observer abunə olana qədər məlumat göndərməyə başlamır (lazy evaluation). Bu, yalnız abunə olduqda aktivləşən məlumat pipeline-i qurmağa imkan verir.

Observable növləri: Observable, Flowable, Single, Maybe, Completable

Observable — onError və ya onComplete ilə 0..N element göndərən əsas tip. Məhdudiyyətsiz uzunluqlu məlumat axınları üçün uyğundur — məsələn, klik hadisələri və ya geolokasiya yenilənmələri. Observable backpressure dəstəkləmir.

Flowable — backpressure dəstəyi ilə Reactive Streams versiyası. Mənbə elementləri Observer-in emal edə biləcəyindən daha süratli yarada bildikdə istifadə olunur. Flowable BACKPRESSURE_BUFFER, DROP, LATEST və ERROR strategiyalarını dəstəkləyir.

Single, Maybe və Completable

Single dəqiq bir element və ya xəta göndərir — şəbəkə sorğuları üçün idealdır. Maybe — 0 və ya 1 element, məlumatın olmaya biləcəyi keş üçün uyğundur. Completable — yalnız onComplete və ya onError, məlumat olmadan, yazı və ya silmə əməliyyatları üçün əlverişlidir. Bu tiplər API-ni sadələşdirir, kontraktı konkret hala daraldır. Retrofit (Android üçün məşhur HTTP kliyenti) bütün beş RxJava tipini birbaşa dəstəkləyir, hər endpoint üçün lazımsız örtük olmadan ən uyğun qaytarma tipini seçməyə imkan verir.

RxJava operatorları: axınların transformasiyası və filtrlənməsi

Operatorlar — bir Observable-ı digərinə çevirən funksiyalardır. Operator zənciri (operator chain) məlumat pipeline-ni təsvir edir: hər operator əvvəlkindən axını qəbul edir, onu transformasiya edir və növbətiyə ötürür. RxJava kateqoriyalara bölünmüş 200-dən çox operator ehtiva edir.

• map — hər elementi çevirir (Integer → String)
• flatMap — elementi Observable-ə çevirir və hamısını bir axında birləşdirir
• filter — elementləri şərtə görə buraxır
• zip — N Observable elementlərini indeks üzrə birləşdirir
• merge — bir neçə Observable-ı birinə birləşdirir, vaxt sırasını qoruyur
• debounce — elementlər arasında göstərilən intervaldan az olduqda buraxır

flatMap — RxJava-nın ən güclü operatorlarından biridir. Hər element üçün asinxron sorğu icra etməyə və nəticələri ümumi axında toplamağa imkan verir. Məsələn, flatMap ID siyahısı üzrə təfərrüatları yükləmək üçün istifadə olunur: hər ID → şəbəkə sorğu → nəticələrin birləşdirilməsi. Sadəcə elementi çevirən map-dən fərqli olaraq, flatMap bir neçə element göndərə və ya başqa Observable-a keçə bilər, bu da onu asinxron pipeline-lər qurmaq üçün əsas edir.

Operatorlar vasitəsilə xəta idarəetməsi

onErrorResumeNext — xəta olduqda ehtiyat Observable-a keçir. retry — xəta olduqda abunəni N dəfə təkrarlayır. onErrorReturn — xəta əvəzinə standart dəyər qaytarır. doOnError — axını dəyişmədən xəta zamanı kənar təsir həyata keçirir (loglama və ya analitika). Bu operatorların birləşməsi əl ilə try/catch olmadan aydın nasazlıq idarəetmə strategiyası ilə etibarlı pipeline-lər qurmağa imkan verir.

Schedulers: RxJava-da axınların idarə edilməsi

Schedulers Observable və Observer-in hansı axında icra olunduğunu müyyənləşdirir. subscribeOn mənbə üçün axını təyin edir, observeOn — Observer və sonrakı operatorlar üçün axını. Bu ayrılma RxJava-nın əsas üstünlüyüdür: mənbə IO axınında, emal computation-da, UI — əsas axında.

Əsas Schedulers: Schedulers.io() — I/O əməliyyatları üçün (şəbəkə, disk), limitsiz hovuz. Schedulers.computation() — hesablamalar üçün, növə sayına görə sabit hovuz. Schedulers.newThread() — hər tapşırıq üçün yeni axın. AndroidSchedulers.mainThread() — Android əsas axını (RxAndroid). Həmçinin testlər üçün faydalı olan FIFO növbəsi ilə cari axında tapşırıqları icra etmək üçün Schedulers.trampoline() mövcuddur.

Google məlumatlarına görə (2025), Schedulers-dən düzgün istifadə RxJava-da yeni başlayanlar üçün ən çətindir. Tipik səhv — observeOn-dan sonra subscribeOn çağırmaq, bu mənbəyə təsir etmir. subscribeOn mənbə üçün zəncirdə birinci olmalıdır, observeOn — UI abunəsindən əvvəl. Qayda: subscribeOn yalnız upstream-ə (mənbə) təsir edir, observeOn downstream-i (abunəçi və ondan sonrakı bütün operatorlar) dəyişdirir.

Android-də RxJava ilə kod nümunələri

Üç ssenarini nəzərdən keçirək: Single ilə şəbəkə sorğu, zip ilə paralel sorğular və debounce ilə axtarış sahəsi üçün debounce.

Single ilə şəbəkə sorğu

Single Retrofit sorğulari üçün idealdır: bir sorğu — bir cavab. UI yenilənməsi üçün əsas axında abunə.

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 ilə paralel sorğular

zip iki müstəqil Single-in nəticələrini birinə birləşdirir. Paralel icra olunur, nəticə — hər ikisi tamamlandıqdan sonra.

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

Axtarış sahəsi üçün Debounce

debounce mətnin sürətli dəyişmələrinə məhəl qoymur və sorğunu yalnız 400 ms fasilədən sonra göndərir. distinctUntilChanged mətn dəyişməyibsə sorğunu ləğv edir.

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: yanaşmaların müqayisəsi

RxJava və Kotlin Coroutines eyni tapşırığı — asinxron proqramlaşdırmanı — lakin prinsipial fərqli yanaşmalarla həll edir. RxJava Observer naxışı üzərində qurulub və push-based-dir: mənbə məlumat göndərir, Observer reaksiya verir. Korutinlər — pull-based: kod ardıcıl olaraq await vasitəsilə məlumat alır.

• RxJava — reaktiv, məlumat axını, >200 operator, push-based, çətin öyrənmə əyrisi
• Coroutines — ardıcıl, suspend/await, ~40 funksiya, pull-based, sadə sintaksis
• RxJava — yetkin (2016), böyük ekosistem, lakin çətin öyrənmə əyrisi
• Coroutines — müasir (2018), Google tərəfindən yeni kod üçün üstünlük verilən seçim
• RxJava — Flowable vasitəsilə qutudan çıxan backpressure, işlənmiş bufer strategiyaları
• Coroutines — Flow backpressure ilə yaxınlarda, lakin JetBrains tərəfindən aktiv inkişaf etdirilir

Google I/O 2024 məlumatlarına görə, Kotlin Coroutines Android-də yeni asinxron kod üçün tövsiyə olunan yanaşmadır. RxJava mövcud layihələr üçün dəstəklənır. Google tədrici miqrasiya üçün keçid kitabxanaları (kotlinx-coroutines-rx3) təqdim edir. AndroidX (LiveData, Room, Paging 3) hər iki yanaşmanı dəstəkləyir, köhnə modullarda RxJava, yeni modullarda korutinlərdən asılılıq münagişəsi olmadan istifadə etməyə imkan verir.

RxJava-dan korutinlərə miqrasiya strategiyası

Tədrici keçid: hər yeni komponent korutinlərdə yazılır, köhnə RxJava kodu dəyişdirilmir. RxJava → korutinlər awaitSingle() və ya awaitFirst() vasitəsilə. Korutinlər → RxJava future() və ya asFlowable() vasitəsilə. Böyük layihələr üçün tam miqrasiya 6–18 ay çəkir.