RxJava: ουσία, συστατικά και αντιδραστικός προγραμματισμός

Τι είναι το RxJava;

RxJava είναι μια υλοποίηση της βιβλιοθήκης ReactiveX (Reactive Extensions) για την εικονική μηχανή Java. Η πρώτη έκδοση του RxJava κυκλοφόρησε από την εταιρεία Netflix το 2013 για τη διαχείριση ασύγχρονων κλήσεων σε εφαρμογές διακομιστή. Κατά τη δημιουργία του, η κύρια εναλλακτική σε Java ήταν τα Future και Callback — και οι δύο προσεγγίσεις οδηγούσαν σε callback-hell και πολύπλοκη διαχείριση νημάτων. Το RxJava πρότεινε τη σύνθεση ασύγχρονων λειτουργιών μέσω Observable με αλυσίδες λειτουργικών τελεστών.

Η αρχιτεκτονική του RxJava βασίζεται στην προδιαγραφή Reactive Streams — ένα πρότυπο για ασύγχρονη επεξεργασία ροών με μη αποκλειστικό backpressure. Η προδιαγραφή ορίζει τέσσερις διεπαφές: Publisher, Subscriber, Subscription και Processor. Το RxJava 2+ υλοποιεί πλήρως τα Reactive Streams μέσω του τύπου Flowable, τηρώντας τα συμβόλαια backpressure σε αντίθεση με το RxJava 1. Το Observable στο RxJava 2 δεν υποστηρίζει backpressure — προορίζεται για ροές με μικρό αριθμό συμβάντων ή συμβάντα UI.

Σύμφωνα με την έρευνα της JetBrains, 2025, το RxJava βρίσκεται στις top-3 βιβλιοθήκες για ανάπτυξη Android. Τα κύρια σενάρια χρήσης: επεξεργασία αιτημάτων δικτύου μέσω Retrofit (ενσωματωμένο με RxJava μέσω CallAdapter), εργασία με Room (αντιδραστικά ερωτήματα επιστρέφουν Flowable ή Maybe), κινούμενα σχέδια και συμβάντα UI μέσω RxBinding και αναζήτηση debounce κατά την εισαγωγή κειμένου. Όλα αυτά τα σενάρια ενώνονται από μια αλυσίδα ίδιου τύπου: πηγή (Observable) → μετασχηματισμός (τελεστές) → εγγραφή (subscribe).

Ιστορία εκδόσεων RxJava

RxJava 1 (2013) έθεσε την έννοια του Observable και των τελεστών, αλλά υπέφερε από προβλήματα με το backpressure — σε γρήγορες ροές, τα δεδομένα συσσωρεύονταν στη μνήμη, προκαλώντας OutOfMemoryError. RxJava 2 (2016) διόρθωσε την αρχιτεκτονική, διαχωρίζοντας το Observable (χωρίς backpressure) και το Flowable (με backpressure). RxJava 3 (2020) πρόσθεσε υποστήριξη για Java 8 Stream API, επιπλέον τελεστές και βελτιωμένη απόδοση κατά την εγγραφή. Αυτή τη στιγμή, το RxJava 3 είναι η συνιστώμενη έκδοση για νέα έργα.

Τύποι αντιδραστικών ροών στο RxJava

RxJava παρέχει πέντε κύριους τύπους αντιδραστικών πηγών, κάθε ένας προσανατολισμένος σε ένα συγκεκριμένο σενάριο. Τα Observable και Flowable εκπέμπουν πολλαπλές τιμές, το Single — μία τιμή ή σφάλμα, το Completable — μόνο το γεγονός ολοκλήρωσης χωρίς δεδομένα, το Maybe — μία τιμή, μηδέν ή σφάλμα. Η επιλογή του σωστού τύπου μειώνει την ποσότητα κώδικα και καθιστά την αλυσίδα αυτο-τεκμηριωμένη.

Flowable είναι ο πιο ευέλικτος τύπος για εργασία με μεγάλες ροές δεδομένων. Υλοποιεί το Reactive Streams Publisher με υποστήριξη backpressure: ο consumer μπορεί να ζητήσει συγκεκριμένο αριθμό στοιχείων μέσω Subscription.request(n). Αυτό αποτρέπει την υπερχείλιση buffer όταν οι ταχύτητες producer και consumer δεν ταιριάζουν. Εάν το backpressure δεν είναι κρίσιμο — χρησιμοποιήστε Observable, έχει μικρότερο overhead λόγω απουσίας μηχανισμού request.

Single είναι η βέλτιστη επιλογή για αιτήματα HTTP. Το Retrofit 2 με RxJava CallAdapter επιστρέφει Single<ResponseBody> για κάθε αίτημα. Το Single εγγυάται ακριβώς μία κλήση onSuccess ή onError, που αντιστοιχεί στη σημασιολογία ενός αιτήματος HTTP — μία απόκριση ή ένα σφάλμα. Completable χρησιμοποιείται για λειτουργίες εγγραφής που δεν επιστρέφουν δεδομένα: insert, update, delete. Maybe είναι βολικό κατά τον έλεγχο της cache — μπορεί να επιστρέψει τιμή, μπορεί και όχι.

