RxJava: temeller, ReactiveX ve veri akışlarıyla çalışma

RxJava ve ReactiveX Nedir

RxJava, ReactiveX spesifikasyonunun Java uygulamasıdır ve gözlemlenebilir akışlar (Observable) kullanarak asenkron programlama için bir kütüphanedir. RxJava 2, 2016 yılında Reactive Streams (Flowable) desteği ve rx.Observable ile io.reactivex.Observable olarak ayrılma ile yayınlandı. RxJava 3 (2019), RxJava 2 ile geriye dönük uyumluluğa sahip mevcut ana sürümdür.

RxJava'nın temel fikri her şeyin bir akış olduğudur: veri akışı, olay akışı, durum akışı. Herhangi bir asenkron işlem, veri, hata veya tamamlanma sinyali yayan bir Observable olarak temsil edilebilir. Bir Observer, Observable'a abone olur ve gerçek zamanlı bildirimler alır.

Badoo'ya (2024) göre, coroutine'lere geçişten önce, Google Play ilk 200'ündeki Android uygulamalarının %76'sı asenkron işlemler için RxJava kullanıyordu. Pay şimdi coroutine'ler lehine azalıyor, ancak RxJava binlerce uygulamanın üretim kodunda kalmaya devam ediyor ve olgun, kanıtlanmış bir teknoloji olarak kabul ediliyor. ReactiveX, JavaScript (RxJS), .NET (Rx.NET), Swift (RxSwift) ve diğer diller için de uygulanmış çapraz platform bir spesifikasyondur.

RxJava'da Observer Deseni

ReactiveX, klasik Observer desenini iki mekanizma ile genişletir: operatör zincirleme ve Scheduler tabanlı iş parçacığı yönetimi. Observable, bir Observer abone olana kadar veri yaymaya başlamaz (tembel değerlendirme). Bu, yalnızca abonelik olduğunda etkinleşen bir veri işlem hattı oluşturmayı sağlar.

Observable Türleri: Observable, Flowable, Single, Maybe, Completable

Observable — onError veya onComplete ile 0..N öğe yayan temel tür. Sınırsız veri akışları için uygundur — örneğin, tıklama olayları veya coğrafi konum güncellemeleri. Observable geri basıncı (backpressure) desteklemez.

Flowable — geri basınç desteğine sahip Observable'ın Reactive Streams sürümü. Veri kaynağının, Observer'ın işleyebileceğinden daha hızlı öğe üretebildiği durumlarda kullanılır. Flowable, BACKPRESSURE_BUFFER, DROP, LATEST ve ERROR stratejilerini destekler.

Single, Maybe ve Completable

Single tam olarak bir öğe veya hata yayar — ağ istekleri için idealdir. Maybe 0 veya 1 öğe yayar, verilerin bulunmayabileceği önbellek için uygundur. Completable veri olmadan yalnızca onComplete veya onError yayar, yazma veya silme işlemleri için kullanışlıdır. Bu türler, sözleşmeyi belirli bir duruma daraltarak API'yi basitleştirir. Retrofit (Android için popüler bir HTTP istemcisi) beş RxJava türünü de doğrudan destekler ve her uç nokta için fazladan kod olmadan en uygun dönüş türünü seçmenize olanak tanır.

RxJava Operatörleri: Akış Dönüşümü ve Filtreleme

Operatörler, bir Observable'ı başka bir Observable'a dönüştüren işlevlerdir. Bir operatör zinciri, veri işlem hattını tanımlar: her operatör akışı bir öncekinden alır, dönüştürür ve bir sonrakine iletir. RxJava, kategorilere ayrılmış 200'den fazla operatör içerir.

• map — her öğeyi dönüştürür (Integer → String)
• flatMap — bir öğeyi Observable'a dönüştürür ve hepsini tek bir akışta birleştirir
• filter — bir koşulu karşılayan öğeleri geçirir
• zip — N Observable'dan öğeleri dizine göre birleştirir
• merge — birden çok Observable'ı kronolojik sırayı koruyarak tek bir akışta birleştirir
• debounce — belirtilen zaman aralığında başka yayın olmazsa öğeleri yayar

flatMap, RxJava'nın en güçlü operatörlerinden biridir. Her öğe için asenkron bir istek yürütmeye ve sonuçları ortak bir akışta toplamaya olanak tanır. Örneğin, flatMap bir ID listesinden ayrıntıları yüklemek için kullanılır: her ID → ağ isteği → sonuçların birleştirilmesi. Yalnızca bir öğeyi dönüştüren map'in aksine, flatMap birden çok öğe yayabilir veya başka bir Observable'a geçebilir, bu da onu asenkron işlem hatları oluşturmanın temeli haline getirir.

Operatörlerle Hata Yönetimi

