RxJava: özü, bileşenleri ve reaktif programlama

RxJava Nedir?

RxJava, Java Sanal Makinesi için ReactiveX (Reactive Extensions) kütüphanesinin bir uygulamasıdır. RxJava'nın ilk sürümü, sunucu tarafı uygulamalarında asenkron çağrıları yönetmek için Netflix tarafından 2013 yılında yayınlanmıştır. Oluşturulduğu sırada Java'daki ana alternatifler Future ve Callback idi — her iki yaklaşım da callback-cehennemine ve karmaşık iş parçacığı yönetimine yol açıyordu. RxJava, işlevsel operatör zincirleriyle Observable aracılığıyla asenkron işlemlerin kompozisyonunu tanıttı.

RxJava mimarisi, Reactive Streams spesifikasyonuna dayanır — bloke etmeyen geri basınç ile asenkron akış işleme için bir standart. Spesifikasyon dört arayüz tanımlar: Publisher, Subscriber, Subscription ve Processor. RxJava 2+, RxJava 1'in aksine geri basınç sözleşmelerine uyarak Flowable tipi aracılığıyla Reactive Streams'i tamamen uygular. RxJava 2'de Observable geri basıncı desteklemez — az sayıda olay veya UI olayları olan akışlar için tasarlanmıştır.

JetBrains, 2025 anketine göre RxJava, Android geliştirme için ilk 3 kütüphane arasındadır. Ana kullanım durumları şunları içerir: Retrofit aracılığıyla ağ isteklerini işleme (CallAdapter üzerinden RxJava ile entegre), Room ile çalışma (reaktif sorgular Flowable veya Maybe döndürür), RxBinding aracılığıyla animasyonlar ve UI olayları ve metin girişinde debounce arama. Tüm bu senaryolar ortak bir zincir desenini paylaşır: kaynak (Observable) → dönüşüm (operatörler) → abonelik (subscribe).

RxJava Sürüm Geçmişi

RxJava 1 (2013) Observable ve operatörlerle temeli attı, ancak geri basınç sorunlarından muzdaripti — hızlı akışlarda veri bellekte birikerek OutOfMemoryError'a neden oluyordu. RxJava 2 (2016), Observable (geri basınçsız) ve Flowable'ı (geri basınçlı) ayırarak mimariyi düzeltti. RxJava 3 (2020), Java 8 Stream API desteği, ek operatörler ve geliştirilmiş abonelik performansı ekledi. Şu anda RxJava 3, yeni projeler için önerilen sürümdür.

RxJava'da Reaktif Akış Türleri

RxJava, her biri belirli bir senaryo için tasarlanmış beş ana reaktif kaynak türü sağlar. Observable ve Flowable birden çok değer yayar, Single bir değer veya hata yayar, Completable veri olmadan yalnızca tamamlanma yayar ve Maybe bir değer, sıfır veya hata yayar. Doğru türü seçmek kod hacmini azaltır ve zinciri kendi kendini belgeleyen hale getirir.

Flowable, büyük veri akışlarıyla çalışmak için en esnek türdür. Geri basınç desteğiyle Reactive Streams Publisher'ı uygular: tüketici, Subscription.request(n) aracılığıyla belirli sayıda öğe talep edebilir. Bu, üretici ve tüketici hızları uyuşmadığında arabellek taşmasını önler. Geri basınç kritik değilse Observable kullanın — istek mekanizmasının olmaması nedeniyle daha az yükü vardır.

Single, HTTP istekleri için en uygun seçimdir. RxJava CallAdapter ile Retrofit 2, her istek için Single<ResponseBody> döndürür. Single, tam olarak bir onSuccess veya onError çağrısını garanti eder; bu, bir HTTP isteğinin anlamına (bir yanıt veya bir hata) karşılık gelir. Completable, veri döndürmeyen yazma işlemleri (insert, update, delete) için kullanılır. Maybe, önbellek kontrolü için kullanışlıdır — bir değer döndürebilir veya döndürmeyebilir.

// HTTP isteği için Single kullanımı örneği
interface ApiService {
    @GET("users/{id}")
    fun getUser(@Path("id") userId: Int): Single<User>
}

// Ana iş parçacığında işleme ile abonelik
apiService.getUser(42)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe({ user ->
        textView.text = user.name
    }, { error ->
        Log.e("API", "Error: ${error.message}")
    })
    .addTo(compositeDisposable)

Dönüşüm ve Akış Yönetim Operatörleri

Operatörler RxJava'da, bir reaktif kaynağı alıp veri akışını dönüştürerek başka bir kaynak döndüren yüksek dereceli işlevlerdir. RxJava 3, 400'den fazla operatör içerir ve kategorilere ayrılır: dönüşüm, filtreleme, birleştirme, hata yönetimi ve zaman yönetimi. Her operatör tembeldir — zincir bildirimde oluşturulur ve abonelikte yürütülür.

Dönüşüm Operatörleri

map, her değeri bir işlev aracılığıyla dönüştüren temel operatördür. flatMap, her öğe için bir Observable döndüren bir işlev alır ve sonucu tek bir akışta düzleştirir. switchMap, flatMap'e benzer, ancak yeni bir öğe geldiğinde önceki Observable'dan aboneliği iptal eder. concatMap, öğe sırasını korur — flatMap'in aksine, sırayla her iç içe Observable'a abone olur.

// Dönüşüm ve filtreleme ile JSON ayrıştırma
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("Hata", it.message) })

Akış birleştirme, RxJava'nın özellikle güçlü olduğu bir alandır. zip, birden çok Observable'daki öğeleri dizine göre çiftler halinde birleştirir: birinci birinciyle, ikinci ikinciyle. combineLatest, herhangi bir akış değiştiğinde yeni bir değer yayar ve tüm akışlardaki en son değerleri birleştirir. merge, birden çok Observable'ı birleştirerek olay geliş sırasını korur. concat, sırayla her Observable'a abone olur ve bir sonrakine geçmeden önce tüm olaylarını iletir.

