Traducerea backend-ului PHP pe baza Redis streams și alegerea unei biblioteci independente de framework-uri

Prefață

Site-ul meu, pe care îl consider un hobby, este destinat stocării de pagini web interesante și site-uri personale. Această temă m-a preocupat încă de la începutul călătoriei mele în programare, când eram fascinat de găsirea unor specialiști consacrați care scriu despre ei înșiși, pasiunile lor și proiectele lor. Obiceiul de a-i descoperi a rămas și acum: aproape pe fiecare site comercial și mai puțin comercial, continui să verific subsolul în căutarea linkurilor către autori.

Implementarea ideii

Prima versiune a fost pur și simplu o pagină HTML pe site-ul meu personal, unde am adunat linkuri cu descrieri într-o listă ul. După ce am adunat vreo 20 de pagini, am început să mă gândesc că nu este foarte eficient și am decis să încerc să automatizez procesul. Pe stackoverflow am observat că mulți indicau site-uri în profilurile lor, așa că am scris un parser în PHP, care pur și simplu mergea prin profiluri, începând cu primul (adresele pe SO sunt până în ziua de astăzi de acest fel: `\/users\/1`), extrăgând linkurile din eticheta dorită și stocându-le în SQLite.

Aceasta poate fi numită a doua versiune: o colecție de zeci de mii de URL-uri într-o tabelă SQLite, care a înlocuit lista statică în HTML. Pe baza acestei liste, am realizat o căutare simplă. Din moment ce erau doar URL-uri, căutarea a fost pur și simplu pe ele.

În această etapă, am abandonat proiectul și m-am întors la el după o perioadă îndelungată. La acel moment, aveam deja peste trei ani de experiență în muncă și simțeam că pot face ceva mai serios. În plus, aveam o mare dorință de a învăța tehnologii relativ noi pentru mine.

Versiunea modernă

Proiect este dezvoltată în Docker, baza a fost migrată pe MongoDB, și, de ceva vreme, a fost adăugat Redis, care, la început, a fost folosit doar pentru caching. Ca bază, se folosește unul dintre microframework-urile PHP.

Problema

Noile site-uri sunt adăugate cu un command line care face sincron pe următoarele:

• Descarcă conținutul de pe URL
• Setează un flag care indică dacă HTTPS-ul a fost disponibil
• Salvează entitatea site-ului web
• Salvează HTML-ul original și anteturile în istoricul "indexării"
• Parsat conținutul, extrăgând Title și Description
• Datele sunt salvate într-o colecție separată

Asta a fost suficient pentru a stoca site-urile și a le afișa într-o listă:

Dar ideea de a indexa, categoriza și a clasifica totul automat, menținându-l actual, se încadra slab în această paradigmă. Chiar și adăugarea unei metode web pentru a insera pagini a necesitat duplicarea codului și blocaje pentru a evita un posibil DDoS.

În general, desigur, totul poate fi făcut și sincronic, iar în metoda web se poate realiza pur și simplu salvarea URL-ului pentru ca demonul monstruos să îndeplinească toate sarcinile pentru URL-urile din listă. Dar chiar și aici se impune cuvântul „coadă”. Dacă se implementa o coadă, toate sarcinile ar putea fi împărțite și executate cel puțin asincron.

Soluție

Implementarea coadelor și crearea unui sistem de procesare a sarcinilor bazat pe evenimente. Și de mult timp mi-am dorit să încerc Redis Streams.

Folosirea Redis streams în PHP

Deoarece cadrul meu nu face parte din trio-ul gigantilor Symfony, Laravel, Yii, mi-ar plăcea să găsesc o bibliotecă independentă. Dar, așa cum s-a dovedit (la prima examinare) - a găsi biblioteci separate serioase este imposibil. Tot ce este legat de cozi fie este un proiect de 3 commit-uri de acum cinci ani, fie este legat de cadrul respectiv.

Am auzit de Symfony ca furnizor de componente utile separate, iar unele dintre ele le folosesc deja. De asemenea, de la Laravel pot fi folosite unele lucruri, cum ar fi ORM-ul lor, fără a avea cadrul în sine.

symfony/messenger

Primul candidat a părut imediat perfect și fără nicio ezitare l-am instalat. Însă găsirea exemplelor de utilizare în afara Symfony s-a dovedit a fi mai complicată. Cum să construiești dintr-o mulțime de clase cu nume universale, care nu spun nimic, un bus pentru transmiterea mesajelor, și încă pe Redis?

