RxJava: সার, উপাদান এবং রিঅ্যাকটিভ প্রোগ্রামিং

RxJava কী?

RxJava হল জাভা ভার্চুয়াল মেশিনের জন্য ReactiveX (Reactive Extensions) লাইব্রেরির একটি বাস্তবায়ন। RxJava-এর প্রথম সংস্করণ Netflix 2013 সালে সার্ভার-সাইড অ্যাপ্লিকেশনে অ্যাসিঙ্ক্রোনাস কল পরিচালনার জন্য প্রকাশ করেছিল। এর তৈরির সময়, Java-তে প্রধান বিকল্প ছিল Future এবং Callback — উভয় পদ্ধতিই callback-hell এবং জটিল থ্রেড ম্যানেজমেন্টের দিকে নিয়ে যেত। RxJava ফাংশনাল অপারেটরের চেইন সহ Observable-এর মাধ্যমে অ্যাসিঙ্ক্রোনাস অপারেশনের কম্পোজিশন চালু করেছিল।

RxJava-এর আর্কিটেকচার Reactive Streams স্পেসিফিকেশনের উপর ভিত্তি করে — নন-ব্লকিং ব্যাকপ্রেশার সহ অ্যাসিঙ্ক্রোনাস স্ট্রিম প্রসেসিংয়ের জন্য একটি মান। স্পেসিফিকেশন চারটি ইন্টারফেস সংজ্ঞায়িত করে: Publisher, Subscriber, Subscription এবং Processor। RxJava 2+ RxJava 1-এর বিপরীতে ব্যাকপ্রেশার কন্ট্রাক্ট মেনে Flowable টাইপের মাধ্যমে Reactive Streams সম্পূর্ণরূপে implement করে। RxJava 2-এ Observable ব্যাকপ্রেশার সমর্থন করে না — এটি অল্প সংখ্যক ইভেন্ট বা UI ইভেন্টের স্ট্রিমের জন্য।

JetBrains, 2025 জরিপ অনুসারে, RxJava Android ডেভেলপমেন্টের জন্য শীর্ষ 3 লাইব্রেরির মধ্যে একটি। প্রধান ব্যবহারের ক্ষেত্রগুলির মধ্যে রয়েছে: Retrofit-এর মাধ্যমে নেটওয়ার্ক রিকোয়েস্ট হ্যান্ডলিং (CallAdapter-এর মাধ্যমে RxJava-এর সাথে সংহত), Room-এর সাথে কাজ (রিঅ্যাকটিভ কোয়েরি Flowable বা Maybe রিটার্ন করে), RxBinding-এর মাধ্যমে অ্যানিমেশন এবং UI ইভেন্ট, এবং টেক্সট ইনপুটে ডিবাউন্স অনুসন্ধান। এই সমস্ত পরিস্থিতি একটি সাধারণ চেইন প্যাটার্ন শেয়ার করে: উৎস (Observable) → রূপান্তর (অপারেটর) → সাবস্ক্রিপশন (subscribe)।

RxJava সংস্করণ ইতিহাস

RxJava 1 (2013) Observable এবং অপারেটরের ভিত্তি স্থাপন করেছিল, কিন্তু ব্যাকপ্রেশার সমস্যায় ভুগছিল — দ্রুত স্ট্রিমে, ডেটা মেমোরিতে জমা হতো, OutOfMemoryError সৃষ্টি করে। RxJava 2 (2016) Observable (ব্যাকপ্রেশার ছাড়া) এবং Flowable (ব্যাকপ্রেশার সহ) আলাদা করে আর্কিটেকচার ঠিক করেছিল। RxJava 3 (2020) Java 8 Stream API সমর্থন, অতিরিক্ত অপারেটর এবং উন্নত সাবস্ক্রিপশন পারফরম্যান্স যোগ করেছে। বর্তমানে, RxJava 3 নতুন প্রজেক্টের জন্য প্রস্তাবিত সংস্করণ।

RxJava-তে রিঅ্যাকটিভ স্ট্রিমের প্রকার

RxJava পাঁচটি প্রধান ধরনের রিঅ্যাকটিভ উৎস প্রদান করে, প্রতিটি নির্দিষ্ট পরিস্থিতির জন্য ডিজাইন করা। Observable এবং Flowable একাধিক মান নির্গত করে, Single একটি মান বা ত্রুটি নির্গত করে, Completable ডেটা ছাড়া শুধুমাত্র সমাপ্তি নির্গত করে, এবং Maybe একটি মান, শূন্য বা ত্রুটি নির্গত করে। সঠিক প্রকার নির্বাচন কোডের পরিমাণ হ্রাস করে এবং চেইনকে স্ব-ডকুমেন্টিং করে তোলে।

