Превод на PHP бекенда на Redis streams и избор на независима от фреймворците библиотека

Предговор

Моя страница, на которую я смотрю как на хобби, предназначена за съхранение интересни домашни страници и лични сайтове. Тази тема започна да ме интересува още в началото на моето пътуване в програмирането, когато бях впечатлен от откритията на големи професионалисти, които пишат за себе си, своите увлечения и проекти. Навикът да ги откривам остава и сега: почти на всеки търговски и не толкова сайт, продължавам да поглеждам в долната част в търсене на връзки към авторите.

Реализация на идеята

Първата версия беше просто html-страница на моя персонален сайт, където събирах линкове с надписи в ul-списък. След като написах около 20 страници, започнах да мисля, че това не е особено ефективно и реших да опитам да автоматизирам процеса. В stackoverflow забелязвах, че много хора посочват сайтове в профилите си, затова написах парсер на php, който просто минаваше през профилите, започвайки от първия (адресите на SO и до днес изглеждат така: `/users/1`), извличаше линкове от необходимия таг и ги складираше в SQLite.

Това може да се нарече втората версия: колекция от десетки хиляди урла в SQLite таблица, която замени статичния списък в html. По този списък направих прост търсене. Тъй като имаше само урли, търсенето беше просто по тях.

На този етап оставих проекта и се върнах към него след дълго време. На този етап опитът ми беше вече над три години и чувствах, че мога да направя нещо по-сериозно. Освен това имаше голямо желание да усвоявам за мен относително нови технологии.

Съвременна версия

Проект развита в docker, базата е преместена на mongoDb, а сравнително от скоро е добавен redis, който първоначално беше просто за кеширане. Като основа се използва един от микро-фреймуърците на PHP.

Проблема

Новите сайтове се добавят с командата в конзолата, която синхронно извършва следното:

• Свали съдържанието по URL
• Поставя флаг, дали е бил достъпен HTTPS
• Съхранява съществото на уеб-сайта
• Запазва оригиналния HTML и заглавките в историята на "индексирането"
• Парсира съдържанието, извлича Title и Description
• Данните се съхраняват в отделна колекция

Това беше достатъчно, за да съхранявам сайтове и да ги показвам в списък:

Но идеята да индексираме всичко автоматично, да го категоризираме и да го класираме, като поддържаме всичко актуално, не се вписва в тази парадигма. Дори простото добавяне на уеб метод за добавяне на страници изискваше дублиране на код и заключвания, за да се избегне потенциален DDoS.

Разбира се, всичко може да се прави и синхронно, а в уеб метода да се извърши просто запазване на URL, за да може монструозният демон да изпълнява всички задачи за URL от списъка. Но все пак тук се набива на ум думата „опашка“. А ако внедрим опашка, можем да разделим всички задачи и да ги изпълняваме поне асинхронно.

Решение

Да внедрим опашки и да направим система за обработка на всички задачи, управлявана от събития. И отдавна искам да опитам Redis Streams.

Използване на Redis streams в PHP

Тъй като фреймворкът ми не е от трите големи, Symfony, Laravel, Yii, исках да намеря независима библиотека. Но, както се оказа (при първоначалното разглеждане), невъзможно е да се намерят сериозни отделни библиотеки. Всичко, свързано с опашките, или е проект от три комита, направен преди пет години, или е свързано с фреймворка.

Чух много за Symfony като доставчик на полезни компоненти, а и някои от тях вече използвам. Освен това от Laravel също може да се използва нещо, например тяхната ORM, без присъствието на самия фреймворк.

symfony/messenger

Първият кандидат веднага ми се стори идеален и го инсталирах без колебание. Но намирането на примери за употреба извън Symfony се оказа по-трудно. Как да сглобя от куп класове с универсални, нищо не означаващи имена, шина за предаване на съобщения, и то с Redis?

Документацията на официалния сайт беше доста подробна, но инициализацията беше описана само за Symfony с помощта на тяхната любима YML и други магически методи за несимфонисти. Нямаше интерес в самия процес на инсталиране, особено по време на новогодишните празници. Но се наложи да се занимавам с това и неочаквано дълго.

