Vertaling van de PHP-backend naar de Redis streams-bus en keuze van een library die onafhankelijk is van frameworks

Voorwoord

Mijn website, die ik als hobby onderhoud, is bedoeld voor het opslaan van interessante startpagina's en persoonlijke websites. Dit onderwerp begon mij te interesseren aan het begin van mijn programmeerreis, toen ik gefascineerd was door de grote professionals die over zichzelf, hun hobby's en projecten schreven. De gewoonte om deze voor mezelf te ontdekken is blijven bestaan: bijna op elke commerciële en minder commerciële website blijf ik de footer bekijken op zoek naar links naar auteurs.

Implementatie van het idee

De eerste versie was gewoon een html-pagina op mijn persoonlijke website, waar ik links met beschrijvingen in een ul-lijst verzamelde. Nadat ik in een bepaalde tijd zo'n 20 pagina's had verzameld, begon ik te denken dat dit niet erg efficiënt was en besloot ik het proces te automatiseren. Op Stack Overflow merkte ik dat velen websites in hun profielen vermeldden, dus schreef ik een parser in PHP, die simpelweg door de profielen ging, beginnend bij de eerste (de adressen op SO zijn tot op de dag van vandaag van dit type: `\/users\/1`), en links uit de benodigde tag haalde en opsloeg in SQLite.

Dit kan de tweede versie worden genoemd: een verzameling van tienduizenden URL's in een SQLite-tabel, die de statische lijst in html verving. Op basis van deze lijst maakte ik een eenvoudige zoekfunctie. Aangezien er alleen URL's waren, was de zoekfunctie ook eenvoudig.

In dit stadium had ik het project even laten liggen en kwam er na lange tijd weer op terug. Op dit moment had ik al meer dan drie jaar ervaring en voelde ik dat ik iets serieuzers kon maken. Bovendien was er een grote wens om relatief nieuwe technologieën voor mezelf te verkennen.

Moderne versie

Project uitgedraaid in Docker, de database is omgezet naar MongoDB, en sinds kort is Redis toegevoegd, dat eerst alleen voor caching was. Als basis wordt een van de PHP-microframeworks gebruikt.

Probleem

Nieuwe websites worden toegevoegd met een console-opdracht die synchroon het volgende doet:

• Downloadt de inhoud via URL
• Stelt een vlag in of HTTPS toegankelijk was
• Beheert de entiteit van de website
• Slaat de oorspronkelijke HTML en headers op in de 'indexeringshistorie'
• Parseert de inhoud en haalt de Titel en Beschrijving eruit
• Slaat de gegevens op in een aparte collectie

Dat was genoeg om simpelweg websites op te slaan en ze in een lijst weer te geven:

Maar het idee om alles automatisch te indexeren, categoriseren en rangschikken, terwijl alles up-to-date wordt gehouden, past slecht in dit paradigma. Zelfs het simpele toevoegen van webmethoden om pagina's toe te voegen vereiste duplicatie van code en blokkades om potentiële DDoS-aanvallen te voorkomen.

Over het algemeen kan alles natuurlijk ook synchronisch gedaan worden, en kan in de webmethode eenvoudig de URL worden opgeslagen zodat het monsterlijke demon alle taken voor de URLs uit de lijst kan uitvoeren. Maar zelfs hier roept het woord 'wachtrij' zich op. En als we een wachtrij implementeren, kunnen we alle taken splitsen en tenminste asynchroon uitvoeren.

Oplossing

Implementeer wachtrijen en maak een event-driven systeem voor de verwerking van alle taken. En ik wilde al een tijdje Redis Streams uitproberen.

Het gebruik van Redis streams in PHP

Aangezien het framework dat ik gebruik niet uit de top drie van giganten zoals Symfony, Laravel, Yii komt, wilde ik ook een onafhankelijke bibliotheek vinden. Maar zoals het bleek (bij de eerste beoordeling) was het onmogelijk om serieuze aparte bibliotheken te vinden. Alles wat met wachtrijen te maken heeft, is ofwel een klein project met 3 commits van vijf jaar geleden, ofwel aan een framework gebonden.

Ik heb gehoord over Symfony als leverancier van nuttige afzonderlijke componenten en ik gebruik er al een aantal. Ook van Laravel kan ik sommige dingen gebruiken, zoals hun ORM, zonder het framework zelf te gebruiken.

symfony/messenger

De eerste kandidaat leek meteen perfect en zonder enige twijfel heb ik hem geïnstalleerd. Maar het bleek moeilijker om gebruiksvoorbeelden buiten Symfony te vinden. Hoe assembleer je een bus voor berichtoverdracht uit een klus met klassen met algemene, nietszeggende namen, en dat ook nog op Redis?