Documentația de pe site-ul oficial a fost destul de detaliată, dar inițializarea a fost descrisă doar pentru Symfony folosind YML-ul lor preferat și alte metode magice pentru cei care nu sunt simfoniani. Nu am avut interes în procesul de instalare, mai ales în vacanțele de Anul Nou. Dar a trebuit să mă ocup de asta și a durat mai mult decât m-am așteptat.

Încercarea de a înțelege sistemul de instanțiere pe baza surselor Symfony este, de asemenea, o sarcină nu foarte trivială în termenele strânse:

După ce m-am scufundat în toate acestea și am încercat să fac ceva manual, am ajuns la concluzia că mă ocup de niște soluții temporare și am decis să încerc ceva diferit.

illuminate/queue

Se pare că această bibliotecă este strâns legată de infrastructura Laravel și de multe alte dependențe, așa că nu am petrecut prea mult timp pe ea: am instalat-o, am verificat, am observat dependențele și am șters-o.

yiisoft/yii2-queue

Aici se presupunea imediat, din denumire, că există o legătură strictă cu Yii2. Am folosit această bibliotecă și a fost destul de bună, dar nu m-am gândit că depinde complet de Yii2.

Altele

Tot ce am găsit pe GitHub era proiecte nesigure, învechite și abandonate, fără stele, fork-uri sau un număr mare de commit-uri.

Întoarcere la symfony/messenger, detalii tehnice

A trebuit să mă familiarizez cu această bibliotecă și, după ce am petrecut ceva timp, am reușit. S-a dovedit că totul este destul de concis și simplu. Pentru instanțierea benzilor, am creat o mică fabrică, deoarece intenționam să am mai multe benzi cu diferiți procesatori.

Doar câțiva pași:

• Creăm procesoare de mesaje, care ar trebui să fie pur și simplu callable
• Le înfășurăm în HandlerDescriptor (o clasă din bibliotecă)
• Aceste «Descriptor» le înfășurăm într-o instanță HandlersLocator
• Adăugăm HandlersLocator în instanța MessageBus
• Transmitem în SendersLocator setul `SenderInterface`, în cazul meu instanțele claselor `RedisTransport`, care se configurează în mod evident
• Adăugăm SendersLocator în instanța MessageBus

MessageBus are metoda `->dispatch()`, care caută procesoare corespunzătoare în HandlersLocator și transmite mesajul lor, folosind polițele corespunzătoare `SenderInterface` pentru a trimite prin bandă (fluxuri Redis).

În configurația containerului (în acest caz php-di), toată această legătură poate fi configurată astfel:

        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);
        },

Aici se poate observa că în SendersLocator pentru două mesaje diferite am atribuit un „transport” diferit, fiecare având propria conexiune la fluxurile corespunzătoare.

Am creat un proiect demo separat, demonstrând o aplicație din trei demoni care comunică între ei folosind un astfel de bus: https://github.com/backend-university/products/tree/master/products/02-redis-streams-bus.

Dar voi arăta cum ar putea fi configurat consumatorul:

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();

Utilizarea acestei infrastructuri în aplicație

După ce am implementat autobuzul în backend-ul meu, am desprins etapele din comanda veche sincronă și am creat handleri separați, fiecare ocupându-se de propriile sale sarcini.

Pipeline-ul pentru adăugarea unui nou site în baza de date a fost astfel:

Și imediat după aceea, mi-a fost mult mai ușor să adaug noi funcționalități, de exemplu, extragerea și parsarea RSS-ului. Deoarece acest proces necesită și conținutul sursă, handler-ul care extrage linkuri pentru RSS, la fel ca și WebsiteIndexHistoryPersistor, se abonează la mesajul „Content/HtmlContent”, îl procesează și trimite mesajul dorit prin pipeline-ul său mai departe.

În cele din urmă, am obținut câteva demoni, fiecare dintre ei menținând conexiuni doar cu resursele necesare. De exemplu, demonul crawlers conține toate handlerii care necesită acces la internet pentru conținut, iar demonul persister menține conexiunea la baza de date.

Acum, în loc de select-uri din baza de date, ID-urile necesare după inserarea de către persister sunt pur și simplu transmise prin autobuz tuturor handlerilor interesați.

Sursa: habr.com