Опитът да разбера как да инстанцирам системата по изходния код на Symfony също не е тривиална задача за кратки срокове:

След като се поразрових в това и се опитах да направя нещо на ръка, стигнах до извода, че се занимавам с някакви скелети и реших да опитам нещо друго.

illuminate/queue

Оказа се, че тази библиотека е дълбоко свързана с инфраструктурата на Laravel и много други зависимости, затова не отделих много време за нея: инсталирах я, погледнах, видях зависимостите и я премахнах.

yiisoft/yii2-queue

Тук веднага се подразбираше от името, че има строг ангажимент към Yii2. Потребявал съм тази библиотека и беше добра, но не предполагах, че тя е напълно зависима от Yii2.

Останалите

Всичко друго, което намерих в GitHub, бяха ненадеждни, остарели и изоставени проекти без звезди, фоксове и множество комити.

Връщане към symfony/messenger, технически подробности

Трябваше да се запозная с тази библиотека и, след като отделих още известно време, успях. Оказа се, че всичко е относително лаконично и просто. За инстанциране на шината направих малка фабрика, тъй като предполагам, че ще имам няколко шини с различни обработчици.

Само няколко стъпки:

• Създаваме обработчици на съобщения, които трябва да бъдат просто callable
• Обгръщаме ги в HandlerDescriptor (клас от библиотеката)
• Тези "Дескриптори" обгръщаме в инстанция на HandlersLocator
• Добавяме HandlersLocator в инстанция на MessageBus
• Предаваме на SendersLocator набор `SenderInterface`, в моя случай инстанции на класове `RedisTransport`, които се конфигурират очевидно
• Добавяме SendersLocator в инстанция на MessageBus

MessageBus има метод `->dispatch()`, който търси съответните обработчици в HandlersLocator и предава съобщението им, използвайки съответните `SenderInterface` за изпращане чрез шината (Redis streams).

В конфигурацията на контейнера (в този случай php-di) цялата тази връзка може да бъде конфигурирана така:

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

Тук се вижда, че в SendersLocator за два различни съобщения сме задавали различен „транспорт“, всеки от които има своето свързване към съответните стриймове.

Създадох отделен демонстрационен проект, демонстриращ приложение от три демона, които комуникират помежду си чрез такава шина: https://github.com/backend-university/products/tree/master/products/02-redis-streams-bus.

Но ще покажа как може да бъде устроен консумер:

използвайте AppMessagesDaemonLogMessage;
използвайте SymfonyComponentMessengerHandlerHandlerDescriptor;
използвайте SymfonyComponentMessengerHandlerHandlersLocator;
използвайте SymfonyComponentMessengerMessageBus;
използвайте SymfonyComponentMessengerMiddlewareHandleMessageMiddleware;
използвайте SymfonyComponentMessengerMiddlewareSendMessageMiddleware;
използвайте 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();

Използване на тази инфраструктура в приложението

След като реализирах шината в задния си край, изведох отделни стъпки от старата синхронна команда и направих отделни хендлери, всеки от които се занимава със своето.

Процесът на добавяне на нов сайт в базата данни стана такъв:

И точно след това стана много по-лесно да добавям нови функционалности, например извличане и парсинг на RSS. Тъй като този процес също изисква оригинално съдържание, хендлерът, извличащ линкове към rss, както и WebsiteIndexHistoryPersistor, се абонира на съобщението «Content/HtmlContent», обработва го и предава нужното съобщение по своя пайплайн напред.

В крайна сметка се получиха няколко демона, всеки от които поддържа връзки само към нужните ресурси. Например демон crawlers съдържа всички обработчици, които изискват достъп до интернет за съдържание, а демон persister поддържа връзка с базата данни.

Сега вместо селекти от базата данни нужните id след вмъкването просто се предават през шината на всички заинтересовани обработчици.

Източник: habr.com