Übersetzung des PHP-Backends auf den Redis Streams-Bus und Auswahl einer framework-unabhängigen Bibliothek

Vorwort

Meine Website, die ich als Hobby betreibe, dient der Speicherung interessanter persönlicher Webseiten. Dieses Thema hat mich zu Beginn meiner Programmierreise interessiert, als ich bewunderte, wie große Profis über sich selbst, ihre Hobbys und Projekte schrieben. Die Gewohnheit, sie zu entdecken, ist mir bis heute geblieben: Auf fast jeder kommerziellen und weniger kommerziellen Website schaue ich weiterhin in den Footer auf der Suche nach Links zu den Autoren.

Umsetzung der Idee

Die erste Version war einfach eine HTML-Seite auf meiner persönlichen Website, auf der ich Links mit Beschreibungen in einer ul-Liste gesammelt habe. Nachdem ich über einige Zeit etwa 20 Seiten erstellt hatte, begann ich zu denken, dass dies nicht sehr effizient ist, und entschied mich, den Prozess zu automatisieren. Auf Stack Overflow bemerkte ich, dass viele Leute ihre Websites in ihren Profilen angaben, weshalb ich einen Parser in PHP schrieb, der einfach die Profile durchging, beginnend mit dem ersten (die Adressen auf SO sind bis heute in folgender Form: `\/users\/1`), extrahierte die Links aus dem entsprechenden Tag und speicherte sie in SQLite.

Das kann man die zweite Version nennen: eine Sammlung von zehntausenden URLs in einer SQLite-Tabelle, die die statische Liste in HTML ersetzt hat. Auf dieser Liste habe ich eine einfache Suche erstellt. Da nur URLs vorhanden waren, war die Suche einfach danach.

In dieser Phase ließ ich das Projekt ruhen und kehrte lange Zeit später zurück. Zu diesem Zeitpunkt hatte meine Berufserfahrung über drei Jahre betragen und ich fühlte, dass ich etwas Seriöseres machen konnte. Zudem hatte ich den starken Wunsch, vergleichsweise neue Technologien für mich zu erschließen.

Moderne Version

Projekt In Docker bereitgestellt, die Datenbank auf MongoDB migriert, und seit relativ kurzer Zeit wurde Redis hinzugefügt, das zunächst nur zum Caching diente. Als Grundlage wird eines der PHP-Mikroframeworks verwendet.

Problem

Neue Websites werden über einen Konsolenbefehl hinzugefügt, der synchron Folgendes macht:

• Lädt Inhalte über die URL herunter
• Setzt ein Flag, ob HTTPS verfügbar war
• Speichert die Entität der Website
• Speichert den ursprünglichen HTML und die Header in der Historie der "Indexierung"
• Parst die Inhalte, extrahiert Title und Description
• Die Daten werden in einer separaten Sammlung gespeichert

Das war ausreichend, um einfach Websites zu speichern und sie in einer Liste anzuzeigen:

Die Idee, alles automatisch zu indizieren, zu kategorisieren und zu bewerten, während alles auf dem neuesten Stand gehalten wird, passte jedoch nur schwach in dieses Paradigma. Selbst das einfache Hinzufügen einer Web-Methode zum Hinzufügen von Seiten erforderte die Duplizierung von Code und Sperren, um potenzielles DDoS zu vermeiden.

Generell kann alles natürlich auch synchron erfolgen, während im Web-Service einfach die URL gespeichert wird, damit das monströse Daemon alle Aufgaben für die URLs aus der Liste ausführt. Doch auch hier drängt sich das Wort „Warteschlange“ auf. Und wenn man eine Warteschlange implementiert, kann man alle Aufgaben aufteilen und zumindest asynchron ausführen.

Lösung

Warteschlangen einführen und ein ereignisgesteuertes System zur Verarbeitung aller Aufgaben erstellen. Ich wollte schon lange Redis Streams ausprobieren.

Verwendung von Redis Streams in PHP

Da mein Framework nicht zu den drei Giganten Symfony, Laravel, Yii gehört, wollte ich eine unabhängige Bibliothek finden. Aber wie sich herausstellte (bei der ersten Durchsicht) ist es unmöglich, ernsthafte Bibliotheken zu finden. Alles, was mit Warteschlangen zu tun hat, ist entweder ein kleines Projekt mit 3 Commits aus vor fünf Jahren oder an ein Framework gebunden.

Ich habe von Symfony gehört, dass es nützliche Komponenten anbietet, und einige davon benutze ich bereits. Auch von Laravel kann man einige Dinge verwenden, wie z. B. deren ORM, ohne das Framework selbst zu benötigen.

symfony/messenger

