PHP में रिएक्टिव प्रोग्रामिंग: एसिंक्रोनस स्ट्रीम के लिए RxPHP लाइब्रेरी

PHP में रिएक्टिव प्रोग्रामिंग: RxPHP के लिए व्यावहारिक गाइड

रिएक्टिव प्रोग्रामिंग एक प्रतिमान है जो एसिंक्रोनस डेटा स्ट्रीम के साथ काम करने पर केंद्रित है। PHP में, जो पारंपरिक रूप से एक सिंक्रोनस भाषा है, इस अवधारणा को RxPHP (PHP के लिए ReactiveX) लाइब्रेरी के माध्यम से लागू किया गया है। यह इवेंट, HTTP अनुरोध, फ़ाइल स्ट्रीम और उपयोगकर्ता क्रियाओं को एक एकीकृत, घोषणात्मक शैली में संसाधित करने की अनुमति देता है।

RxPHP क्या है?

RxPHP लोकप्रिय ReactiveX लाइब्रेरी का एक पोर्ट है। मुख्य घटक:

• Observable — डेटा का स्रोत जो इवेंट "उत्सर्जित" करता है (onNext, onError, onCompleted)।
• Observer — सब्सक्राइबर जो इवेंट पर प्रतिक्रिया करता है।
• Schedulers — निष्पादन संदर्भ प्रबंधित करते हैं (सिंक्रोनस, एसिंक्रोनस, नए थ्रेड पर)।
• ऑपरेटर — स्ट्रीम को बदलने, फ़िल्टर करने और संयोजित करने के लिए फ़ंक्शन (map, filter, merge, zip)।

इंस्टॉलेशन और सेटअप

RxPHP के साथ काम करने के लिए Composer का उपयोग करें:

composer require reactivex/rxphp

मूल उदाहरण: Observable बनाना और सब्सक्राइब करना:

<?phpuse Rx\\Observable;use Rx\\ObserverInterface;

$observable = Observable::fromArray([1, 2, 3, 4, 5]);

$observer = new class implements ObserverInterface {    public function onNext($value) {        echo "प्राप्त: $value";    }    public function onError(\\Throwable $error) {        echo "त्रुटि: " . $error->getMessage() . "";    }    public function onCompleted() {        echo "पूर्ण!";    }};

$observable->subscribe($observer);// आउटपुट:// प्राप्त: 1// प्राप्त: 2// प्राप्त: 3// प्राप्त: 4// प्राप्त: 5// पूर्ण!?>

RxPHP के मुख्य ऑपरेटर

ऑपरेटर मूल Observable को बदले बिना डेटा को रूपांतरित करने की अनुमति देते हैं:

1. map — प्रत्येक तत्व का रूपांतरण

$observable = Observable::fromArray([1, 2, 3])    ->map(function($value) {        return $value * 2;    });

$observable->subscribe(function($value) {    echo $value . "";});// 2, 4, 6

2. filter — शर्त के अनुसार चयन

$observable = Observable::fromArray([1, 2, 3, 4, 5])    ->filter(function($value) {        return $value % 2 === 0;    });

// परिणाम: 2, 4

3. merge — स्ट्रीम का संयोजन

$stream1 = Observable::fromArray([1, 2]);$stream2 = Observable::fromArray([3, 4]);

$merged = Observable::merge($stream1, $stream2);// 1, 2, 3, 4 (क्रम Scheduler पर निर्भर हो सकता है)

कस्टम Observable बनाना

किसी भी स्रोत से Observable बनाया जा सकता है: डेटाबेस, सॉकेट या उपयोगकर्ता इवेंट।

<?phpuse Rx\\Observable;use Rx\\Scheduler\\EventLoopScheduler;

// एसिंक्रोनस फ़ाइल रीडिंग का अनुकरणfunction readFileAsync(string $filename): Observable {    return Observable::create(function($observer) use ($filename) {        $handle = fopen($filename, 'r');        if (!$handle) {            $observer->onError(new \\RuntimeException("फ़ाइल खोलने में विफल"));            return;        }                while (($line = fgets($handle)) !== false) {            $observer->onNext(trim($line));        }        fclose($handle);        $observer->onCompleted();    });}

// EventLoop के साथ उपयोग (react/event-loop आवश्यक है)$loop = \\React\\EventLoop\\Loop::get();$scheduler = new EventLoopScheduler($loop);

readFileAsync('data.log')    ->subscribeOn($scheduler)    ->filter(function($line) {        return strpos($line, 'ERROR') !== false;    })    ->subscribe(function($line) {        echo "त्रुटि मिली: $line";    });

$loop->run();?>

त्रुटि प्रबंधन और समाप्ति

रिएक्टिव शैली में, त्रुटियों को संभाला जाता है