Reaktive Programmierung in PHP mit RxPHP: Vollständiger Leitfaden

Reaktive Programmierung in PHP mit RxPHP

\n\n

Reaktive Programmierung ist ein Paradigma, das sich auf asynchrone Datenströme und die Verbreitung von Änderungen konzentriert. In PHP wird dieses Paradigma durch die Bibliothek RxPHP (ReactiveX für PHP) umgesetzt, die es ermöglicht, mit Ereignissen, asynchronen Anfragen und Datensequenzen als beobachtbare Ströme (Observables) zu arbeiten.

\n\n

Was ist RxPHP?

\n

RxPHP ist eine PHP-portierte Version des beliebten Frameworks ReactiveX (Rx). Es bietet Werkzeuge zur Komposition asynchroner und ereignisorientierter Programme mithilfe von beobachtbaren Sequenzen und LINQ-ähnlichen Abfrageoperatoren.

\n\n

Die grundlegenden Konzepte von RxPHP:

\n

\n
• Observable — eine Datenquelle, die Werte, Fehler oder ein Abschlusssignal emittieren kann.
\n
• Observer — ein Abonnent, der auf Ereignisse vom Observable reagiert.
\n
• Subscription — die Verbindung zwischen Observable und Observer, die das Abbestellen ermöglicht.
\n
• Operators — Funktionen zur Transformation, Filterung und Kombination von Strömen (map, filter, merge, zip usw.).
\n
• Scheduler — Steuerung der Ausführung (synchron, asynchron, mit Verzögerung).
\n

\n\n

Installation von RxPHP

\n

Die Installation erfolgt über Composer:

\n

composer require reactivex/rxphp

\n\n

Beispiel 1: Erstellen eines einfachen Observable

\n

Erstellen wir einen Strom, der nacheinander drei Zahlen ausgibt und dann endet:

\n

<?php\nrequire 'vendor/autoload.php';\n\nuse Rx\\Observable;\n\n$observable = Observable::fromArray([10, 20, 30]);\n\n$disposable = $observable->subscribe(\n    function ($value) {\n        echo "Erhalten: $value\\n";\n    },\n    function (\\Throwable $error) {\n        echo "Fehler: " . $error->getMessage() . "\\n";\n    },\n    function () {\n        echo "Abgeschlossen!\\n";\n    }\n);\n

\n

Ausgabe:

\n

Erhalten: 10\nErhalten: 20\nErhalten: 30\nAbgeschlossen!

\n\n

Beispiel 2: Verwendung der Operatoren map und filter

\n

Transformieren wir einen Zahlenstrom: Filtern wir gerade Zahlen und multiplizieren sie mit 10:

\n

<?php\nuse Rx\\Observable;\n\n$observable = Observable::fromArray([1, 2, 3, 4, 5])\n    ->filter(function ($value) {\n        return $value % 2 === 0;\n    })\n    ->map(function ($value) {\n        return $value * 10;\n    });\n\n$observable->subscribe(function ($value) {\n    echo "Ergebnis: $value\\n";\n});\n

\n

Ausgabe:

\n

Ergebnis: 20\nErgebnis: 40

\n\n

Beispiel 3: Asynchrone Arbeit mit HttpClient

\n

RxPHP eignet sich hervorragend für die Arbeit mit asynchronen HTTP-Anfragen. Verwenden wir Observable::fromPromise mit der Bibliothek GuzzleHttp:

\n

<?php\nuse Rx\\Observable;\nuse GuzzleHttp\\Client;\nuse GuzzleHttp\\Promise;\n\n$client = new Client();\n$promise1 = $client->getAsync('https://api.example.com/data1');\n$promise2 = $client->getAsync('https://api.example.com/data2');\n\nObservable::fromArray([$promise1, $promise2])\n    ->flatMap(function ($promise) {\n        return Observable::fromPromise($promise);\n    })\n    ->map(function ($response) {\n        return json_decode($response->getBody(), true);\n    })\n    ->subscribe(function ($data) {\n        print_r($data);\n    });\n

\n\n

Beispiel 4: Zusammenführen von Strömen (merge und zip)

\n

Häufig müssen mehrere Ströme zu einem zusammengeführt werden. Betrachten wir zwei Operatoren:

\n\n

merge — führt Ströme zusammen, indem ihre Elemente beim Eintreffen abwechselnd ausgegeben werden:

\n

<?php\nuse Rx\\Observable;\n\n$stream1 = Observable::fromArray(['A1', 'A2', 'A3']);\n$stream2 = Observable::fromArray(['B1', 'B2']);\n\nObservable::merge([$stream1, $stream2])\n    ->subscribe(function ($value) {\n        echo "$value ";\n    });\n// Ausgabe: A1 B1 A2 B2 A3\n

\n\n