RxJava: βασικά, ReactiveX και εργασία με ροές δεδομένων

Τι είναι το RxJava και το ReactiveX

RxJava — υλοποίηση Java της προδιαγραφής ReactiveX, μια βιβλιοθήκη για ασύγχρονο προγραμματισμό χρησιμοποιώντας παρατηρήσιμες ροές (Observable). Η RxJava 2 κυκλοφόρησε το 2016 με υποστήριξη για Reactive Streams (Flowable) και διαχωρισμό σε rx.Observable και io.reactivex.Observable. Η RxJava 3 (2019) — τρέχουσα κύρια έκδοση με αντίστροφη συμβατότητα με RxJava 2.

Η βασική ιδέα του RxJava — όλα είναι ροή: ροή δεδομένων, ροή συμβάντων, ροή καταστάσεων. Οποιαδήποτε ασύγχρονη λειτουργία μπορεί να αναπαρασταθεί ως Observable που εκπέμπει δεδομένα, σφάλμα ή σήμα ολοκλήρωσης. Ο Observer εγγράφεται στο Observable και λαμβάνει ειδοποιήσεις σε πραγματικό χρόνο.

Σύμφωνα με δεδομένα της Badoo (2024), πριν από τη μετάβαση στα coroutines, το 76% των εφαρμογών Android από το top-200 του Google Play χρησιμοποιούσαν RxJava για ασύγχρονες λειτουργίες. Τώρα το μερίδιο μειώνεται υπέρ των coroutines, αλλά το RxJava παραμένει στον κώδικα παραγωγής χιλιάδων εφαρμογών και θεωρείται ώριμη, δοκιμασμένη τεχνολογία. ReactiveX — μια cross-platform προδιαγραφή, υλοποιημένη επίσης για JavaScript (RxJS), .NET (Rx.NET), Swift (RxSwift) και άλλες γλώσσες.

Το μοτίβο Observer στο RxJava

Το ReactiveX επεκτείνει το κλασικό μοτίβο Observer με δύο μηχανισμούς: αλυσίδα τελεστών (operator chaining) και διαχείριση νημάτων (schedulers). Το Observable δεν αρχίζει να εκπέμπει δεδομένα μέχρι να εγγραφεί ο Observer (τεμπέλικη αξιολόγηση). Αυτό επιτρέπει τη δημιουργία ενός pipeline δεδομένων που ενεργοποιείται μόνο παρουσία συνδρομής.

Τύποι Observable: Observable, Flowable, Single, Maybe, Completable

Observable — βασικός τύπος που εκπέμπει 0..N στοιχεία με onError ή onComplete. Κατάλληλο για ροές δεδομένων απεριόριστου μήκους — για παράδειγμα, συμβάντα κλικ ή ενημερώσεις γεωτοποθεσίας. Το Observable δεν υποστηρίζει backpressure.

Flowable — η έκδοση Reactive Streams του Observable με υποστήριξη backpressure. Χρησιμοποιείται όταν η πηγή δεδομένων μπορεί να παράγει στοιχεία ταχύτερα από όσο μπορεί να επεξεργαστεί ο Observer. Το Flowable υποστηρίζει τις στρατηγικές BACKPRESSURE_BUFFER, DROP, LATEST και ERROR.

Single, Maybe και Completable

Single εκπέμπει ακριβώς ένα στοιχείο ή σφάλμα — ιδανικό για αιτήματα δικτύου. Maybe — 0 ή 1 στοιχείο, κατάλληλο για cache όπου τα δεδομένα μπορεί να απουσιάζουν. Completable — μόνο onComplete ή onError, χωρίς δεδομένα, βολικό για λειτουργίες εγγραφής ή διαγραφής. Αυτοί οι τύποι απλοποιούν το API, περιορίζοντας τη σύμβαση σε μια συγκεκριμένη περίπτωση. Το Retrofit (δημοφιλής HTTP πελάτης για Android) υποστηρίζει και τους πέντε τύπους RxJava άμεσα, επιτρέποντας την επιλογή του καταλληλότερου τύπου επιστροφής για κάθε endpoint χωρίς περιττό περιτύλιγμα.

