RxGo के साथ Go में रिएक्टिव प्रोग्रामिंग: संपूर्ण गाइड

Go (Golang) में RxGo लाइब्रेरी के साथ रिएक्टिव प्रोग्रामिंग

रिएक्टिव प्रोग्रामिंग एक ऐसी प्रतिमान है जो एसिंक्रोनस डेटा स्ट्रीम और परिवर्तनों के प्रसार पर केंद्रित है। Go (Golang) इकोसिस्टम में इस प्रतिमान के सबसे शक्तिशाली कार्यान्वयनों में से एक है RxGo लाइब्रेरी (Reactive Extensions for Go)। यह डेवलपर्स को घोषणात्मक शैली का उपयोग करके घटनाओं के अनुक्रमों के साथ काम करने की अनुमति देता है, जो विशेष रूप से उपयोगकर्ता इनपुट, नेटवर्क अनुरोधों, चैनलों से संदेशों और किसी भी अन्य एसिंक्रोनस डेटा स्रोतों को संसाधित करते समय उपयोगी है।

RxGo क्या है?

RxGo, लोकप्रिय ReactiveX (Reactive Extensions) लाइब्रेरी का Go भाषा पर पोर्ट है। मुख्य विचार एक Observable (अवलोकनीय स्ट्रीम) बनाना है, जो तत्वों का उत्सर्जन करता है, और एक Observer (पर्यवेक्षक) जो इस स्ट्रीम की सदस्यता लेता है और आने वाले डेटा, त्रुटियों या स्ट्रीम के समाप्त होने पर प्रतिक्रिया करता है।

लाइब्रेरी स्ट्रीम को बदलने, फ़िल्टर करने, संयोजित करने और प्रबंधित करने के लिए ऑपरेटरों का एक समृद्ध सेट प्रदान करती है। मुख्य लाभ यह है कि कोड संक्षिप्त, पढ़ने में आसान और आसानी से परीक्षण योग्य हो जाता है, खासकर कई घटना स्रोतों वाले जटिल परिदृश्यों में।

RxGo की स्थापना

आरंभ करने के लिए, go get के माध्यम से पैकेज स्थापित करें:

go get -u github.com/reactivex/rxgo/v2

सुनिश्चित करें कि आप v2 संस्करण का उपयोग कर रहे हैं, क्योंकि इसमें नवीनतम API और Go कॉन्टेक्स्ट के लिए समर्थन शामिल है।

बुनियादी अवधारणाएँ और उदाहरण

1. Observable बनाना

स्ट्रीम बनाने का सबसे सरल तरीका rxgo.Just या rxgo.FromChannel का उपयोग करना है। Just के साथ एक उदाहरण देखें:

package main

import (
    "context"
    "fmt"
    "github.com/reactivex/rxgo/v2"
)

func main() {
    observable := rxgo.Just(1, 2, 3, 4, 5)()
    ch := observable.Observe()
    for item := range ch {
        if item.Error() {
            fmt.Println("त्रुटि:", item.Err)
        } else {
            fmt.Println("प्राप्त:", item.V)
        }
    }
}

यह कोड पाँच संख्याओं की एक स्ट्रीम बनाता है। Observe() विधि एक चैनल लौटाती है जिसके माध्यम से तत्व आते हैं। महत्वपूर्ण: RxGo में backpressure (प्रतिदबाव) का उपयोग किया जाता है - यदि उपभोक्ता धीमा है, तो डेटा बफ़र किया जाता है।

2. रूपांतरण ऑपरेटर

ऑपरेटर रिएक्टिव प्रोग्रामिंग का हृदय हैं। उदाहरण के लिए, Map प्रत्येक तत्व को रूपांतरित करता है:

package main

import (
    "context"
    "fmt"
    "github.com/reactivex/rxgo/v2"
)

func main() {
    observable := rxgo.Just(1, 2, 3)().
        Map(func(ctx context.Context, i interface{}) (interface{}, error) {
            num := i.(int)
            return num * 10, nil
        }, rxgo.WithPool(2)) // समानांतर प्रसंस्करण

    for item := range observable.Observe() {
        fmt.Println(item.V)
    }
}

आउटपुट: 10, 20, 30। Map ऑपरेटर प्रत्येक तत्व पर एक फ़ंक्शन लागू करता है। पैरामीटर rxgo.WithPool(2) दो गोरूटीन के साथ समानांतर प्रसंस्करण सक्षम करता है।

3. स्ट्रीम को फ़िल्टर करना और संयोजित करना

Filter की मदद से अनावश्यक तत्वों को हटाया जा सकता है, और Merge कई स्ट्रीम को जोड़ता है:

package main

import (
    "fmt"
    "github.com/reactivex/rxgo/v2"
)

func main() {
    // सम संख्याओं को फ़िल्टर करना
    even := rxgo.Just(1, 2, 3, 4, 5)().
        Filter(func(i interface{}) bool {
            return i.(int)%2 == 0
        })

    // दूसरी स्ट्रीम
    odd := rxgo.Just(10, 20, 30)()

    // संयोजन
    merged := rxgo.Merge(even, odd)

    for item := range merged.Observe() {
        fmt.Println(item.V) // 2, 4, 10, 20, 30
    }
}

4. चैनल और कॉन्टेक्स्ट के साथ काम करना

RxGo, Go के नेटिव चैनलों के साथ अच्छी तरह से एकीकृत होता है। चैनल से Observable बनाएं और कॉन्टेक्स्ट के माध्यम से रद्दीकरण जोड़ें:

package main

import (
    "context"
    "fmt"