RxJava: essens, komponenter och reaktiv programmering

Vad är RxJava?

RxJava är en implementation av ReactiveX-biblioteket (Reactive Extensions) för Java Virtual Machine. Den första versionen av RxJava släpptes av Netflix 2013 för att hantera asynkrona anrop i serverapplikationer. Vid skapandet var de främsta alternativen i Java Future och Callback — båda metoderna ledde till callback-hell och komplex trådhantering. RxJava föreslog komposition av asynkrona operationer via Observable med kedjor av funktionella operatorer.

Arkitekturen för RxJava är baserad på Reactive Streams-specifikationen — en standard för asynkron strömbehandling med icke-blockerande backpressure. Specifikationen definierar fyra gränssnitt: Publisher, Subscriber, Subscription och Processor. RxJava 2+ implementerar Reactive Streams fullt ut via typen Flowable, med respekt för backpressure-kontrakt till skillnad från RxJava 1. Observable i RxJava 2 stöder inte backpressure — det är avsett för strömmar med ett litet antal händelser eller UI-händelser.

Enligt undersökningen från JetBrains, 2025 är RxJava bland de tre främsta biblioteken för Android-utveckling. De viktigaste användningsscenarierna: bearbetning av nätverksförfrågningar via Retrofit (integrerat med RxJava via CallAdapter), arbete med Room (reaktiva frågor returnerar Flowable eller Maybe), animationer och UI-händelser via RxBinding och debounce-sökning vid textinmatning. Alla dessa scenarier förenas av en kedja av samma typ: källa (Observable) → transformation (operatorer) → prenumeration (subscribe).

Historik över RxJava-versioner

RxJava 1 (2013) lade grunden för konceptet Observable och operatorer, men led av problem med backpressure — i snabba strömmar ackumulerades data i minnet och orsakade OutOfMemoryError. RxJava 2 (2016) reparerade arkitekturen genom att separera Observable (utan backpressure) och Flowable (med backpressure). RxJava 3 (2020) lade till stöd för Java 8 Stream API, ytterligare operatorer och förbättrad prestanda vid prenumeration. För närvarande är RxJava 3 den rekommenderade versionen för nya projekt.

Typer av reaktiva strömmar i RxJava

RxJava tillhandahåller fem huvudtyper av reaktiva källor, var och en inriktad på ett specifikt scenario. Observable och Flowable sänder ut flera värden, Single — ett värde eller fel, Completable — endast slutförandet utan data, Maybe — ett värde, noll eller fel. Att välja rätt typ minskar mängden kod och gör kedjan självdokumenterande.

Flowable är den mest flexibla typen för arbete med stora dataströmmar. Det implementerar Reactive Streams Publisher med stöd för backpressure: konsumenten kan begära ett specifikt antal element via Subscription.request(n). Detta förhindrar buffertspill vid obalans i hastighet mellan producent och konsument. Om backpressure inte är kritisk — använd Observable, som har mindre overhead på grund av avsaknaden av request-mekanismen.

Single är det optimala valet för HTTP-förfrågningar. Retrofit 2 med RxJava CallAdapter returnerar Single<ResponseBody> för varje förfrågan. Single garanterar exakt ett anrop av onSuccess eller onError, vilket motsvarar semantiken för en HTTP-förfrågan — ett svar eller ett fel. Completable används för skrivoperationer som inte returnerar data: insert, update, delete. Maybe är praktiskt vid kontroll av cache — kan returnera ett värde, kan också låta bli.

// Exempel på användning av Single för HTTP-förfrågan
interface ApiService {
    @GET("users/{id}")
    fun getUser(@Path("id") userId: Int): Single<User>
}

// Prenumeration med bearbetning på huvudtråden
apiService.getUser(42)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe({ user ->
        textView.text = user.name
    }, { error ->
        Log.e("API", "Error: ${error.message}")
    })
    .addTo(compositeDisposable)

Transformations- och strömhanteringsoperatorer

Operatorer RxJava är funktioner av högre ordning som tar emot en reaktiv källa och returnerar en annan, vilket transformerar dataströmmen. RxJava 3 innehåller över 400 operatorer uppdelade i kategorier: transformation, filtrering, kombination, felhantering och tidsstyrning. Varje operator är lat — kedjan byggs vid deklaration och utförs vid prenumeration.

Transformationsoperatorer

map är den grundläggande operatorn som transformerar varje värde genom en funktion. flatMap tar emot en funktion som returnerar ett Observable för varje element och vecklar ut resultatet i en enda ström. switchMap liknar flatMap, men vid mottagning av ett nytt element avregistrerar det sig från det föregående Observable. concatMap bevarar ordningen på elementen — till skillnad från flatMap prenumererar den sekventiellt på varje kapslat Observable.

// JSON-tolkning med transformation och filtrering
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("Fel", it.message) })

