😶‍🌫️ Diese Person existiert nicht – Synchrone PHP-zu-asynchrone Orchestrierung mit Flow

vom 13. Juni 2026

Sie müssen zehn Bilder von einer entfernten API herunterladen. Zwischen den einzelnen Anfragen warten Sie jeweils eine Sekunde, um ein Ratenlimit einzuhalten. Jedes Bild durchläuft denselben Prozess: Abrufen, Hashen, Speichern, Berichten.

Der Code sieht aus wie eine bekannte for-Schleife. Er funktioniert. Doch während ein Bild heruntergeladen wird, warten die anderen neun. Solange ein Prozess in einer sleep()-Funktion pausiert, steht der gesamte Prozess still. Die Arbeitseinheiten sind unabhängig – nur die Orchestrierung behandelt sie so, als wären sie es nicht.

Dies ist das konkrete Problem, das wir im Projekt flow-thispersondoesnotexist isoliert haben: ein minimaler Symfony-Befehl, der mehrere Bilder aus ThisPersonDoesNotExist abruft und auf der Festplatte speichert.

Ziel war es nicht, Bilder schneller herunterzuladen.

Ziel war es, zu verstehen, wie man eine sequentielle Ausführung mit darkwood/flow in eine asynchrone Orchestrierung umwandeln kann – ohne die Geschäftslogik zu ändern, sondern durch eine Änderung des Ausführungsmodells.

Der Übergang lässt sich in zwei Commits zusammenfassen:

• fae47e8 — Befehl app:fetch-thispersondoesnotexist, synchrone Schleife
• c61c929 — derselbe Befehl, orchestriert von Flow und FiberDriver

Zwischen den beiden Versionen ändert sich nur eine Datei: src/Command/FetchThisPersonDoesNotExistCommand.php.

Der Ausgangspunkt: eine synchrone Schleife

Die erste Version (fae47e8) verwendet eine klassische for-Schleife. Abrufen, Hashen, Speichern und Berichten werden inline ausgeführt; ein sleep(1) blockiert den Prozess zwischen den einzelnen Downloads:

$savedFiles = [];

for ($index = 1; $index <= $count; ++$index) {
    if ($count > 1) {
        $this->io->section(sprintf('Download %d/%d', $index, $count));
    }

    try {
        $image = $this->fetchImage();
        $checksum = $this->computeChecksum($image);
        $filename = $this->generateFilename($checksum, $file);
        $outputPath = $this->saveFile($filename, $image);
        $this->report($outputPath, $checksum, \strlen($image));
    } catch (\RuntimeException $exception) {
        $this->io->error($exception->getMessage());

        return Command::FAILURE;
    }

    $savedFiles[] = $outputPath;

    if ($index < $count) {
        $this->io->note(sprintf('Waiting %d second(s) before next download...', self::DOWNLOAD_DELAY_SECONDS));
        sleep(self::DOWNLOAD_DELAY_SECONDS);
    }
}

Der Prozess verläuft linear: Herunterladen, Verarbeiten, Warten, Weiter zum nächsten Bild. Jeder Schritt blockiert den Prozess vollständig – einschließlich der sleep()-Funktion, die das Programm selbst dann einfriert, wenn keine sinnvolle Arbeit ausgeführt wird.

Dieser Ansatz ist leicht verständlich, verschwendet aber E/A-Wartezeit: Solange ein Bild nicht fertig ist, kann kein anderes weiterbearbeitet werden.

Hinweis: Die synchrone Version behielt auch die alternative Methode fetchImageViaFileGetContents() bei. Diese wird in der Flow-Version entfernt – explizite Streams (fopen) werden zukünftig bevorzugt.

Beobachten Sie den verborgenen Arbeitsablauf

Hinter der Schleife verbirgt sich eine Pipeline, die für jedes Bild wiederholt wird:

Fetch Image → Save Image → Report Result

Image #1 : Fetch → Save → Report
Image #2 : Fetch → Save → Report
Image #3 : Fetch → Save → Report

