Prefácio
Meu site, que mantenho como hobby, destina-se a armazenar páginas pessoais interessantes e sites pessoais. Essa temática começou a me interessar no início da minha jornada na programação, quando eu me fascinava por encontrar grandes profissionais que escreviam sobre si mesmos, suas paixões e projetos. O hábito de descobri-los permanece até hoje: quase em cada site comercial ou não, continuo a olhar para o rodapé em busca de links para os autores.
Implementação da ideia
A primeira versão foi simplesmente uma página html no meu site pessoal, onde eu armazenava links com descrições em uma lista ul. Depois de criar cerca de 20 páginas, comecei a pensar que isso não era muito eficiente e decidi tentar automatizar o processo. No stackoverflow, percebi que muitos indicavam sites em seus perfis, então escrevi um parser em php que simplesmente percorria os perfis, começando do primeiro (os endereços no SO até hoje são desse tipo: `/users/1`), extraía links do tag correto e armazenava em SQLite.
Isso pode ser considerado a segunda versão: uma coleção de dezenas de milhares de URLs em uma tabela SQLite, que substituiu a lista estática em html. Com essa lista, fiz uma busca simples. Como havia apenas URLs, a busca era apenas por elas.
Nessa etapa, abandonei o projeto e retornei a ele após muito tempo. Nesse estágio, minha experiência de trabalho já era superior a três anos e eu sentia que poderia fazer algo mais sério. Além disso, havia um grande desejo de aprender tecnologias relativamente novas para mim.
Versão moderna
desdobrada em Docker, o banco de dados foi migrado para MongoDB, e, relativamente recentemente, o Redis foi adicionado, que inicialmente servia apenas para cache. Como base, foi utilizado um dos microframeworks PHP.
O problema
Novos sites são adicionados com um comando de console que faz o seguinte de forma síncrona:
- Baixa o conteúdo pela URL
- Define um sinalizador indicando se o HTTPS estava disponível
- Salva a entidade do site
- O HTML original e os cabeçalhos são salvos na história de 'indexação'
- Analisa o conteúdo, extrai o Title e a Description
- Os dados são salvos em uma coleção separada
Isso foi suficiente para simplesmente armazenar sites e exibi-los em uma lista:
Mas a ideia de indexar, categorizar e classificar automaticamente, mantendo tudo atualizado, se encaixava mal nessa paradigma. Mesmo a simples adição de um método web para adicionar páginas exigiu duplicação de código e bloqueios para evitar um potencial DDoS.
Na verdade, tudo pode ser feito de forma síncrona, e no método web é possível apenas armazenar a URL para que um monstruoso daemon execute todas as tarefas para as URLs da lista. Mas ainda assim, a palavra "fila" se faz necessária. E se implementarmos uma fila, podemos dividir todas as tarefas e executá-las pelo menos de forma assíncrona.
Solução
Implementar filas e criar um sistema de processamento de tarefas baseado em eventos. E há muito tempo queria testar o Redis Streams.
Uso de Redis streams no PHP
Como meu framework não pertence ao trio dos gigantes Symfony, Laravel, Yii, eu queria encontrar uma biblioteca independente. Mas, como se provou (na primeira análise) — não é possível encontrar bibliotecas sérias específicas. Tudo relacionado a filas é ou um projetinho com 3 commits de cinco anos atrás, ou está atrelado a um framework.
Ouvi falar do Symfony como fornecedor de componentes úteis separados, e alguns deles já utilizo. Também é possível usar algumas coisas do Laravel, como seu ORM, sem ter que incluir o framework em si.
symfony/messenger
O primeiro candidato me pareceu perfeito de imediato e, sem qualquer dúvida, eu o instalei. Porém, encontrar exemplos de uso fora do Symfony foi mais difícil do que pensei. Como montar um barramento para transmissão de mensagens a partir de um monte de classes com nomes universais e sem significado, ainda por cima utilizando Redis?
A documentação no site oficial era bastante detalhada, mas a inicialização foi descrita apenas para Symfony usando seu querido YML e outros métodos mágicos que não são para symfonistas. Eu não estava muito interessado no processo de instalação, especialmente durante as férias de Ano Novo. Mas tive que lidar com isso e acabei levando mais tempo do que esperava.
Tentar entender a instância do sistema com base no código-fonte do Symfony também não é uma tarefa trivial em prazos apertados:
Depois de mexer em tudo isso e tentar fazer algo manualmente, cheguei à conclusão de que estava usando algumas gambiarras e decidi tentar algo diferente.
illuminate/queue
Descobriu-se que essa biblioteca está rigidamente vinculada à infraestrutura do Laravel e a uma série de outras dependências, por isso não gastei muito tempo com ela: instalei, olhei, vi as dependências e a removi.
yiisoft/yii2-queue
Aqui já se pressupunha pelo nome que havia uma forte ligação ao Yii2. Eu usei essa biblioteca e ela era boa, mas não pensava que fosse totalmente dependente do Yii2.
Os outros
Tudo o que encontrei no GitHub eram projetos obsoletos, não confiáveis e abandonados, sem estrelas, forks e um grande número de commits.
Retorno ao symfony/messenger, detalhes técnicos
Tive que entender essa biblioteca e, depois de gastar mais algum tempo, consegui. Descobri que tudo é bastante conciso e simples. Para instanciar o barramento, fiz uma pequena fábrica, já que eu pretendia ter várias barragens com diferentes manipuladores.
Apenas alguns passos:
- Criamos manipuladores de mensagens, que devem ser simplesmente chamáveis
- Envolvemos eles em HandlerDescriptor (classe da biblioteca)
- Esses 'Descritores' são envolvidos em uma instância de HandlersLocator
- Adicionamos HandlersLocator na instância de MessageBus
- Passamos para SendersLocator um conjunto de `SenderInterface`, no meu caso instâncias das classes `RedisTransport`, que são configuradas de forma óbvia
- Adicionamos SendersLocator na instância de MessageBus
MessageBus tem o método `->dispatch()`, que procura os manipuladores correspondentes em HandlersLocator e passa a mensagem para eles, utilizando os correspondentes `SenderInterface` para enviar através do barramento (streams Redis).
Na configuração do contêiner (neste caso php-di), toda essa ligação pode ser configurada assim:
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);
},
Aqui, podemos ver que no SendersLocator atribuímos diferentes "transports" para duas mensagens diferentes, cada um com sua própria conexão para os streams correspondentes.
Eu fiz um projeto demo separado, demonstrando um aplicativo com três demônios que se comunicam entre si usando esse barramento: .
Mas vou mostrar como um consumidor pode ser estruturado:
use AppMessagesDaemonLogMessage;
use SymfonyComponentMessengerHandlerHandlerDescriptor;
use SymfonyComponentMessengerHandlerHandlersLocator;
use SymfonyComponentMessengerMessageBus;
use SymfonyComponentMessengerMiddlewareHandleMessageMiddleware;
use SymfonyComponentMessengerMiddlewareSendMessageMiddleware;
use SymfonyComponentMessengerTransportSenderSendersLocator;
require_once __DIR__ . '/../vendor/autoload.php';
/** @var PsrContainerContainerInterface $container *\/n$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();
O uso dessa infraestrutura no aplicativo
Depois de implementar o barramento no meu backend, separei etapas distintas da antiga equipe síncrona e criei manipuladores separados, cada um lidando com seu respectivo trabalho.
O pipeline para adicionar um novo site ao banco de dados ficou assim:
E logo após isso, ficou muito mais fácil adicionar novas funcionalidades, como a extração e parsing de Rss. Como esse processo também requer conteúdo original, o manipulador de extração de links para rss, assim como o WebsiteIndexHistoryPersistor, se inscreve na mensagem "Content/HtmlContent", processa-a e encaminha a mensagem necessária pelo seu próprio pipeline.
No final, resultou em vários demônios, cada um mantendo conexões apenas com os recursos necessários. Por exemplo, o demônio crawlers contém todos os manipuladores que requerem acesso à internet em busca de conteúdo, enquanto o demônio persister mantém a conexão com o banco de dados.
Agora, em vez de seleções do banco de dados, os IDs necessários após a inserção pelo persister são simplesmente transmitidos pelo barramento a todos os manipuladores interessados.
Fonte: habr.com
