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

Предговор

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

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

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

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

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

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

Проект развита в Docker, базата е преместена на MongoDB, и от относително скоро е добавен Redis, който първоначално беше само за кеширане. За основа се използва един от микрофреймворците на PHP.

Проблема

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

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

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

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

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

Решение

Да внедрим опашки и да направим събитийно-ориентирана система за обработка на всички задачи. И отдавна исках да опитам Redis Streams.

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

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

Чух много за 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 => функция (ContainerInterface $c) {
            return new RedisTransport(
                $c->get(CONTAINER_REDIS_STREAM_CONNECTION_SECRET),
                $c->get(CONTAINER_SERIALIZER))
            ;
        },
        CONTAINER_REDIS_TRANSPORT_LOG => функция (ContainerInterface $c) {
            return new RedisTransport(
                $c->get(CONTAINER_REDIS_STREAM_CONNECTION_LOG),
                $c->get(CONTAINER_SERIALIZER))
            ;
        },
        CONTAINER_REDIS_STREAM_RECEIVER_SECRET => функция (ContainerInterface $c) {
            return new RedisReceiver(
                $c->get(CONTAINER_REDIS_STREAM_CONNECTION_SECRET),
                $c->get(CONTAINER_SERIALIZER)
            );
        },
        CONTAINER_REDIS_STREAM_RECEIVER_LOG => функция (ContainerInterface $c) {
            return new RedisReceiver(
                $c->get(CONTAINER_REDIS_STREAM_CONNECTION_LOG),
                $c->get(CONTAINER_SERIALIZER)
            );
        },
        CONTAINER_REDIS_STREAM_BUS => функция (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 => функция (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 => функция (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;

изисквайте веднъж __DIR__ . '/.. /vendor/autoload.php';
/** @var PsrContainerContainerInterface $container * /
$container = изисквайте веднъж('config/container.php');

$handlers = [
    DaemonLogMessage::class => [
        нов HandlerDescriptor(
            функция (DaemonLogMessage $m) {
                error_log('DaemonLogHandler: съобщение обработено: / ' . $m->getMessage());
            },
            ['from_transport' => CONTAINER_REDIS_TRANSPORT_LOG]
        )
    ],
];
$middleware = [];
$middleware[] = нов HandleMessageMiddleware(new HandlersLocator($handlers));
$sendersLocator = нов SendersLocator(['*' => [CONTAINER_REDIS_TRANSPORT_LOG]], $container);
$middleware[] = нов SendMessageMiddleware($sendersLocator);

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

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

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

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

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

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

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

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