// Παράδειγμα χρήσης Single για αίτημα HTTP
interface ApiService {
    @GET("users/{id}")
    fun getUser(@Path("id") userId: Int): Single<User>
}

// Εγγραφή με επεξεργασία στο κύριο νήμα
apiService.getUser(42)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe({ user ->
        textView.text = user.name
    }, { error ->
        Log.e("API", "Error: ${error.message}")
    })
    .addTo(compositeDisposable)

Τελεστές μετασχηματισμού και διαχείρισης ροών

Τελεστές RxJava είναι συναρτήσεις ανώτερης τάξης που λαμβάνουν μια αντιδραστική πηγή και επιστρέφουν μια άλλη, μετασχηματίζοντας τη ροή δεδομένων. Το RxJava 3 περιέχει περισσότερους από 400 τελεστές χωρισμένους σε κατηγορίες: μετασχηματισμός, φιλτράρισμα, συνδυασμός, διαχείριση σφαλμάτων και διαχείριση χρόνου. Κάθε τελεστής είναι τεμπέλης — η αλυσίδα χτίζεται κατά τη δήλωση, εκτελείται κατά την εγγραφή.

Τελεστές μετασχηματισμού

map είναι ο βασικός τελεστής που μετασχηματίζει κάθε τιμή μέσω μιας συνάρτησης. Το flatMap λαμβάνει μια συνάρτηση που επιστρέφει Observable για κάθε στοιχείο και αναπτύσσει το αποτέλεσμα σε μια ενιαία ροή. Το switchMap είναι παρόμοιο με το flatMap, αλλά κατά τη λήψη ενός νέου στοιχείου αποεγγράφεται από το προηγούμενο Observable. Το concatMap διατηρεί τη σειρά των στοιχείων — σε αντίθεση με το flatMap, εγγράφεται διαδοχικά σε κάθε ένθετο Observable.

// Ανάλυση JSON με μετασχηματισμό και φιλτράρισμα
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("Σφάλμα", it.message) })

Συνδυασμός ροών είναι ο τομέας όπου το RxJava είναι ιδιαίτερα ισχυρό. Το zip συνδυάζει στοιχεία από πολλαπλά Observable σε ζεύγη ανά δείκτη: πρώτο με πρώτο, δεύτερο με δεύτερο. Το combineLatest εκπέμπει μια νέα τιμή όταν αλλάξει οποιαδήποτε από τις ροές, συνδυάζοντας τις τελευταίες τιμές όλων των ροών. Το merge συνδυάζει πολλαπλά Observable σε ένα, διατηρώντας τη σειρά άφιξης των συμβάντων. Το concat εγγράφεται διαδοχικά σε κάθε Observable και μεταδίδει όλα τα συμβάντα του πριν προχωρήσει στο επόμενο.

Διαχείριση χρόνου περιλαμβάνει debounce (αναμονή για παύση στη ροή πριν από την αποστολή), throttleFirst (πέρασμα πρώτου συμβάντος, αγνόηση των υπολοίπων στο παράθυρο), timeout (σφάλμα εάν το συμβάν δεν έφτασε στο διάστημα). Η αναζήτηση debounce κατά την εισαγωγή κειμένου είναι το πιο κοινό σενάριο: searchObservable.debounce(300, MILLISECONDS).distinctUntilChanged() αποτρέπει περιττά αιτήματα κατά τη γρήγορη πληκτρολόγηση.

Schedulers και πολυνηματικότητα

Scheduler στο RxJava είναι μια αφαίρεση πάνω από μια ομάδα νημάτων. Η βιβλιοθήκη παρέχει πέντε ενσωματωμένους Scheduler: Schedulers.io() για λειτουργίες I/O (δίκτυο, αρχεία), Schedulers.computation() για εργασίες έντασης CPU, Schedulers.newThread() για κάθε νέο νήμα, Schedulers.single() για μονονηματική εκτέλεση και Schedulers.trampoline() για άμεση εκτέλεση στο τρέχον νήμα.

subscribeOn και observeOn

subscribeOn καθορίζει σε ποιο Scheduler εκτελείται η πηγή Observable. Εάν υπάρχουν πολλαπλά subscribeOn στην αλυσίδα — προτεραιότητα έχει το πλησιέστερο στην πηγή. observeOn μεταφέρει το downstream στο καθορισμένο Scheduler — κάθε χρήση του observeOn αλλάζει το νήμα για τους επόμενους τελεστές. Ένα τυπικό μοτίβο Android: subscribeOn(Schedulers.io()) για εργασία με δίκτυο, observeOn(AndroidSchedulers.mainThread()) για ενημέρωση UI.