Τελεστές RxJava: μετασχηματισμός και φιλτράρισμα ροών

Οι τελεστές είναι συναρτήσεις που μετατρέπουν ένα Observable σε άλλο. Η αλυσίδα τελεστών (operator chain) περιγράφει το pipeline δεδομένων: κάθε τελεστής λαμβάνει τη ροή από τον προηγούμενο, τη μετασχηματίζει και τη μεταβιβάζει στον επόμενο. Το RxJava περιέχει πάνω από 200 τελεστές χωρισμένους σε κατηγορίες.

• map — μετασχηματίζει κάθε στοιχείο (Integer → String)
• flatMap — μετατρέπει το στοιχείο σε Observable και συνδυάζει όλα σε μία ροή
• filter — επιτρέπει τη διέλευση στοιχείων βάσει συνθήκης
• zip — συνδυάζει στοιχεία N Observable βάσει δείκτη
• merge — συνδυάζει πολλά Observable σε ένα, διατηρώντας τη χρονική σειρά
• debounce — επιτρέπει τη διέλευση στοιχείων αν το μεταξύ τους διάστημα είναι μικρότερο από το καθορισμένο

flatMap — ένας από τους πιο ισχυρούς τελεστές RxJava. Επιτρέπει την εκτέλεση ενός ασύγχρονου αιτήματος για κάθε στοιχείο και τη συλλογή των αποτελεσμάτων σε μια κοινή ροή. Για παράδειγμα, το flatMap χρησιμοποιείται για τη φόρτωση λεπτομερειών βάσει λίστας ID: κάθε ID → αίτημα δικτύου → συνδυασμός αποτελεσμάτων. Σε αντίθεση με το map, το οποίο απλά μετασχηματίζει ένα στοιχείο, το flatMap μπορεί να εκπέμψει πολλαπλά στοιχεία ή να μεταβεί σε άλλο Observable, καθιστώντας το τη βάση για τη δημιουργία ασύγχρονων pipelines.

Διαχείριση σφαλμάτων μέσω τελεστών

onErrorResumeNext — σε περίπτωση σφάλματος μεταβαίνει σε εφεδρικό Observable. retry — επαναλαμβάνει τη συνδρομή σε σφάλμα N φορές. onErrorReturn — επιστρέφει προεπιλεγμένη τιμή αντί για σφάλμα. doOnError — εκτελεί παρενέργεια σε σφάλμα χωρίς να αλλάζει τη ροή (καταγραφή ή αναλυτικά). Ο συνδυασμός αυτών των τελεστών επιτρέπει τη δημιουργία αξιόπιστων pipelines με σαφή στρατηγική διαχείρισης αποτυχιών χωρίς χειροκίνητο try/catch.

Schedulers: διαχείριση νημάτων στο RxJava

Οι Schedulers καθορίζουν σε ποιο νήμα εκτελούνται το Observable και ο Observer. Το subscribeOn ορίζει το νήμα για την πηγή, το observeOn — το νήμα για τον Observer και τους επόμενους τελεστές. Αυτός ο διαχωρισμός — το βασικό πλεονέκτημα του RxJava: πηγή στο νήμα IO, επεξεργασία στο computation, UI — στο κύριο νήμα.

Κύριοι Schedulers: Schedulers.io() — για λειτουργίες I/O (δίκτυο, δίσκος), απεριόριστος pool. Schedulers.computation() — για υπολογισμούς, σταθερός pool βάσει αριθμού πυρήνων. Schedulers.newThread() — νέο νήμα για κάθε εργασία. AndroidSchedulers.mainThread() — κύριο νήμα Android (RxAndroid). Υπάρχει επίσης Schedulers.trampoline() για εκτέλεση εργασιών στο τρέχον νήμα με ουρά FIFO, χρήσιμο για δοκιμές.