Jedes Bild stellt eine in sich abgeschlossene Arbeitseinheit dar. Die Frage lautet daher: Warum warten, bis Bild 1 vollständig fertiggestellt ist, bevor mit Bild 2 begonnen wird?

Vorher/Nachher: ​​Die architektonische Veränderung

Vor der Einführung von Flow erfolgte die Ausführung streng sequenziell:

Image 1 → Fetch → Save → Sleep
Image 2 → Fetch → Save → Sleep
Image 3 → Fetch → Save → Sleep

Anschließend werden die Arbeitseinheiten unabhängig voneinander geplant und durchlaufen einen gemeinsamen Prozess:

Image 1 ─┐
Image 2 ─┼─> Flow Pipeline (Fetch → Save + Report) ─> résultats
Image 3 ─┘

Die for-Schleife steuert nicht mehr die schrittweise Ausführung. Sie reiht Ips (Arbeitseinheiten) in eine Warteschlange ein. Flow orchestriert deren Durchlauf durch die Pipeline; await() synchronisiert am Ende.

Einführung in Flow

Die Abhängigkeit darkwood/flow wird vor dem Befehl (e71899a) zum Projekt hinzugefügt. Version c61c929 verwendet sie schließlich zur Orchestrierung der Arbeit.

eingeführte Importe:

use Flow\Driver\FiberDriver;
use Flow\ExceptionInterface;
use Flow\FlowFactory;
use Flow\Ip;

Jedes Bild wird zu einer eigenständigen Arbeitseinheit.

In Flow wird diese Einheit durch einen Ip (Befehlszeiger) dargestellt:

$flow(new Ip($index));

Die Pipeline wird in Stufen mit FlowFactory und einem Generator beschrieben:

$driver = new FiberDriver();
$savedFiles = [];

$flow = (new FlowFactory())->create(function () use ($file, &$savedFiles, $driver) {
    yield [
        fn (int $index) => $this->fetchImage($index, $driver),
        fn (ExceptionInterface $exception) => throw new \RuntimeException($exception->getMessage()),
    ];
    yield function (string $image) use ($file, &$savedFiles): string {
        $outputPath = $this->saveAndReport($image, $file);
        $savedFiles[] = $outputPath;

        return $outputPath;
    };
}, ['driver' => $driver]);

for ($index = 1; $index <= $count; ++$index) {
    $flow(new Ip($index));
}

try {
    $flow->await();
} catch (\RuntimeException $exception) {
    $this->io->error($exception->getMessage());

    return Command::FAILURE;
}

Konzeptionell entspricht dies dem Schreiben:

Ip($index)
 ↓
Fetch Image        ← étape async (fiber + delay)
 ↓
Save + Report      ← étape suivante, reçoit le string image

Jedes Bild durchläuft exakt dieselbe Pipeline.

Die Verantwortlichkeiten werden mithilfe spezieller Methoden ermittelt:

• fetchImage(int $index, FiberDriver $driver): string — Herunterladen mit kooperativer Suspendierung
• saveAndReport(string $image, ?string $fileOverride): string — Hash, Speichern, Berichterstellung

Die Fehlerbehandlung beendet die Schleife: Ein errorJob im Fetch-Schritt löst eine RuntimeException aus, die nur einmal um await() herum abgefangen wird.

Ersetze die blockierende Wartezeit.

Die bedeutendste Änderung betrifft das Wartezeitmanagement.

In der synchronen Version ist die Verzögerung global und blockierend:

private const int DOWNLOAD_DELAY_SECONDS = 1;

// dans la boucle, entre chaque image :
sleep(self::DOWNLOAD_DELAY_SECONDS);

In der Flow-Version gilt die Verzögerung pro Faser und Kooperative:

private const int DELAY_MIN_SECONDS = 1;
private const int DELAY_MAX_SECONDS = 3;

// dans fetchImage(), avant le téléchargement :
$delay = random_int(self::DELAY_MIN_SECONDS, self::DELAY_MAX_SECONDS);
$this->io->note(sprintf('#%d: suspending fiber for %ds before download...', $index, $delay));
$driver->delay($delay);

