Przeniesienie backendu PHP na strumienie Redis i wybór biblioteki niezależnej od frameworków

Wstęp

Moja strona, którą zajmuję się jako hobby, jest przeznaczona do zbierania interesujących stron domowych i osobistych witryn. Temat ten zafascynował mnie na samym początku mojej drogi w programowaniu, kiedy to podziwiałem wybitnych profesjonalistów, którzy piszą o sobie, swoich pasjach i projektach. Przypadłość odkrywania ich dla siebie pozostała do dziś: prawie na każdej stronie, zarówno komercyjnej, jak i nieco mniej, wciąż zaglądam do stopki, szukając linków do autorów.

Realizacja pomysłu

Pierwsza wersja była po prostu stroną HTML na mojej osobistej stronie, gdzie zbierałem linki z opisami w liście ul. Po jakimś czasie uzbierało się około 20 stron, zaczynałem myśleć, że to nie jest zbyt efektywne, więc postanowiłem spróbować zautomatyzować ten proces. Na Stack Overflow zauważyłem, że wielu użytkowników podaje linki do swoich stron w swoich profilach, dlatego napisałem parser w PHP, który po prostu przechodził przez profile, zaczynając od pierwszego (adresy na SO są do dziś w takim formacie: `\/users\/1`), wyciągał linki z odpowiedniego tagu i zapisywał je w SQLite.

Można to nazwać drugą wersją: kolekcja z kilku tysięcy URL-i w tabeli SQLite, która zastąpiła statyczną listę w HTML. Na podstawie tej listy stworzyłem prostą wyszukiwarkę. Ponieważ były tylko URL-e, więc wyszukiwanie polegało tylko na nich.

Na tym etapie porzuciłem projekt i wróciłem do niego po dłuższym czasie. W tym momencie doświadczenie mojej pracy wynosiło już ponad trzy lata i czułem, że mogę zrobić coś bardziej zaawansowanego. Ponadto bardzo chciałem zgłębiać stosunkowo nowe dla mnie technologie.

Współczesna wersja

Projekt wdrożona w Dockerze, baza przeniesiona na MongoDB, a od niedawna dodano Redis, który najpierw służył tylko do cache'owania. Jako bazę wykorzystano jeden z mikroframeworków PHP.

Problem

Nowe strony są dodawane za pomocą komendy konsolowej, która jednocześnie wykonuje następujące czynności:

• Pobiera zawartość z URL
• Ustala flagę, czy HTTPS był dostępny
• Zapisuje encję witryny
• Zachowuje oryginalny HTML i nagłówki w historii 'indeksowania'
• Parsuje zawartość, wyciąga tytuł i opis
• Dane zapisuje w oddzielnej kolekcji

To wystarczyło, aby po prostu przechowywać strony i wyświetlać je na liście:

Jednak koncepcja automatycznego indeksowania, kategoryzowania i rangowania, utrzymując wszystko w aktualności, słabo pasowała do tej paradygmy. Nawet proste dodanie metody webowej do dodawania stron wymagało duplikacji kodu i blokad, aby uniknąć potencjalnego DDoS.

W rzeczywistości wszystko można robić równolegle, a w metodzie webowej po prostu zapisać URL, aby potworny demon wykonywał wszystkie zadania dla URL-i z listy. Ale nawet tutaj kusi słowo „kolejka”. A jeśli wdrożymy kolejkę, to można podzielić wszystkie zadania i wykonywać je przynajmniej asynchronicznie.

Rozwiązanie

Wdrożyć kolejki i stworzyć system przetwarzania wszystkich zadań oparty na zdarzeniach. I od dłuższego czasu chciałem spróbować Redis Streams.

Użycie Redis Streams w PHP

Ponieważ mój framework nie należy do wielkiej trójki: Symfony, Laravel, Yii, chciałem znaleźć niezależną bibliotekę. Ale jak się okazało (przy pierwszym przeglądzie) - nie można znaleźć żadnych poważnych bibliotek. Wszystko związane z kolejkami to albo projekt z 3 commitami sprzed pięciu lat, albo jest związane z frameworkiem.

Słyszałem o Symfony jako dostawcy użytecznych komponentów, zresztą niektóre już używam. Również z Laravela można wykorzystać coś, na przykład ich ORM, bez obecności samego frameworka.

symfony/messenger

Pierwszy kandydat od razu wydawał się idealny i bez żadnych wątpliwości go zainstalowałem. Ale znalezienie przykładów użycia poza Symfony okazało się trudniejsze. Jak zebrać z mnóstwa klas o uniwersalnych, nic nie mówiących nazwach, szynę do przesyłania wiadomości, i to na Redis?