Flowable বড় ডেটা স্ট্রিমের সাথে কাজ করার জন্য সবচেয়ে নমনীয় প্রকার। এটি ব্যাকপ্রেশার সমর্থন সহ Reactive Streams Publisher implement করে: কনজিউমার Subscription.request(n)-এর মাধ্যমে নির্দিষ্ট সংখ্যক উপাদান অনুরোধ করতে পারে। এটি প্রযোজক এবং কনজিউমারের গতি মেলে না গেলে বাফার ওভারফ্লো প্রতিরোধ করে। ব্যাকপ্রেশার গুরুত্বপূর্ণ না হলে Observable ব্যবহার করুন — অনুরোধ মেকানিজমের অনুপস্থিতির কারণে এর ওভারহেড কম।

Single HTTP রিকোয়েস্টের জন্য সর্বোত্তম পছন্দ। RxJava CallAdapter সহ Retrofit 2 প্রতিটি রিকোয়েস্টের জন্য Single<ResponseBody> রিটার্ন করে। Single ঠিক একটি onSuccess বা onError কলের গ্যারান্টি দেয়, যা HTTP রিকোয়েস্টের শব্দার্থের সাথে মেলে — একটি রেসপন্স বা একটি ত্রুটি। Completable রাইট অপারেশনের জন্য ব্যবহৃত হয় যা ডেটা রিটার্ন করে না: insert, update, delete। Maybe ক্যাশ চেক করার জন্য সুবিধাজনক — এটি মান রিটার্ন করতে পারে বা নাও করতে পারে।

// HTTP রিকোয়েস্টের জন্য Single ব্যবহারের উদাহরণ
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 একাধিক Observables থেকে উপাদানগুলিকে সূচক অনুসারে জোড়ায় জোড়ায় একত্রিত করে: প্রথম প্রথমটির সাথে, দ্বিতীয় দ্বিতীয়টির সাথে। combineLatest যেকোনো স্ট্রিম পরিবর্তিত হলে একটি নতুন মান নির্গত করে, সমস্ত স্ট্রিম থেকে সর্বশেষ মানগুলি একত্রিত করে। merge একাধিক Observables-কে একটিতে একত্রিত করে, ইভেন্ট আগমনের ক্রম সংরক্ষণ করে। concat ক্রমানুসারে প্রতিটি Observable-এ সাবস্ক্রাইব করে এবং পরবর্তীতে যাওয়ার আগে তার সমস্ত ইভেন্ট পাস করে।

সময় ম্যানেজমেন্ট-এর মধ্যে রয়েছে debounce (নির্গতের আগে স্ট্রিমে বিরতির জন্য অপেক্ষা), throttleFirst (প্রথম ইভেন্ট নির্গত, বাকিগুলো উইন্ডোর মধ্যে উপেক্ষা), timeout (ব্যবধানে কোনো ইভেন্ট না এলে ত্রুটি)। টেক্সট ইনপুটে ডিবাউন্স অনুসন্ধান সবচেয়ে সাধারণ পরিস্থিতি: searchObservable.debounce(300, MILLISECONDS).distinctUntilChanged() দ্রুত টাইপিংয়ের সময় অপ্রয়োজনীয় রিকোয়েস্ট প্রতিরোধ করে।

Schedulers এবং মাল্টিথ্রেডিং

Scheduler RxJava-তে থ্রেড পুলের উপর একটি অ্যাবস্ট্র্যাকশন। লাইব্রেরি পাঁচটি বিল্ট-ইন Scheduler প্রদান করে: I/O অপারেশনের জন্য Schedulers.io() (নেটওয়ার্ক, ফাইল), CPU-ইনটেনসিভ কাজের জন্য Schedulers.computation(), প্রতিবার নতুন থ্রেডের জন্য Schedulers.newThread(), একক-থ্রেড এক্সিকিউশনের জন্য Schedulers.single(), এবং বর্তমান থ্রেডে তাৎক্ষণিক এক্সিকিউশনের জন্য Schedulers.trampoline()।

subscribeOn এবং observeOn

subscribeOn নির্ধারণ করে কোন Scheduler উৎস Observable সম্পাদন করে। চেইনে একাধিক subscribeOn থাকলে, উৎসের কাছাকাছি থাকাটি অগ্রাধিকার পায়। observeOn ডাউনস্ট্রিম নির্দিষ্ট Scheduler-এ সুইচ করে — observeOn-এর প্রতিটি ব্যবহার পরবর্তী অপারেটরগুলির জন্য থ্রেড পরিবর্তন করে। একটি সাধারণ Android প্যাটার্ন: নেটওয়ার্ক অপারেশনের জন্য subscribeOn(Schedulers.io()), UI আপডেটের জন্য observeOn(AndroidSchedulers.mainThread())।

// কনটেক্সট সুইচিং সহ মাল্টিথ্রেডেড প্রক্রিয়াকরণ
Observable.fromCallable(() -> database.getItems())
    .subscribeOn(Schedulers.io())            // io-তে DB
    .map(items -> processItems(items))     // io-তে রূপান্তর
    .observeOn(Schedulers.computation())    // computation-এ সুইচ
    .map(processed -> compressImages(processed))
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(result -> ui.showResult(result))