Zaman yönetimi, debounce (yaymadan önce akışta bir duraklama bekleme), throttleFirst (ilk olayı yayma, pencere içindeki geri kalanı yok sayma), timeout (aralık içinde hiçbir olay gelmezse hata) içerir. Metin girişinde debounce arama en yaygın senaryodur: searchObservable.debounce(300, MILLISECONDS).distinctUntilChanged() hızlı yazma sırasında gereksiz istekleri önler.

Scheduler'lar ve Çoklu İş Parçacığı

Scheduler RxJava'da bir iş parçacığı havuzu üzerindeki soyutlamadır. Kütüphane beş yerleşik Scheduler sağlar: G/Ç işlemleri (ağ, dosyalar) için Schedulers.io(), CPU yoğun görevler için Schedulers.computation(), her seferinde yeni bir iş parçacığı için Schedulers.newThread(), tek iş parçacıklı yürütme için Schedulers.single() ve geçerli iş parçacığında hemen yürütme için Schedulers.trampoline().

subscribeOn ve observeOn

subscribeOn, kaynak Observable'ı hangi Scheduler'ın yürüteceğini belirler. Zincirde birden çok subscribeOn varsa, kaynağa en yakın olan önceliklidir. observeOn, downstream'i belirtilen Scheduler'a geçirir — her observeOn kullanımı, sonraki operatörler için iş parçacığını değiştirir. Tipik bir Android deseni: ağ işlemleri için subscribeOn(Schedulers.io()), UI güncellemeleri için observeOn(AndroidSchedulers.mainThread()).

// Bağlam değiştirme ile çoklu iş parçacıklı işleme
Observable.fromCallable(() -> database.getItems())
    .subscribeOn(Schedulers.io())            // io'da DB
    .map(items -> processItems(items))     // io'da dönüşüm
    .observeOn(Schedulers.computation())    // computation'a geç
    .map(processed -> compressImages(processed))
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(result -> ui.showResult(result))

AndroidSchedulers.mainThread(), Android ana iş parçacığında kod yürüten RxAndroid kütüphanesinden bir Scheduler'dır. Reaktif bir zincirdeki herhangi bir UI güncellemesi için zorunludur. Kütüphane dahili olarak Handler kullanır ve yüksek yük altında bile UI iş parçacığında yürütmeyi garanti eder. Arka plan işlemleri için Schedulers.io() sınırsız bir iş parçacığı havuzunu destekler ve herhangi bir bloke edici işlem için uygundur. Schedulers.computation() CPU çekirdeği sayısına eşit sabit bir havuz kullanır.

Android'de RxJava: Pratik Uygulamalar

RxJava Android'de üç ana senaryo için kullanılır: Room'a reaktif sorgular, Retrofit ile entegrasyon ve RxBinding aracılığıyla reaktif UI bağlama. Her senaryonun kendi tür kümesi vardır: Room gözlemlenebilir sorgular için Flowable döndürür, Retrofit HTTP istekleri için Single döndürür, RxBinding UI olayları için Observable döndürür.

Room + RxJava

Room, Google'dan bir veri kalıcılık kütüphanesidir. Room 2.1'den itibaren veritabanı, reaktif dönüş türlerini destekler: Flowable ve Observable. Tablodaki herhangi bir kayıt değiştiğinde, Room otomatik olarak akışa yeni bir değer gönderir. Geliştirici, ViewModel'de Flowable'a abone olur ve her değişiklikte manuel sorgular olmadan güncel verileri alır.

// Reaktif sorgu ile 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 kompozisyonu
class UserViewModel(private val dao: UserDao) : ViewModel() {
    val users: Flowable<List<User>> = dao.getAllUsers()
        .subscribeOn(Schedulers.io())
}

MVVM + RxJava deseni, ViewModel'in View'a referansı olmaması temeline dayanır. ViewModel, reaktif kaynaklar (Flowable, Transformations aracılığıyla LiveData) yayınlar ve Activity veya Fragment bunlara abone olur. Bu, test edilebilirlik sağlar: ViewModel, RxJavaPlugins.setComputationScheduler aracılığıyla Scheduler'ları değiştirerek UI olmadan test edilir. ViewModel'deki CompositeDisposable, abonelik yaşam döngüsünü yönetir — onCleared()'da tüm abonelikler iptal edilir.

RxJava vs Kotlin Flow

Kotlin Flow, Kotlin'de yerel soğuk akış uygulamasıdır, coroutine'lere entegre edilmiştir ve Kotlin 1.3'te tanıtılmıştır. Flow, RxJava ile aynı sorunları çözer ancak temel farklılıklarla: yerleşik coroutine desteği (suspend işlevleri), coroutine cancellation ile iptal ve geri basınç sorunlarının olmaması — Flow, arabelleğe alma yerine suspend kullanır. Flow, Kotlin standart kütüphanesinin bir parçasıdır ve ek bağımlılık gerektirmez.

RxJava, Java projeleri, Java 7-8'i destekleyen projeler ve mevcut RxJava kod tabanları için tercih edilen seçenek olmaya devam etmektedir. RxJava ekosistemi önemli ölçüde daha zengindir: Flow'daki yaklaşık 50'ye karşı 400'den fazla operatör, yerleşik CallAdapter aracılığıyla Retrofit ile entegrasyon, Flowable aracılığıyla geri basınç desteği ve Android için RxBinding, RxPermissions, RxLocation. Kotlin Flow hızla yetişiyor, ancak karmaşık akış birleştirme senaryolarında RxJava'nın esnekliği hala daha yüksektir.