Dokumentacja na oficjalnej stronie była dość szczegółowa, ale inicjalizacja opisana była tylko dla Symfony przy użyciu ich ulubionego YML i innych magicznych metod dla nie-Symfonistów. Nie miałem zainteresowania samym procesem instalacji, zwłaszcza w okresie noworocznym. Ale musiałem się tym zająć i trwało to zaskakująco długo.

Próba zrozumienia instancjonowania systemu na podstawie źródeł Symfony to także nie jest trywialne zadanie przy napiętych terminach:

Badając to wszystko i próbując coś zrobić ręcznie, doszedłem do wniosku, że zajmuję się jakimiś kontuzjami i postanowiłem spróbować czegoś innego.

illuminate/queue

Okazało się, że ta biblioteka jest ściśle związana z infrastrukturą Laravel i mnóstwem innych zależności, więc nie spędziłem na niej dużo czasu: zainstalowałem, spojrzałem, zobaczyłem zależności i usunąłem.

yiisoft/yii2-queue

Tutaj od razu zakładano z nazwy, znowu mocne powiązanie z Yii2. Musiałem korzystać z tej biblioteki i była niezła, ale nie pomyślałem, że całkowicie zależy od Yii2.

Inne

Wszystko inne, co znalazłem na githubie — to niezawodne, przestarzałe i porzucone projektiki bez gwiazdek, forków i dużej liczby commitów.

Powrót do symfony/messenger, szczegóły techniczne

Musiałem zrozumieć tę bibliotekę i, spędzając jeszcze trochę czasu, udało mi się. Okazało się, że wszystko jest dość zwięzłe i proste. Aby zainstancjować busa, zrobiłem małą fabrykę, ponieważ planowałem mieć kilka busów z różnymi handlerami.

Tylko kilka kroków:

• Tworzymy obsługiwacze wiadomości, które powinny być po prostu callable
• Opakowujemy je w HandlerDescriptor (klasa z biblioteki)
• Te „Deskryptory” opakowujemy w instancję HandlersLocator
• Dodajemy HandlersLocator do instancji MessageBus
• Przekazujemy do SendersLocator zestaw `SenderInterface`, w moim przypadku instancje klas `RedisTransport`, które konfigurowane są w oczywisty sposób
• Dodajemy SendersLocator do instancji MessageBus

MessageBus ma metodę `->dispatch()`, która szuka odpowiednich handlerów w HandlersLocator i przekazuje wiadomość do nich, korzystając z odpowiednich `SenderInterface` do wysyłania przez bus (Redis streams).

W konfiguracji kontenera (w tym przypadku php-di) całe to powiązanie może być skonfigurowane tak:

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

W SendersLocator przypisaliśmy różny „transport” dla dwóch różnych wiadomości, z każdym z nich mającym własne połączenie do odpowiednich strumieni.

Stworzyłem oddzielny projekt demo, pokazujący aplikację składającą się z trzech demonów, które komunikują się ze sobą za pomocą takiej szyny: https://github.com/backend-university/products/tree/master/products/02-redis-streams-bus.

Pokażę, jak może być zorganizowany konsument:

użyj AppMessagesDaemonLogMessage;
użyj SymfonyComponentMessengerHandlerHandlerDescriptor;
użyj SymfonyComponentMessengerHandlerHandlersLocator;
użyj SymfonyComponentMessengerMessageBus;
użyj SymfonyComponentMessengerMiddlewareHandleMessageMiddleware;
użyj SymfonyComponentMessengerMiddlewareSendMessageMiddleware;
użyj 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();

Wykorzystanie tej infrastruktury w aplikacji

Realizując szynę w swoim backendzie, wyodrębniłem oddzielne etapy z starego synchronego polecenia i stworzyłem oddzielne handlery, z których każdy zajmuje się swoimi zadaniami.

Pipeline dodawania nowej strony do bazy danych wygląda następująco:

I tuż po tym było mi znacznie łatwiej dodawać nową funkcjonalność, na przykład ekstrakcję i parsowanie Rss. Ponieważ ten proces również wymaga źródłowej treści, handler-ekstraktor linków do rss, podobnie jak WebsiteIndexHistoryPersistor, subskrybuje wiadomość „Content/HtmlContent”, przetwarza ją i przekazuje odpowiednią wiadomość w swoim pipeline dalej.

W ostateczności powstało kilka demonów, z których każdy utrzymuje połączenia tylko z potrzebnymi zasobami. Na przykład demon crawlers zawiera wszystkie handlery, które wymagają dostępu do internetu po treść, a demon persister utrzymuje połączenie z bazą danych.

Teraz zamiast selekcji z bazy danych, potrzebne id po wstawieniu przez persistera są po prostu przekazywane przez szynę do wszystkich zainteresowanych handlerów.

Źródło: habr.com