AndroidSchedulers.mainThread() হল RxAndroid লাইব্রেরি থেকে একটি Scheduler যা Android প্রধান থ্রেডে কোড সম্পাদন করে। এটি রিঅ্যাকটিভ চেইনে যেকোনো UI আপডেটের জন্য বাধ্যতামূলক। লাইব্রেরি অভ্যন্তরীণভাবে Handler ব্যবহার করে এবং উচ্চ লোডের অধীনেও UI থ্রেডে এক্সিকিউশনের গ্যারান্টি দেয়। ব্যাকগ্রাউন্ড অপারেশনের জন্য, Schedulers.io() একটি সীমাহীন থ্রেড পুল সমর্থন করে এবং যেকোনো ব্লকিং অপারেশনের জন্য উপযুক্ত। Schedulers.computation() CPU কোরের সংখ্যার সমান একটি ফিক্সড পুল ব্যবহার করে।

Android-এ RxJava: ব্যবহারিক প্রয়োগ

RxJava Android-এ তিনটি প্রধান পরিস্থিতির জন্য ব্যবহৃত হয়: Room-এর জন্য রিঅ্যাকটিভ কোয়েরি, Retrofit-এর সাথে সংহতি, এবং RxBinding-এর মাধ্যমে রিঅ্যাকটিভ UI বাইন্ডিং। প্রতিটি পরিস্থিতির নিজস্ব প্রকারের সেট রয়েছে: Room পর্যবেক্ষণযোগ্য কোয়েরির জন্য Flowable রিটার্ন করে, Retrofit HTTP রিকোয়েস্টের জন্য Single রিটার্ন করে, RxBinding UI ইভেন্টের জন্য Observable রিটার্ন করে।

Room + RxJava

Room হল Google-এর একটি ডেটা পার্সিস্টেন্স লাইব্রেরি। Room 2.1 থেকে শুরু করে, ডাটাবেস রিঅ্যাকটিভ রিটার্ন টাইপ সমর্থন করে: Flowable এবং Observable। যখন টেবিলের যেকোনো রেকর্ড পরিবর্তিত হয়, Room স্বয়ংক্রিয়ভাবে স্ট্রিমে একটি নতুন মান পাঠায়। ডেভেলপার ViewModel-এ Flowable-এ সাবস্ক্রাইব করে এবং প্রতিটি পরিবর্তনে ম্যানুয়াল কোয়েরি ছাড়াই আপ-টু-ডেট ডেটা পায়।

// রিঅ্যাকটিভ কোয়েরি সহ 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, Transformations-এর মাধ্যমে LiveData) প্রকাশ করে, এবং Activity বা Fragment সেগুলিতে সাবস্ক্রাইব করে। এটি পরীক্ষাযোগ্যতা প্রদান করে: ViewModel UI ছাড়াই পরীক্ষা করা হয়, RxJavaPlugins.setComputationScheduler-এর মাধ্যমে Schedulers প্রতিস্থাপন করে। ViewModel-এ CompositeDisposable সাবস্ক্রিপশন লাইফসাইকেল পরিচালনা করে — onCleared()-এ, সমস্ত সাবস্ক্রিপশন বাতিল হয়ে যায়।

RxJava বনাম Kotlin Flow

Kotlin Flow হল Kotlin-এ কোল্ড স্ট্রিমের একটি নেটিভ বাস্তবায়ন, করুটিনে নির্মিত এবং Kotlin 1.3-এ প্রবর্তিত। Flow RxJava-এর মতো একই সমস্যা সমাধান করে কিন্তু মৌলিক পার্থক্যের সাথে: বিল্ট-ইন করুটিন সমর্থন (suspend ফাংশন), coroutine cancellation-এর মাধ্যমে বাতিলকরণ, এবং ব্যাকপ্রেশার সমস্যার অনুপস্থিতি — Flow বাফারিংয়ের পরিবর্তে suspend ব্যবহার করে। Flow Kotlin স্ট্যান্ডার্ড লাইব্রেরির অংশ, অতিরিক্ত নির্ভরতার প্রয়োজন নেই।

RxJava Java প্রজেক্ট, Java 7-8 সমর্থনকারী প্রজেক্ট এবং বিদ্যমান RxJava কোডবেসের জন্য পছন্দের বিকল্প রয়ে গেছে। RxJava ইকোসিস্টেম উল্লেখযোগ্যভাবে সমৃদ্ধ: Flow-এ প্রায় 50-এর তুলনায় 400-এর বেশি অপারেটর, বিল্ট-ইন CallAdapter-এর মাধ্যমে Retrofit-এর সাথে সংহতি, Flowable-এর মাধ্যমে ব্যাকপ্রেশার সমর্থন, এবং Android-এর জন্য RxBinding, RxPermissions, RxLocation। Kotlin Flow দ্রুত ধরা দিচ্ছে, কিন্তু জটিল স্ট্রিম একত্রীকরণ পরিস্থিতিতে RxJava-এর নমনীয়তা এখনও বেশি।