Der erste Kandidat schien sofort optimal und ich habe ihn ohne jegliche Zweifel installiert. Aber es war schwieriger, Beispiele für die Verwendung außerhalb von Symfony zu finden. Wie kann man aus einer Vielzahl von Klassen mit universellen, bedeutungslosen Namen eine Nachrichtenübertragungsbus aufbauen, der auch noch auf Redis basiert?

Die Dokumentation auf der offiziellen Website war ausreichend detailliert, jedoch wurde die Initialisierung nur für Symfony mit ihrer bevorzugten YML und anderen magischen Methoden für Nicht-Symfonisten beschrieben. Ich hatte kein Interesse am Installationsprozess, besonders nicht während der Neujahrsferien. Aber ich musste mich damit beschäftigen und unerwartet lange.

Der Versuch, das System mit den Quellcodes von Symfony zu instanziieren, ist ebenfalls keine triviale Aufgabe bei knappen Fristen:

Nachdem ich in allem herumgestochert und versucht habe, irgendetwas manuell zu tun, kam ich zu dem Schluss, dass ich an irgendwelchen Provisorien arbeite und beschloss, etwas anderes auszuprobieren.

illuminate/queue

Es stellte sich heraus, dass diese Bibliothek fest an die Laravel-Infrastruktur und eine Menge anderer Abhängigkeiten gebunden ist, weshalb ich nicht viel Zeit damit verbracht habe: Ich habe sie installiert, angesehen, die Abhängigkeiten entdeckt und sie dann entfernt.

yiisoft/yii2-queue

Hier war schon aus dem Namen klar, dass es eine starke Bindung an Yii2 gibt. Ich hatte diese Bibliothek bereits benutzt und sie war nicht schlecht, aber dass sie vollständig von Yii2 abhängt, habe ich nicht bedacht.

Sonstige

Alles andere, was ich auf GitHub gefunden habe – unzuverlässige, veraltete und aufgegebene Projekte ohne Sterne, Forks und viele Commits.

Rückkehr zu symfony/messenger, technische Details

Ich musste mich mit dieser Bibliothek auseinandersetzen und nach etwas weiterer Zeit gelang es mir. Es stellte sich heraus, dass alles recht übersichtlich und einfach ist. Für die Instanziierung der Busse habe ich eine kleine Fabrik erstellt, da ich mehrere Busse mit verschiedenen Handlern plante.

Nur ein paar Schritte:

• Wir erstellen Nachrichtenhandler, die einfach callable sein sollten.
• Wir wickeln sie in HandlerDescriptor (eine Klasse aus der Bibliothek) ein.
• Diese „Deskriptoren“ wickeln wir in eine Instanz von HandlersLocator ein.
• Wir fügen HandlersLocator einer Instanz von MessageBus hinzu.
• Wir übergeben SendersLocator eine Menge von `SenderInterface`, in meinem Fall Instanzen der Klassen `RedisTransport`, die auf offensichtlich Weise konfiguriert werden.
• Wir fügen SendersLocator einer Instanz von MessageBus hinzu.

MessageBus hat die Methode `->dispatch()`, die die entsprechenden Handler in HandlersLocator sucht und die Nachricht an sie übergibt, wobei die entsprechenden `SenderInterface` zum Versenden über den Bus (Redis Streams) verwendet werden.

In der Konfiguration des Containers (in diesem Fall php-di) kann diese gesamte Verkettung so konfiguriert werden:

        CONTAINER_REDIS_TRANSPORT_SECRET => function (ContainerInterface $c) {
            return new RedisTransport(
                $c->get(CONTAINER_REDIS_STREAM_CONNECTION_SECRET),
                $c->get(CONTAINER_SERIALIZER))
            ;
        },
        CONTAINER_REDIS_TRANSPORT_LOG => function (ContainerInterface $c) {
            return new RedisTransport(
                $c->get(CONTAINER_REDIS_STREAM_CONNECTION_LOG),
                $c->get(CONTAINER_SERIALIZER))
            ;
        },
        CONTAINER_REDIS_STREAM_RECEIVER_SECRET => function (ContainerInterface $c) {
            return new RedisReceiver(
                $c->get(CONTAINER_REDIS_STREAM_CONNECTION_SECRET),
                $c->get(CONTAINER_SERIALIZER)
            );
        },
        CONTAINER_REDIS_STREAM_RECEIVER_LOG => function (ContainerInterface $c) {
            return new RedisReceiver(
                $c->get(CONTAINER_REDIS_STREAM_CONNECTION_LOG),
                $c->get(CONTAINER_SERIALIZER)
            );
        },
        CONTAINER_REDIS_STREAM_BUS => function (ContainerInterface $c) {
            $sendersLocator = new SendersLocator([
                AppMessagesSecretJsonMessages::class => [CONTAINER_REDIS_TRANSPORT_SECRET],
                AppMessagesDaemonLogMessage::class => [CONTAINER_REDIS_TRANSPORT_LOG],
            ], $c);
            $middleware[] = new SendMessageMiddleware($sendersLocator);

            return new MessageBus($middleware);
        },
        CONTAINER_REDIS_STREAM_CONNECTION_SECRET => function (ContainerInterface $c) {
            $host = 'bu-02-redis';
            $port = 6379;
            $dsn = "redis://$host:$port";
            $options = [
                'stream' => 'secret',
                'group' => 'default',
                'consumer' => 'default',
            ];

            return Connection::fromDsn($dsn, $options);
        },
        CONTAINER_REDIS_STREAM_CONNECTION_LOG => function (ContainerInterface $c) {
            $host = 'bu-02-redis';
            $port = 6379;
            $dsn = "redis://$host:$port";
            $options = [
                'stream' => 'log',
                'group' => 'default',
                'consumer' => 'default',
            ];

            return Connection::fromDsn($dsn, $options);
        },