Σύμφωνα με δεδομένα της Google (2025), η σωστή χρήση των Schedulers είναι το πιο δύσκολο στο RxJava για αρχάριους. Το τυπικό λάθος — κλήση του subscribeOn μετά το observeOn, το οποίο δεν επηρεάζει την πηγή. Το subscribeOn πρέπει να είναι πρώτο στην αλυσίδα για την πηγή, το observeOn — πριν από τη συνδρομή UI. Κανόνας: το subscribeOn επηρεάζει μόνο το upstream (πηγή), το observeOn αλλάζει το downstream (συνδρομητής και όλοι οι τελεστές μετά από αυτόν).

Παραδείγματα κώδικα με RxJava στο Android

Ας εξετάσουμε τρία σενάρια: αίτημα δικτύου με Single, παράλληλα αιτήματα με zip και debounce για το πεδίο αναζήτησης με debounce.

Αίτημα δικτύου με Single

Single είναι ιδανικό για αιτήματα Retrofit: ένα αίτημα — μία απόκριση. Συνδρομή στο κύριο νήμα για ενημέρωση UI.

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

zip συνδυάζει τα αποτελέσματα δύο ανεξάρτητων Single σε ένα. Εκτελούνται παράλληλα, το αποτέλεσμα — μετά την ολοκλήρωση και των δύο.

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

Debounce για το πεδίο αναζήτησης

debounce αγνοεί τις γρήγορες αλλαγές κειμένου και στέλνει το αίτημα μόνο μετά από παύση 400 ms. distinctUntilChanged ακυρώνει το αίτημα αν το κείμενο δεν άλλαξε.

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: σύγκριση προσεγγίσεων

RxJava και Kotlin Coroutines λύνουν την ίδια εργασία — ασύγχρονο προγραμματισμό — αλλά με θεμελιωδώς διαφορετικές προσεγγίσεις. Το RxJava είναι χτισμένο στο μοτίβο Observer και είναι push-based: η πηγή στέλνει δεδομένα, ο Observer αντιδρά. Τα coroutines — pull-based: ο κώδικας λαμβάνει δεδομένα ακολουθιακά μέσω await.

• RxJava — αντιδραστικό, ροή δεδομένων, >200 τελεστές, push-based, απότομη καμπύλη μάθησης
• Coroutines — ακολουθιακό, suspend/await, ~40 συναρτήσεις, pull-based, απλή σύνταξη
• RxJava — ώριμο (2016), τεράστιο οικοσύστημα, αλλά απότομη καμπύλη μάθησης
• Coroutines — σύγχρονο (2018), προτιμώμενη επιλογή της Google για νέο κώδικα
• RxJava — backpressure από το κουτί μέσω Flowable, επεξεργασμένες στρατηγικές προσωρινής αποθήκευσης
• Coroutines — Flow με backpressure πρόσφατα, αλλά ενεργά αναπτύσσεται από τη JetBrains

Σύμφωνα με το Google I/O 2024, τα Kotlin Coroutines είναι η συνιστώμενη προσέγγιση για νέο ασύγχρονο κώδικα στο Android. Το RxJava παραμένει υποστηριζόμενο για υπάρχοντα έργα. Η Google παρέχει μεταβατικές βιβλιοθήκες (kotlinx-coroutines-rx3) για σταδιακή μετανάστευση. Το AndroidX (LiveData, Room, Paging 3) υποστηρίζει και τις δύο προσεγγίσεις, επιτρέποντας τη χρήση RxJava σε παλιά modules και coroutines σε νέα modules χωρίς συγκρούσεις εξαρτήσεων.

Στρατηγική μετανάστευσης από RxJava σε coroutines

Σταδιακή μετάβαση: κάθε νέο στοιχείο γράφεται με coroutines, ο παλιός κώδικας RxJava δεν αγγίζεται. RxJava → coroutines μέσω awaitSingle() ή awaitFirst(). Coroutines → RxJava μέσω future() ή asFlowable(). Η πλήρης μετανάστευση διαρκεί 6–18 μήνες για μεγάλα έργα.