De documentatie op de officiële website was behoorlijk gedetailleerd, maar de initialisatie werd alleen voor Symfony beschreven met behulp van hun geliefde YML en andere magische methoden voor niet-Symfonisten. Ik had geen interesse in het installatieproces, vooral niet tijdens de nieuwjaarsvakantie. Maar ik moest me er toch mee bezighouden en het duurde onverwacht lang.

Proberen te begrijpen hoe het systeem moet worden geïnstantieerd op basis van de Symfony-bron is ook geen triviale taak binnen korte tijd:

Na wat rondgraven en te proberen iets met mijn handen te doen, kwam ik tot de conclusie dat ik met allerlei tijdelijke oplossingen bezig was en besloot om iets anders te proberen.

illuminate/queue

Het bleek dat deze bibliotheek stevig verbonden is met de Laravel-infrastructuur en een aantal andere afhankelijkheden, dus ik heb er niet veel tijd aan besteed: ik heb het geïnstalleerd, gekeken, de afhankelijkheden gezien en het verwijderd.

yiisoft/yii2-queue

Hieruit kon meteen worden afgeleid uit de naam dat het opnieuw een sterke koppeling heeft met Yii2. Ik heb deze bibliotheek eerder gebruikt en het was niet slecht, maar ik had er niet aan gedacht dat het volledig afhankelijk is van Yii2.

Overige

Al het andere dat ik op GitHub vond, waren onbetrouwbare, verouderde en verwaarloosde projecten zonder sterren, forks en een groot aantal commits.

Terug naar symfony/messenger, technische details

Ik moest deze bibliotheek beter begrijpen en na wat extra tijd lukte dat. Het bleek vrij eenvoudig en overzichtelijk te zijn. Voor het instantiëren van de bus heb ik een kleine fabriek gemaakt, omdat ik meerdere bussen met verschillende handlers had gepland.

Slechts een paar stappen:

• We creëren bericht handlers die gewoon callable moeten zijn
• We verpakken ze in HandlerDescriptor (een klasse uit de bibliotheek)
• Deze 'Descriptors' verpakken we in een instantie van HandlersLocator
• We voegen HandlersLocator toe aan een instantie van MessageBus
• We geven de SendersLocator een set `SenderInterface`, in mijn geval instanties van de klassen `RedisTransport`, die op een duidelijke manier geconfigureerd kunnen worden
• We voegen SendersLocator toe aan een instantie van MessageBus

MessageBus heeft een methode `->dispatch()`, die zoekt naar de overeenkomstige handlers in HandlersLocator en het bericht aan hen doorgeeft, met gebruik van de bijbehorende `SenderInterface` voor het verzenden via de bus (Redis streams).

In de configuratie van de container (in dit geval php-di) kan deze koppeling als volgt worden geconfigureerd:

        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 zien we dat we in SendersLocator voor twee verschillende berichten verschillende "transporten" hebben toegewezen, elk met hun eigen connectie naar de bijbehorende streams.

Ik heb een aparte demo-project gemaakt dat een toepassing toont met drie demonprocessen die met elkaar communiceren via zo'n bus: https://github.com/backend-university/products/tree/master/products/02-redis-streams-bus.

Maar ik zal laten zien hoe een consumer kan worden ingericht:

use AppMessagesDaemonLogMessage;
use SymfonyComponentMessengerHandlerHandlerDescriptor;
use SymfonyComponentMessengerHandlerHandlersLocator;
use SymfonyComponentMessengerMessageBus;
use SymfonyComponentMessengerMiddlewareHandleMessageMiddleware;
use SymfonyComponentMessengerMiddlewareSendMessageMiddleware;
use 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: message handled: / ' . $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();

Het gebruik van deze infrastructuur in de applicatie

Door de bus in mijn backend te implementeren, heb ik afzonderlijke stappen uit het oude synchrone commando gehaald en aparte handlers gemaakt, elk met zijn eigen taak.

De pipeline voor het toevoegen van een nieuwe website aan de database is als volgt:

En meteen daarna werd het veel eenvoudiger voor mij om nieuwe functionaliteit toe te voegen, zoals het ophalen en parseren van Rss. Aangezien dit proces ook de oorspronkelijke inhoud vereist, abonneert de handler die links naar rss ophaalt zich ook op het bericht 'Content/HtmlContent', verwerkt het en verzendt het vereiste bericht verder door zijn pipeline.

Uiteindelijk waren er verschillende daemonprocessen, elk met verbindingen alleen naar de benodigde bronnen. Bijvoorbeeld, de daemon crawlers bevat alle handlers die inhoud van internet moeten ophalen, terwijl de daemon persister de verbinding met de database onderhoudt.

Nu worden in plaats van selecties uit de database de benodigde id's na de invoer door de persister gewoon via de bus aan alle geïnteresseerde handlers doorgegeven.

Bron: habr.com