$driver->delay() unterbricht nur die aktuell laufende Fiber. Die anderen Fibers werden fortgesetzt:

Image #1 attend (delay 2s)
Image #2 télécharge
Image #3 sauvegarde
Image #4 démarre son delay

Bei --count=5 nähert sich die Gesamtzeit eher max(delays + fetch) als deren Summe an.

Fasern als Grundlage für die Ausführung

Im Hintergrund basiert FiberDriver auf PHP 8.1 Fibers: Ein Fiber kann angehalten und später fortgesetzt werden, wodurch mehrere unabhängige Prozesse mit einem Code ausgeführt werden können, der dem klassischen PHP sehr nahe kommt.

$driver = new FiberDriver();

Jedes Bild hat seine eigene Fiber. Wenn eine Fiber wartet ($driver->delay()), setzen die anderen ihren Fortschritt fort. Das Logging spiegelt dieses Verhalten wider:

#2: suspending fiber for 2s before download...
#2: downloading (other fibers may run while this one waits)...
#2: fetch complete in 1.34s

Flow ersetzt Fibers nicht, sondern nutzt sie über einen Treiber. Die Wahl des Treibers bestimmt, wie die Pipeline ausgeführt wird; die Pipeline-Beschreibung bleibt unverändert.

Der Synchronisationspunkt

Die for-Abfrage identifiziert die Arbeitseinheiten:

for ($index = 1; $index <= $count; ++$index) {
    $flow(new Ip($index));
}

await() wird zum einzigen Synchronisationspunkt:

$flow->await();

Planifier Ip(1)
Planifier Ip(2)
Planifier Ip(3)
…
await() — barrière finale

Die Verantwortung für die Orchestrierung wurde an Flow übertragen. Die for-Schleife übernimmt diese Aufgabe nicht mehr: Sie reiht Einheiten in die Warteschlange ein. Flow führt diese parallel aus; await() wartet, bis die gesamte Pipeline abgeschlossen ist.

Warum PHP Streams beibehalten?

Das Projekt verwendet absichtlich:

$stream = fopen(self::FETCH_URL, 'r', false, $context);
$content = stream_get_contents($stream);

statt file_get_contents().

Ziel ist es nicht, einen HTTP-Client neu zu erfinden.

Ziel ist es, sich eng an die PHP-Grundfunktionen zu halten – der Codekommentar gibt dies ausdrücklich an:

Bevorzugt für Flow-Experimente: Das Ressourcenhandle ist das Primitive, das später mit stream_select(), dem nicht-blockierenden Modus oder Fibers verbunden werden kann.

Diese Grundformen können sich dann weiterentwickeln zu:

stream_set_blocking(false);
stream_select(...);

oder in fortgeschrittenere Treiber integriert werden (Amp, React, Swoole – alle werden von darkwood/flow unterstützt).

Streams bieten eine natürliche Grundlage für Experimente mit asynchronen Modellen in PHP.

Wettbewerbsbeschränkung (nächster Schritt)

In diesem Proof of Concept werden alle Bilder parallel gesendet (bis zu --count aktiven Fasern).

Für den Produktiveinsatz bietet Flow Strategien wie MaxIpStrategy:

use Flow\IpStrategy\MaxIpStrategy;

yield [$job1, $errorJob1, new MaxIpStrategy(5)];

Das offizielle Beispielpaket (examples/flow.php) verwendet MaxIpStrategy(2), um die Anzahl der gleichzeitig ausgeführten Jobs in jedem Schritt zu begrenzen.

Es ist noch nicht mit flow-thispersondoesnotexist verbunden, aber es ist der natürliche Hebel zur Ratenbegrenzung, ohne auf sequentielles sleep() zurückzugreifen.

Ein Modell, das weit über Bilder hinaus anwendbar ist

Das Beispiel ThisPersonDoesNotExist ist bewusst einfach gehalten. Dasselbe Muster findet sich in Darkwood-Pipelines wieder:

Scraping YouTube     → Lister → Télécharger / transcrire → Sauvegarder
Traitement vidéo     → Générer assets → Encoder → Persister
Agent IA             → Lire → Transformer → Publier

In allen Fällen handelt es sich um unabhängige Arbeitseinheiten, die eine Abfolge von Schritten durchlaufen. Der Arbeitsfluss zielt nicht darauf ab, eine Aufgabe auszuführen, sondern er orchestriert Arbeitsabläufe.

Migrationsübersicht

Was Flow nicht ist

Bevor wir zum Schluss kommen, noch eine hilfreiche Klarstellung, um häufige Missverständnisse zu vermeiden.

Flow ist keine Ereignisschleife. Es verarbeitet nicht direkt einen select / poll-Zyklus auf E/A-Deskriptoren.

Flow ist keine Implementierung von Fibers. Fibers sind ein PHP 8.1-Primitiv. Flow orchestriert sie über Treiber, ersetzt sie aber nicht.

Flow ist keine Laufzeitumgebung. Es verändert das PHP-Ausführungsmodell nicht. Es läuft in einem Standard-PHP-Prozess.

Flow ist kein Ersatz für Amp, ReactPHP oder Swoole. Diese Bibliotheken stellen Laufzeitumgebungen und Ereignisschleifen bereit. Flow kann auf diese – AmpDriver, ReactDriver, SwooleDriver – zurückgreifen, ohne mit ihnen in Konkurrenz zu treten.

Flow bietet ein Modell der Orchestrierung:

• Arbeitseinheiten beschreiben (Ip)
• Pipelines aus Schritten zusammenstellen (yield in FlowFactory)
• Delegieren Sie die Ausführung an einen ausgewählten Treiber (FiberDriver, AmpDriver, ReactDriver, SwooleDriver, …)

Das Orchestrierungsmodell ist vom Ausführungsmodell getrennt. Eine Pipeline kann einmal beschrieben werden, und der Treiber kann dann je nach Kontext geändert werden – CLI mit Fibers, Swoole-Worker, React-Service – ohne die Geschäftslogik neu schreiben zu müssen.

Abschluss

Der Übergang von synchron zu asynchron ist nicht in erster Linie eine Frage der Leistungsfähigkeit.

Es handelt sich um einen Wandel im mentalen Modell:

Avant : exécuter l'opération A, puis B, puis C — et attendre entre chaque étape.

Après : décrire un pipeline, enqueue des unités de travail,
        laisser l'orchestrateur planifier l'exécution.

Die for-Schleife wird zum IP-Produzenten. Die Schritte fetch und save werden zu verketteten Prozessen. await() ersetzt die Abfolge von sleep()-Aufrufen. Die Geschäftslogik (Herunterladen, Hashen, Speichern) bleibt unverändert; geändert hat sich lediglich, wer entscheidet, wann welcher Schritt ausgeführt wird.

Genau dieses Modell benötigt Darkwood für seine zukünftigen Pipelines: YouTube-Scraping und Transkriptextraktion, Medienverarbeitung in MediaBundle, mehrstufige KI-Agenten und die Orchestrierung langlaufender Workflows. Unabhängige Einheiten, kombinierbare Phasen und ein expliziter Synchronisationspunkt.

Das Projekt ThisPersonDoesNotExist ist die minimale Demonstration: eine einzelne Datei, eine zweistufige Pipeline, ein Treiber – ausreichend, um den Übergang zu veranschaulichen, ohne seinen Umfang zu verschleiern.

Um weiterzukommen

• POC-Repository: flow-thispersondoesnotexist — Commits fae47e8 → c61c929
• Paket: darkwood/flow — Beispiel examples/flow.php Frédéric Boucherys Artikel zeichnet die Geschichte der asynchronen Programmierung in PHP nach, von den in PHP 4.3 eingeführten Streams bis zu den modernen Fibers von PHP 8.1: https://f2r.github.io/fr/asynchrone.html