// Πολυνηματική επεξεργασία με εναλλαγή περιβάλλοντος
Observable.fromCallable(() -> database.getItems())
    .subscribeOn(Schedulers.io())            // ΒΔ στο io
    .map(items -> processItems(items))     // μετασχηματισμός στο io
    .observeOn(Schedulers.computation())    // μεταφέρουμε στο computation
    .map(processed -> compressImages(processed))
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(result -> ui.showResult(result))

AndroidSchedulers.mainThread() είναι ένας Scheduler από τη βιβλιοθήκη RxAndroid που εκτελεί κώδικα στο κύριο νήμα του Android. Είναι υποχρεωτικός για οποιαδήποτε ενημέρωση UI στην αντιδραστική αλυσίδα. Η βιβλιοθήκη χρησιμοποιεί εσωτερικά Handler και εγγυάται εκτέλεση στο νήμα UI ακόμη και υπό υψηλό φορτίο. Για λειτουργίες παρασκηνίου, το Schedulers.io() υποστηρίζει απεριόριστη ομάδα νημάτων και είναι κατάλληλο για οποιεσδήποτε λειτουργίες αποκλεισμού. Το Schedulers.computation() χρησιμοποιεί σταθερή ομάδα, ίση με τον αριθμό των πυρήνων του επεξεργαστή.

RxJava στο Android: πρακτική εφαρμογή

RxJava στο Android χρησιμοποιείται για τρία κύρια σενάρια: αντιδραστικά ερωτήματα στο Room, ενσωμάτωση με Retrofit και αντιδραστική σύνδεση UI μέσω RxBinding. Κάθε σενάριο έχει το δικό του σύνολο τύπων: το Room επιστρέφει Flowable για παρατηρήσιμα ερωτήματα, το Retrofit — Single για αιτήματα HTTP, το RxBinding — Observable για συμβάντα UI.

Room + RxJava

Room είναι μια βιβλιοθήκη μονιμότητας δεδομένων από την Google. Από το Room 2.1, η βάση δεδομένων υποστηρίζει αντιδραστικούς τύπους επιστροφής: Flowable και Observable. Όταν αλλάξει οποιαδήποτε εγγραφή στον πίνακα, το Room στέλνει αυτόματα μια νέα τιμή στη ροή. Ο προγραμματιστής εγγράφεται στο Flowable στο ViewModel και λαμβάνει ενημερωμένα δεδομένα χωρίς χειροκίνητα ερωτήματα σε κάθε αλλαγή.

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

Το μοτίβο MVVM + RxJava βασίζεται στο γεγονός ότι το ViewModel δεν έχει αναφορές στο View. Το ViewModel δημοσιεύει αντιδραστικές πηγές (Flowable, LiveData μέσω Transformations) και το Activity ή Fragment εγγράφεται σε αυτές. Αυτό παρέχει δυνατότητα δοκιμής: το ViewModel δοκιμάζεται χωρίς UI, αντικαθιστώντας τον Scheduler μέσω RxJavaPlugins.setComputationScheduler. Το CompositeDisposable στο ViewModel διαχειρίζεται τον κύκλο ζωής των εγγραφών — κατά το onCleared(), όλες οι εγγραφές ακυρώνονται.

RxJava εναντίον Kotlin Flow

Kotlin Flow είναι μια εγγενής υλοποίηση ψυχρών ροών στο Kotlin, ενσωματωμένη στα coroutines και εισαχθείσα στο Kotlin 1.3. Το Flow επιλύει τις ίδιες εργασίες με το RxJava, αλλά με θεμελιώδεις διαφορές: ενσωματωμένη υποστήριξη για coroutines (συναρτήσεις suspend), ακύρωση μέσω coroutine cancellation και απουσία προβλημάτων με backpressure — το Flow χρησιμοποιεί suspend αντί για buffering. Το Flow αποτελεί μέρος της τυπικής βιβλιοθήκης Kotlin, χωρίς να απαιτεί πρόσθετες εξαρτήσεις.

RxJava παραμένει η προτιμώμενη επιλογή για έργα σε Java, έργα με υποστήριξη Java 7-8 και υπάρχουσες βάσεις κώδικα σε RxJava. Το οικοσύστημα RxJava είναι σημαντικά πλουσιότερο: >400 τελεστές έναντι ~50 στο Flow, ενσωμάτωση με Retrofit μέσω ενσωματωμένου CallAdapter, υποστήριξη backpressure μέσω Flowable και ύπαρξη RxBinding, RxPermissions, RxLocation για Android. Το Kotlin Flow προλαβαίνει γρήγορα, αλλά η ευελιξία του RxJava σε πολύπλοκα σενάρια συνδυασμού ροών είναι ακόμη υψηλότερη.