Java में रिएक्टिव प्रोग्रामिंग: Reactor लाइब्रेरी

Java में Project Reactor के साथ रिएक्टिव प्रोग्रामिंग

रिएक्टिव प्रोग्रामिंग एक डेवलपमेंट पैराडाइम है जो डेटा के एसिंक्रोनस फ्लो और बदलावों के प्रसार पर केंद्रित है। Java में यह कॉन्सेप्ट Reactive Streams स्पेसिफिकेशन और इसकी सबसे लोकप्रिय लाइब्रेरी Project Reactor (Spring WebFlux इकोसिस्टम का हिस्सा) के जरिए लागू किया गया है।

पारंपरिक इम्पेरेटिव कोड के विपरीत, जहां प्रोग्राम कमांड को क्रम से एक्जीक्यूट करता है, रिएक्टिव अप्रोच डेटा को आते ही प्रोसेस करने की अनुमति देता है, बिना थ्रेड्स को ब्लॉक किए। यह हाई-लोड सिस्टम्स, माइक्रोसर्विसेज और एक साथ कई रिक्वेस्ट वाले ऐप्लिकेशन के लिए बेहद जरूरी है।

Reactor के मुख्य कॉम्पोनेंट्स

यह लाइब्रेरी दो मुख्य प्रकार की रिएक्टिव सीक्वेंस प्रदान करती है:

• Mono<T> — एक ऐसा फ्लो जिसमें 0 या 1 एलिमेंट हो सकता है। इसका उपयोग सिंगल एसिंक्रोनस ऑपरेशन (जैसे HTTP रिक्वेस्ट, DB से रिकॉर्ड लाना) के लिए किया जाता है।
• Flux<T> — एक ऐसा फ्लो जिसमें 0 से N एलिमेंट हो सकते हैं। यह कलेक्शन, इवेंट स्ट्रीम या टुकड़ों में आने वाले डेटा के लिए उपयुक्त है।

दोनों टाइप Reactive Streams स्पेसिफिकेशन के Publisher इंटरफेस को इम्प्लीमेंट करते हैं और backpressure मैकेनिज्म को सपोर्ट करते हैं — जिससे सब्सक्राइबर डेटा रिसीव करने की स्पीड को कंट्रोल कर सकता है।

उदाहरण: Flux बनाना और प्रोसेस करना

आइए एक सरल उदाहरण देखें जहां हम नंबरों का रिएक्टिव फ्लो बनाते और प्रोसेस करते हैं:

import reactor.core.publisher.Flux;

public class ReactorExample {
    public static void main(String[] args) {
        Flux<Integer> numbers = Flux.just(1, 2, 3, 4, 5);

        numbers
            .map(n -> n * 2)          // ट्रांसफॉर्मेशन
            .filter(n -> n % 3 == 0)   // फिल्टरेशन
            .subscribe(
                data -> System.out.println("प्राप्त: " + data),
                error -> System.err.println("त्रुटि: " + error),
                () -> System.out.println("पूर्ण हुआ!")
            );
    }
}

इस उदाहरण में:

• Flux.just() पांच नंबरों का फ्लो बनाता है।
• map() हर एलिमेंट को 2 से गुणा करता है।
• filter() सिर्फ 3 से विभाज्य नंबरों को रखता है।
• subscribe() प्रोसेसिंग शुरू करता है — रिजल्ट प्रिंट करता है, एरर हैंडल करता है और पूरा होने का सिग्नल देता है।

एसिंक्रोनसी और Mono के साथ काम करना

Reactor थ्रेड्स को ब्लॉक किए बिना एसिंक्रोनस ऑपरेशन करने की अनुमति देता है। उदाहरण के लिए, रिमोट सर्विस पर रिक्वेस्ट का सिमुलेशन:

import reactor.core.publisher.Mono;
import java.time.Duration;

public class AsyncExample {
    public static void main(String[] args) throws InterruptedException {
        Mono<String> asyncResult = Mono.fromSupplier(() -> {
            try {
                Thread.sleep(1000); // देरी का अनुकरण
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "डेटा प्राप्त हुआ";
        }).subscribeOn(Schedulers.boundedElastic()); // दूसरे थ्रेड में एक्जीक्यूट करें

        System.out.println("रिक्वेस्ट भेज दी गई...");
        
        asyncResult.subscribe(System.out::println);
        
        Thread.sleep(2000); // पूरा होने का इंतजार
    }
}

मुख्य बात: .subscribeOn() कॉल भारी ऑपरेशन को अलग थ्रेड में ले जाता है, मेन थ्रेड को ब्लॉक नहीं करता। सब्सक्राइबर को रिजल्ट एसिंक्रोनसली मिलेगा।

रिएक्टिव फ्लो में एरर हैंडलिंग

try-catch के विपरीत, रिएक्टिव प्रोग्रामिंग में एरर को ऑपरेटर्स के जरिए हैंडल किया जाता है:

Flux<String> fluxWithError = Flux.just("A", "B", "C")
    .map(s -> {
        if ("B".equals(s)) throw new RuntimeException("B पर त्रुटि");
        return s.toLowerCase();
    })