Kombination av strömmar är området där RxJava är särskilt starkt. zip kombinerar element från flera Observable i par efter index: första med första, andra med andra. combineLatest sänder ut ett nytt värde när någon av strömmarna ändras och kombinerar de senaste värdena från alla strömmar. merge kombinerar flera Observable till en och bevarar ordningen för händelsernas ankomst. concat prenumererar sekventiellt på varje Observable och vidarebefordrar alla dess händelser innan den går vidare till nästa.

Tidsstyrning inkluderar debounce (vänta på en paus i strömmen före sändning), throttleFirst (släpp igenom den första händelsen, ignorera resten i fönstret), timeout (fel om händelsen inte kom inom intervallet). Debounce-sökning vid textinmatning är det vanligaste scenariot: searchObservable.debounce(300, MILLISECONDS).distinctUntilChanged() förhindrar onödiga förfrågningar vid snabb skrivning.

Schedulers och flertrådning

Scheduler i RxJava är en abstraktion över en trådpool. Biblioteket tillhandahåller fem inbyggda Schedulers: Schedulers.io() för I/O-operationer (nätverk, filer), Schedulers.computation() för CPU-intensiva uppgifter, Schedulers.newThread() för varje ny tråd, Schedulers.single() för enkeltrådad exekvering och Schedulers.trampoline() för omedelbar exekvering i den aktuella tråden.

subscribeOn och observeOn

subscribeOn bestämmer på vilken Scheduler Observable-källan exekveras. Om det finns flera subscribeOn i kedjan — har den närmast källan prioritet. observeOn växlar downstream till den angivna Schedulern — varje användning av observeOn ändrar tråden för efterföljande operatorer. Ett typiskt Android-mönster: subscribeOn(Schedulers.io()) för arbete med nätverk, observeOn(AndroidSchedulers.mainThread()) för UI-uppdatering.

// Flertrådad bearbetning med kontextväxling
Observable.fromCallable(() -> database.getItems())
    .subscribeOn(Schedulers.io())            // DB på io
    .map(items -> processItems(items))     // transformation på io
    .observeOn(Schedulers.computation())    // växlar till computation
    .map(processed -> compressImages(processed))
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(result -> ui.showResult(result))

AndroidSchedulers.mainThread() är en Scheduler från RxAndroid-biblioteket som exekverar kod på Android-huvudtråden. Den är obligatorisk för alla UI-uppdateringar i den reaktiva kedjan. Biblioteket använder internt Handler och garanterar exekvering i UI-tråden även under hög belastning. För bakgrundsoperationer stöder Schedulers.io() en obegränsad trådpool och är lämplig för alla blockerande operationer. Schedulers.computation() använder en fast pool som är lika med antalet processorkärnor.

RxJava i Android: praktisk tillämpning

RxJava i Android används för tre huvudscenarier: reaktiva frågor till Room, integration med Retrofit och reaktiv bindning av UI via RxBinding. Varje scenario har sin egen uppsättning typer: Room returnerar Flowable för observerbara frågor, Retrofit — Single för HTTP-förfrågningar, RxBinding — Observable för UI-händelser.

Room + RxJava

Room är ett datapersistensbibliotek från Google. Från och med Room 2.1 stöder databasen reaktiva returtyper: Flowable och Observable. När en post i tabellen ändras skickar Room automatiskt ett nytt värde till strömmen. Utvecklaren prenumererar på Flowable i ViewModel och får aktuell data utan manuella frågor vid varje ändring.

// Room DAO med reaktiv fråga
@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 — komposition Room + Network
class UserViewModel(private val dao: UserDao) : ViewModel() {
    val users: Flowable<List<User>> = dao.getAllUsers()
        .subscribeOn(Schedulers.io())
}

Mönstret MVVM + RxJava bygger på att ViewModel inte har referenser till View. ViewModel publicerar reaktiva källor (Flowable, LiveData via Transformations) och Activity eller Fragment prenumererar på dem. Detta ger testbarhet: ViewModel testas utan UI, genom att ersätta Scheduler via RxJavaPlugins.setComputationScheduler. CompositeDisposable i ViewModel hanterar prenumerationernas livscykel — vid onCleared() avbryts alla prenumerationer.

RxJava vs Kotlin Flow

Kotlin Flow är en nativ implementation av kalla strömmar i Kotlin, inbyggd i korutiner och introducerad i Kotlin 1.3. Flow löser samma uppgifter som RxJava men med grundläggande skillnader: inbyggt stöd för korutiner (suspend-funktioner), avbrytning via coroutine cancellation och frånvaro av problem med backpressure — Flow använder suspend istället för buffring. Flow är en del av Kotlin-standardbiblioteket och kräver inga ytterligare beroenden.

RxJava förblir det föredragna valet för projekt i Java, projekt med stöd för Java 7-8 och befintliga kodbaser på RxJava. RxJava-ekosystemet är betydligt rikare: >400 operatorer jämfört med ~50 i Flow, integration med Retrofit via inbyggd CallAdapter, stöd för backpressure via Flowable och tillgången till RxBinding, RxPermissions, RxLocation för Android. Kotlin Flow kommer snabbt ikapp, men RxJavas flexibilitet i komplexa strömkombinationsscenarier är fortfarande högre.