Reaktive Programmierung in Kotlin: RxKotlin und Koroutinen…

Reaktive Programmierung in Kotlin: Von RxKotlin zu Koroutinen

Reaktive Programmierung ist ein Paradigma, das auf die asynchrone Verarbeitung von Datenströmen und die Verbreitung von Änderungen ausgerichtet ist. Im Kotlin-Ökosystem wird dieses Konzept auf zwei Hauptarten umgesetzt: durch die Bibliothek RxKotlin (reaktive Erweiterungen für Kotlin auf Basis von RxJava) und durch den integrierten Mechanismus der Koroutinen mit Flow. In diesem Artikel werden wir beide Ansätze, ihre Besonderheiten und Anwendungsbeispiele im Detail besprechen.

Was ist reaktive Programmierung?

Die grundlegende Idee der reaktiven Programmierung ist es, mit Daten als kontinuierlichen Strömen (Streams) zu arbeiten. Anstatt Daten anzufordern und auf eine Antwort zu warten, abonniert man einen Strom und reagiert auf jedes neue Ereignis. Dies ist besonders nützlich für UI-Anwendungen, die Verarbeitung von Benutzereingaben, Netzwerkanfragen und die Arbeit mit Datenbanken.

Die Bibliothek RxKotlin

RxKotlin ist ein Wrapper um RxJava, der praktische Erweiterungen für Kotlin hinzufügt, wie Lambdas, Erweiterungsfunktionen und Unterstützung für Null-Sicherheit. Die wichtigsten Typen:

Observable<T> — Ein Strom, der 0 bis N Elemente aussenden und mit einem Fehler oder erfolgreich enden kann.
Single<T> — Ein Strom, der genau ein Element oder einen Fehler aussendet.
Flowable<T> — Ein Strom mit Unterstützung für Backpressure (Steuerung der Verarbeitungsgeschwindigkeit).
Completable — Ein Strom, der nur abschließt (Erfolg/Fehler) ohne Daten.

Beispiel RxKotlin

import io.reactivex.rxjava3.core.Observableimport io.reactivex.rxjava3.schedulers.Schedulers

fun main() {    val observable = Observable.create<String> { emitter ->        emitter.onNext("Hallo")        emitter.onNext("von")        emitter.onNext("RxKotlin")        emitter.onComplete()    }

    observable        .subscribeOn(Schedulers.io())        .observeOn(Schedulers.single())        .subscribe(            { value -> println("Erhalten: $value") },            { error -> println("Fehler: ${error.message}") },            { println("Strom abgeschlossen") }        )

    Thread.sleep(1000) // Warten auf Abschluss der asynchronen Operationen}

In diesem Beispiel wird ein Observable erstellt, das drei Zeichenketten aussendet und abschließt. Die Operatoren subscribeOn und observeOn steuern die Ausführungsthreads.

Koroutinen und Flow in Kotlin

Seit Kotlin 1.3 sind Koroutinen Teil der Sprache, und Flow (eingeführt in 1.3.70) ist ein reaktiver Strom, der vollständig in Koroutinen integriert ist. Vorteile gegenüber RxKotlin:

Weniger Overhead (leichtgewichtige Koroutinen vs. Java-Threads).
Natürliche Integration mit suspend-Funktionen.
Unterstützung für Structured Concurrency (automatische Abbrechung bei Scope-Ende).
Sauberere Syntax ohne Callbacks.

Beispiel Flow

import kotlinx.coroutines.*import kotlinx.coroutines.flow.*

fun main() = runBlocking {    val flow = flow {        for (i in 1..5) {            delay(100) // Simulation asynchroner Arbeit            emit(i)    // Element aussenden        }    }

    flow        .map { it * it } // Transformation        .catch { e -> println("Fehler: $e") }        .collect { value -> println("Quadrat: $value") }}

Flow ist faul: Daten werden erst bei Aufruf von collect generiert. Die Operatoren map, filter, flatMapConcat funktionieren genauso wie in Rx.