onErrorResumeNext — hata durumunda yedek Observable'a geçer. retry — hata durumunda N kez yeniden abone olur. onErrorReturn — hata yerine varsayılan bir değer döndürür. doOnError — akışı değiştirmeden hata durumunda yan etki yürütür (günlükleme veya analiz). Bu operatörlerin birleşimi, manuel try/catch olmadan net bir hata yönetimi stratejisine sahip sağlam işlem hatları oluşturmayı sağlar.

Schedulers: RxJava'da İş Parçacığı Yönetimi

Schedulers, Observable ve Observer'ın hangi iş parçacığında yürütüleceğini belirler. subscribeOn kaynak için iş parçacığını ayarlar, observeOn Observer ve sonraki operatörler için iş parçacığını ayarlar. Bu ayrım RxJava'nın önemli bir avantajıdır: kaynak IO iş parçacığında, işleme computation'da, UI ana iş parçacığında çalışır.

Ana Schedulers: Schedulers.io() — G/Ç işlemleri (ağ, disk) için, sınırsız havuz. Schedulers.computation() — hesaplamalar için, çekirdek sayısına göre sabit havuz. Schedulers.newThread() — her görev için yeni bir iş parçacığı. AndroidSchedulers.mainThread() — Android ana iş parçacığı (RxAndroid). Ayrıca FIFO kuyruğu ile geçerli iş parçacığında görevleri yürütmek için Schedulers.trampoline() da vardır, testler için kullanışlıdır.

Google'a (2025) göre, Schedulers'ın doğru kullanımı RxJava'da yeni başlayanlar için en zor kısımdır. Tipik bir hata, observeOn'den sonra subscribeOn'u çağırmaktır, bu kaynağı etkilemez. subscribeOn kaynak için zincirde ilk olmalı, observeOn UI aboneliğinden önce olmalıdır. Kural: subscribeOn yalnızca yukarı akışı (kaynak) etkiler, observeOn aşağı akışı (abone ve ondan sonraki tüm operatörler) değiştirir.

Android'de RxJava Kod Örnekleri

Üç senaryoyu ele alalım: Single ile ağ isteği, zip ile paralel istekler ve debounce ile arama alanı için debounce.

Single ile Ağ İsteği

Single, Retrofit istekleri için mükemmeldir: bir istek — bir yanıt. UI güncellemeleri için ana iş parçacığında abone olun.

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 ile Paralel İstekler

zip, iki bağımsız Single'ın sonuçlarını birleştirir. Paralel olarak yürütülür, sonuç her ikisi de tamamlandıktan sonra üretilir.

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

Arama Alanı için Debounce

debounce hızlı metin değişikliklerini yok sayar ve yalnızca 400 ms duraklamadan sonra istek gönderir. distinctUntilChanged metin değişmediyse isteği iptal eder.

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: Yaklaşımların Karşılaştırması

RxJava ve Kotlin Coroutines aynı sorunu — asenkron programlama — ancak temelde farklı yaklaşımlarla çözer. RxJava, Observer deseni üzerine kuruludur ve push tabanlıdır: kaynak veri gönderir, Observer tepki verir. Coroutine'ler pull tabanlıdır: kod, await aracılığıyla sırayla veri talep eder.

• RxJava — reaktif, veri akışı, >200 operatör, push tabanlı, dik öğrenme eğrisi
• Coroutines — sıralı, suspend/await, ~40 işlev, pull tabanlı, basit sözdizimi
• RxJava — olgun (2016), büyük ekosistem, ancak dik öğrenme eğrisi
• Coroutines — modern (2018), Google'ın yeni kod için tercih ettiği seçenek
• RxJava — Flowable aracılığıyla yerleşik geri basınç, iyi test edilmiş tamponlama stratejileri
• Coroutines — Geri basınçlı Flow yenidir, ancak JetBrains tarafından aktif olarak geliştirilmektedir

Google I/O 2024'e göre, Kotlin Coroutines, Android'de yeni asenkron kod için önerilen yaklaşımdır. RxJava mevcut projeler için desteklenmeye devam etmektedir. Google, kademeli geçiş için köprü kütüphaneleri (kotlinx-coroutines-rx3) sağlar. AndroidX (LiveData, Room, Paging 3) her iki yaklaşımı da destekler, bağımlılık çakışmaları olmadan eski modüllerde RxJava ve yenilerinde coroutine kullanılmasına olanak tanır.

RxJava'dan Coroutine'lere Geçiş Stratejisi

Kademeli geçiş: her yeni bileşen coroutine'lerle yazılır, eski RxJava kodu değiştirilmez. RxJava → coroutine'ler awaitSingle() veya awaitFirst() aracılığıyla. Coroutine'ler → RxJava future() veya asFlowable() aracılığıyla. Büyük projeler için tam geçiş 6–18 ay sürer.