Reaktive Programmierung in PHP mit RxPHP

Reaktive Programmierung in PHP: Praxisleitfaden für RxPHP

Reaktive Programmierung ist ein Paradigma, das auf die Arbeit mit asynchronen Datenströmen ausgerichtet ist. In PHP, einer traditionell synchronen Sprache, wird dieses Konzept durch die Bibliothek RxPHP (ReactiveX für PHP) umgesetzt. Sie ermöglicht die Verarbeitung von Ereignissen, HTTP-Anfragen, Dateiströmen und Benutzeraktionen in einem einheitlichen, deklarativen Stil.

Was ist RxPHP?

RxPHP ist ein Port der beliebten Bibliothek ReactiveX. Die Hauptkomponenten sind:

Observable — eine Datenquelle, die Ereignisse "emittiert" (onNext, onError, onCompleted).
Observer — ein Abonnent, der auf Ereignisse reagiert.
Schedulers — verwalten den Ausführungskontext (synchron, asynchron, in einem neuen Thread).
Operatoren — Funktionen zur Transformation, Filterung und Kombination von Streams (map, filter, merge, zip).

Installation und Einrichtung

Für die Arbeit mit RxPHP verwenden Sie Composer:

composer require reactivex/rxphp

Ein einfaches Beispiel: Erstellen eines Observable und Abonnieren:

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

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

$observer = new class implements ObserverInterface {    public function onNext($value) {        echo "Empfangen: $value";    }    public function onError(\\Throwable $error) {        echo "Fehler: " . $error->getMessage() . "";    }    public function onCompleted() {        echo "Abgeschlossen!";    }};

$observable->subscribe($observer);// Ausgabe:// Empfangen: 1// Empfangen: 2// Empfangen: 3// Empfangen: 4// Empfangen: 5// Abgeschlossen!?>

Wichtige Operatoren in RxPHP

Operatoren ermöglichen die Transformation von Daten, ohne das ursprüngliche Observable zu verändern:

1. map — Transformation jedes Elements

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

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

2. filter — Auswahl nach Bedingung

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

// Ergebnis: 2, 4

3. merge — Zusammenführen von Streams

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

$merged = Observable::merge($stream1, $stream2);// 1, 2, 3, 4 (Reihenfolge kann vom Scheduler abhängen)

Erstellen benutzerdefinierter Observable

Sie können Observable aus beliebigen Quellen erstellen: Datenbanken, Sockets oder benutzerdefinierten Ereignissen.

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

// Emulation eines asynchronen Dateilesevorgangsfunction readFileAsync(string $filename): Observable {    return Observable::create(function($observer) use ($filename) {        $handle = fopen($filename, 'r');        if (!$handle) {            $observer->onError(new \\RuntimeException("Datei konnte nicht geöffnet werden"));            return;        }                while (($line = fgets($handle)) !== false) {            $observer->onNext(trim($line));        }        fclose($handle);        $observer->onCompleted();    });}

// Verwendung mit EventLoop (erfordert 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 "Fehler gefunden: $line";    });

$loop->run();?>