Hier ist zu sehen, dass wir im SendersLocator für zwei verschiedene Nachrichten einen unterschiedlichen „Transport“ zugewiesen haben, von denen jeder eine eigene Verbindung zu den entsprechenden Streams hat.

Ich habe ein separates Demoprojekt erstellt, das eine Anwendung aus drei Daemons zeigt, die über einen solchen Bus miteinander kommunizieren. https://github.com/backend-university/products/tree/master/products/02-redis-streams-bus.

Aber ich werde zeigen, wie ein Consumer aufgebaut sein kann:

verwende AppMessagesDaemonLogMessage;
verwende SymfonyComponentMessengerHandlerHandlerDescriptor;
verwende SymfonyComponentMessengerHandlerHandlersLocator;
verwende SymfonyComponentMessengerMessageBus;
verwende SymfonyComponentMessengerMiddlewareHandleMessageMiddleware;
verwende SymfonyComponentMessengerMiddlewareSendMessageMiddleware;
verwende SymfonyComponentMessengerTransportSenderSendersLocator;

require_once __DIR__ . '/../../vendor/autoload.php';
/** @var PsrContainerContainerInterface $container */
$container = require_once('config/container.php');

$handlers = [
    DaemonLogMessage::class => [
        new HandlerDescriptor(
            function (DaemonLogMessage $m) {
                error_log('DaemonLogHandler: Nachricht verarbeitet: / ' . $m->getMessage());
            },
            ['from_transport' => CONTAINER_REDIS_TRANSPORT_LOG]
        )
    ],
];
$middleware = [];
$middleware[] = new HandleMessageMiddleware(new HandlersLocator($handlers));
$sendersLocator = new SendersLocator(['*' => [CONTAINER_REDIS_TRANSPORT_LOG]], $container);
$middleware[] = new SendMessageMiddleware($sendersLocator);

$bus = new MessageBus($middleware);
$receivers = [
    CONTAINER_REDIS_TRANSPORT_LOG => $container->get(CONTAINER_REDIS_STREAM_RECEIVER_LOG),
];
$w = new SymfonyComponentMessengerWorker($receivers, $bus, $container->get(CONTAINER_EVENT_DISPATCHER));
$w->run();

Verwendung dieser Infrastruktur in der Anwendung

Nachdem ich den Bus in meinem Backend implementiert hatte, habe ich separate Schritte aus dem alten synchronen Befehl herausgefiltert und separate Handler erstellt, von denen jeder seine eigene Aufgabe hat.

Die Pipeline zum Hinzufügen einer neuen Website zur Datenbank sieht so aus:

Und sofort nach diesem Schritt wurde es viel einfacher, neue Funktionen hinzuzufügen, zum Beispiel das Abrufen und Parsen von RSS. Da dieser Prozess ebenfalls den ursprünglichen Inhalt benötigt, abonniert der Handler, der Links zu RSS abruft, wie auch der WebsiteIndexHistoryPersistor, die Nachricht „Content/HtmlContent“, verarbeitet sie und leitet die benötigte Nachricht weiter durch seine Pipeline.

Letztendlich gibt es mehrere Dämonen, von denen jeder nur Verbindungen zu den benötigten Ressourcen hält. Zum Beispiel hält der Dämon crawlers alle Handler, die einen Internetzugang für Inhalte benötigen, während der Dämon persister eine Verbindung zur Datenbank hält.

Jetzt werden anstelle von Selects aus der Datenbank die erforderlichen IDs nach der Einfügung durch den Persister einfach über den Bus an alle interessierten Handler weitergegeben.

Quelle: habr.com