BackpressureStrategy (BUFFER, DROP, LATEST, MISSING)

RxJava-তে Backpressure একটি গুরুত্বপূর্ণ বিষয়, বিশেষত যখন ডেটা প্রডিউসার (Observable) দ্রুত ডেটা তৈরি করে এবং কনজিউমার (Observer) সেগুলো প্রক্রিয়া করতে পারে না। এই সমস্যাগুলো সমাধানের জন্য BackpressureStrategy ব্যবহার করা হয়।

RxJava এর Flowable ক্লাস ব্যাকপ্রেশার হ্যান্ডলিংয়ের জন্য ব্যবহৃত হয়, যেখানে BackpressureStrategy বিভিন্ন পদ্ধতিতে ডেটা ম্যানেজ করে।

BackpressureStrategy-এর প্রকারভেদ

RxJava-তে নিম্নলিখিত BackpressureStrategy উপলব্ধ রয়েছে:

1. BUFFER
2. DROP
3. LATEST
4. MISSING

1. BackpressureStrategy.BUFFER

BUFFER স্ট্র্যাটেজিতে সমস্ত ডেটা একটি বাফারে সংরক্ষণ করা হয়।

• এটি ডেটা মিস না করার জন্য কার্যকর, কিন্তু যদি প্রডিউসার অনেক দ্রুত কাজ করে, তাহলে OutOfMemoryError (OOM) হতে পারে।

উদাহরণ:

import io.reactivex.rxjava3.core.Flowable;
import io.reactivex.rxjava3.schedulers.Schedulers;

public class BufferStrategyExample {
    public static void main(String[] args) {
        Flowable<Integer> flowable = Flowable.create(emitter -> {
            for (int i = 0; i < 1000; i++) {
                emitter.onNext(i);
            }
            emitter.onComplete();
        }, BackpressureStrategy.BUFFER);

        flowable
            .observeOn(Schedulers.io())
            .subscribe(
                data -> {
                    Thread.sleep(10); // Simulate slow consumer
                    System.out.println("Received: " + data);
                },
                Throwable::printStackTrace
            );
    }
}

2. BackpressureStrategy.DROP

DROP স্ট্র্যাটেজিতে অতিরিক্ত ডেটা ফেলে দেওয়া হয়।

• শুধুমাত্র কনজিউমার যতটুকু প্রক্রিয়া করতে পারে ততটুকুই ডেটা গ্রহণ করে।
• দ্রুত ডেটা স্ট্রিমের ক্ষেত্রে কার্যকর।

উদাহরণ:

import io.reactivex.rxjava3.core.Flowable;
import io.reactivex.rxjava3.schedulers.Schedulers;

public class DropStrategyExample {
    public static void main(String[] args) {
        Flowable<Integer> flowable = Flowable.create(emitter -> {
            for (int i = 0; i < 1000; i++) {
                emitter.onNext(i);
            }
            emitter.onComplete();
        }, BackpressureStrategy.DROP);

        flowable
            .observeOn(Schedulers.io())
            .subscribe(
                data -> {
                    Thread.sleep(10); // Simulate slow consumer
                    System.out.println("Received: " + data);
                },
                Throwable::printStackTrace
            );
    }
}

3. BackpressureStrategy.LATEST

LATEST স্ট্র্যাটেজিতে শুধুমাত্র সর্বশেষ ডেটা ধরে রাখা হয়।

• পুরোনো ডেটা মিস হয়ে যায়।
• বাস্তব সময়ের ডেটার ক্ষেত্রে কার্যকর।

উদাহরণ:

import io.reactivex.rxjava3.core.Flowable;
import io.reactivex.rxjava3.schedulers.Schedulers;

public class LatestStrategyExample {
    public static void main(String[] args) {
        Flowable<Integer> flowable = Flowable.create(emitter -> {
            for (int i = 0; i < 1000; i++) {
                emitter.onNext(i);
            }
            emitter.onComplete();
        }, BackpressureStrategy.LATEST);

        flowable
            .observeOn(Schedulers.io())
            .subscribe(
                data -> {
                    Thread.sleep(10); // Simulate slow consumer
                    System.out.println("Received: " + data);
                },
                Throwable::printStackTrace
            );
    }
}

4. BackpressureStrategy.MISSING

MISSING স্ট্র্যাটেজি ডিফল্ট কোনো ব্যাকপ্রেশার স্ট্র্যাটেজি প্রয়োগ করে না।

• ব্যবহারকারীকে নিজে থেকে ব্যাকপ্রেশার হ্যান্ডলিংয়ের ব্যবস্থা করতে হবে।
• সাধারণত Flowable অপারেটর যেমন onBackpressureBuffer(), onBackpressureDrop() ইত্যাদির মাধ্যমে ম্যানেজ করা হয়।

উদাহরণ:

import io.reactivex.rxjava3.core.Flowable;
import io.reactivex.rxjava3.schedulers.Schedulers;

public class MissingStrategyExample {
    public static void main(String[] args) {
        Flowable<Integer> flowable = Flowable.create(emitter -> {
            for (int i = 0; i < 1000; i++) {
                emitter.onNext(i);
            }
            emitter.onComplete();
        }, BackpressureStrategy.MISSING)
        .onBackpressureDrop(); // Explicitly handling backpressure

        flowable
            .observeOn(Schedulers.io())
            .subscribe(
                data -> {
                    Thread.sleep(10); // Simulate slow consumer
                    System.out.println("Received: " + data);
                },
                Throwable::printStackTrace
            );
    }
}

কোন স্ট্র্যাটেজি কবে ব্যবহার করবেন?

উপসংহার

RxJava-এর BackpressureStrategy ডেটা স্ট্রিম পরিচালনার জন্য গুরুত্বপূর্ণ, বিশেষত যখন প্রডিউসার দ্রুত গতিতে ডেটা তৈরি করে এবং কনজিউমার সেগুলো প্রক্রিয়া করতে পারে না। উপযুক্ত স্ট্র্যাটেজি নির্বাচন করলে অ্যাপ্লিকেশন আরও কার্যকর এবং মেমোরি ব্যবহারে দক্ষ হয়।