Symfony
Táto stránka pridá Hermes do nového projektu v Symfony 7.4, vedľa Messengera alebo namiesto neho. Controller vloží user.registered a bin/console hermes:work ju spracuje. Najprv si prečítajte Integráciu do frameworkov, kde sú časti spoločné pre všetky frameworky.
1. Inštalácia
composer create-project symfony/skeleton:"7.4.*" app
cd app
composer require tomaj/hermes:dev-master symfony/monolog-bundle
composer require predis/predis # only for the predis driver
composer require php-amqplib/php-amqplib # only for the rabbitmq driver
composer require aws/aws-sdk-php # only for the sqs driverKým Hermes 6.0 nemá tag, vyžadujte tomaj/hermes:dev-master ako vyššie; po vydaní prejdite na ^6.0. Bez monolog-bundle je autowirovaný LoggerInterface minimálny logger Symfony, ktorý zapisuje na stderr.
2. Konfigurácia služieb
Celý config/services.yaml je nižšie. Parameter hermes zbiera premenné prostredia, DriverFactory z neho vytvorí driver a obe rozhrania sú aliasy tried Hermesu. Handler sa registruje položkou calls na dispatcheri.
# yaml-language-server: $schema=../vendor/symfony/dependency-injection/Loader/schema/services.schema.json
parameters:
# Defaults for the environment variables; override them in .env.local or the real environment.
env(HERMES_DRIVER): redis # redis (ext-redis), predis, rabbitmq or sqs
env(HERMES_QUEUE): hermes
env(HERMES_REDIS_HOST): 127.0.0.1
env(HERMES_REDIS_PORT): '6379'
env(HERMES_RABBITMQ_HOST): 127.0.0.1
env(HERMES_RABBITMQ_PORT): '5672'
env(HERMES_RABBITMQ_USER): guest
env(HERMES_RABBITMQ_PASSWORD): guest
env(HERMES_RABBITMQ_VHOST): /
env(HERMES_SQS_REGION): eu-central-1
env(HERMES_SQS_ENDPOINT): ''
env(AWS_ACCESS_KEY_ID): ''
env(AWS_SECRET_ACCESS_KEY): ''
hermes:
driver: '%env(HERMES_DRIVER)%'
queue: '%env(HERMES_QUEUE)%'
max_items: 1000
redis:
host: '%env(HERMES_REDIS_HOST)%'
port: '%env(int:HERMES_REDIS_PORT)%'
rabbitmq:
host: '%env(HERMES_RABBITMQ_HOST)%'
port: '%env(int:HERMES_RABBITMQ_PORT)%'
user: '%env(HERMES_RABBITMQ_USER)%'
password: '%env(HERMES_RABBITMQ_PASSWORD)%'
vhost: '%env(HERMES_RABBITMQ_VHOST)%'
sqs:
region: '%env(HERMES_SQS_REGION)%'
endpoint: '%env(HERMES_SQS_ENDPOINT)%'
key: '%env(AWS_ACCESS_KEY_ID)%'
secret: '%env(AWS_SECRET_ACCESS_KEY)%'
services:
_defaults:
autowire: true
autoconfigure: true
App\:
resource: '../src/'
# One driver for the web process and the worker.
Tomaj\Hermes\Driver\DriverInterface:
factory: ['App\Hermes\DriverFactory', 'create']
arguments: ['%hermes%']
Tomaj\Hermes\Shutdown\ShutdownInterface:
class: Tomaj\Hermes\Shutdown\SharedFileShutdown
arguments: ['%kernel.project_dir%/var/hermes-shutdown']
# Producers type-hint EmitterInterface; the logger is autowired (Monolog).
Tomaj\Hermes\EmitterInterface:
class: Tomaj\Hermes\Emitter
# The worker command type-hints DispatcherInterface.
Tomaj\Hermes\DispatcherInterface:
class: Tomaj\Hermes\Dispatcher
calls:
- registerHandler: ['user.registered', '@App\Hermes\UserRegisteredHandler']V .env.local nastavte HERMES_DRIVER na redis, predis, rabbitmq alebo sqs a každej aplikácii dajte vlastnú HERMES_QUEUE.
<?php
declare(strict_types=1);
namespace App\Hermes;
use Aws\Sqs\SqsClient;
use InvalidArgumentException;
use PhpAmqpLib\Connection\AMQPLazyConnection;
use Predis\Client;
use Redis;
use Tomaj\Hermes\Driver\AmazonSqsDriver;
use Tomaj\Hermes\Driver\DriverInterface;
use Tomaj\Hermes\Driver\LazyRabbitMqDriver;
use Tomaj\Hermes\Driver\PredisSetDriver;
use Tomaj\Hermes\Driver\RedisSetDriver;
/**
* Builds the Hermes driver from plain configuration, so the web process and the worker
* always use the same driver, queue name and connection.
*/
final class DriverFactory
{
/**
* @param array{
* driver: string,
* queue: string,
* max_items: int,
* redis: array{host: string, port: int},
* rabbitmq: array{host: string, port: int, user: string, password: string, vhost: string},
* sqs: array{region: string, endpoint: ?string, key: string, secret: string},
* } $config
*/
public static function create(array $config): DriverInterface
{
$queue = $config['queue'];
$driver = match ($config['driver']) {
'redis' => new RedisSetDriver(self::redis($config['redis']), $queue),
'predis' => new PredisSetDriver(new Client(['host' => $config['redis']['host'], 'port' => $config['redis']['port']]), $queue),
'rabbitmq' => new LazyRabbitMqDriver(
new AMQPLazyConnection(
$config['rabbitmq']['host'],
$config['rabbitmq']['port'],
$config['rabbitmq']['user'],
$config['rabbitmq']['password'],
$config['rabbitmq']['vhost'],
),
$queue,
[],
1, // seconds to sleep when the queue is empty
),
'sqs' => new AmazonSqsDriver(self::sqs($config['sqs']), $queue),
default => throw new InvalidArgumentException("Unknown Hermes driver \"{$config['driver']}\""),
};
// Start a fresh worker process after this many messages.
$driver->setMaxProcessItems($config['max_items']);
return $driver;
}
/**
* @param array{host: string, port: int} $config
*/
private static function redis(array $config): Redis
{
$redis = new Redis();
$redis->connect($config['host'], $config['port']);
return $redis;
}
/**
* @param array{region: string, endpoint: ?string, key: string, secret: string} $config
*/
private static function sqs(array $config): SqsClient
{
$options = [
'region' => $config['region'],
'version' => 'latest',
'credentials' => ['key' => $config['key'], 'secret' => $config['secret']],
];
if ($config['endpoint'] !== null && $config['endpoint'] !== '') {
$options['endpoint'] = $config['endpoint']; // a local SQS such as moto or ElasticMQ
}
return new SqsClient($options);
}
}Monolog sa autowiruje ako LoggerInterface, takže riadky Hermesu skončia vo var/log/dev.log (alebo vo vašich produkčných handleroch). Pozrite Logovanie.
3. Vloženie správy z controllera
<?php
declare(strict_types=1);
namespace App\Controller;
use Symfony\Component\HttpFoundation\JsonResponse;
use Symfony\Component\HttpFoundation\Request;
use Symfony\Component\Routing\Attribute\Route;
use Tomaj\Hermes\EmitterInterface;
use Tomaj\Hermes\Message;
final class UserController
{
public function __construct(private readonly EmitterInterface $emitter) {}
#[Route('/users', name: 'user_register', methods: ['POST'])]
public function register(Request $request): JsonResponse
{
$email = $request->getPayload()->getString('email');
if (filter_var($email, FILTER_VALIDATE_EMAIL) === false) {
return new JsonResponse(['error' => 'A valid email is required.'], 422);
}
// Save the user here, then hand the slow work to the worker.
$this->emitter->emit(new Message('user.registered', ['email' => $email]));
return new JsonResponse(['status' => 'registered'], 201);
}
}Príkaz producenta, ktorým to nižšie vyskúšate:
<?php
declare(strict_types=1);
namespace App\Command;
use Symfony\Component\Console\Attribute\AsCommand;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputArgument;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Output\OutputInterface;
use Tomaj\Hermes\EmitterInterface;
use Tomaj\Hermes\Message;
#[AsCommand(name: 'app:register-user', description: 'Emit user.registered for one email address')]
final class RegisterUserCommand extends Command
{
public function __construct(private readonly EmitterInterface $emitter)
{
parent::__construct();
}
#[\Override]
protected function configure(): void
{
$this->addArgument('email', InputArgument::REQUIRED);
}
#[\Override]
protected function execute(InputInterface $input, OutputInterface $output): int
{
$email = (string) $input->getArgument('email');
$this->emitter->emit(new Message('user.registered', ['email' => $email]));
$output->writeln("Queued the welcome email for {$email}.");
return Command::SUCCESS;
}
}4. Napíšte handler
<?php
declare(strict_types=1);
namespace App\Hermes;
use Psr\Log\LoggerInterface;
use Tomaj\Hermes\Handler\RetryableHandlerInterface;
use Tomaj\Hermes\Handler\RetryTrait;
use Tomaj\Hermes\MessageInterface;
final class UserRegisteredHandler implements RetryableHandlerInterface
{
use RetryTrait;
public function __construct(private readonly LoggerInterface $logger) {}
#[\Override]
public function handle(MessageInterface $message): bool
{
$email = $message->getPayload()['email'] ?? null;
if (!is_string($email)) {
return false; // a broken message stays broken, do not retry it
}
// Send the welcome email here. Throw an exception to retry later.
$this->logger->info("Welcome email sent to {$email}", ['message_id' => $message->getId()]);
return true;
}
}Je to autowirovaná služba, takže jej konštruktor si môže vypýtať čokoľvek z kontajnera. Výnimkou spustíte opakovanie, vrátením false to vzdáte; pozrite Opakovanie.
5. Spustite workera
<?php
declare(strict_types=1);
namespace App\Command;
use Symfony\Component\Console\Attribute\AsCommand;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Input\InputOption;
use Symfony\Component\Console\Output\OutputInterface;
use Tomaj\Hermes\DispatcherInterface;
#[AsCommand(name: 'hermes:work', description: 'Run the Hermes worker until it is told to stop')]
final class HermesWorkCommand extends Command
{
public function __construct(private readonly DispatcherInterface $dispatcher)
{
parent::__construct();
}
#[\Override]
protected function configure(): void
{
$this->addOption('priority', null, InputOption::VALUE_REQUIRED | InputOption::VALUE_IS_ARRAY, 'Handle only these priorities');
}
#[\Override]
protected function execute(InputInterface $input, OutputInterface $output): int
{
/** @var list<string> $priorities */
$priorities = $input->getOption('priority');
// Blocks until shutdown (hermes:restart) or the max items limit.
$this->dispatcher->handle(array_map(intval(...), $priorities));
return Command::SUCCESS;
}
}bin/console hermes:work # terminal 1
bin/console app:register-user jane@example.com # terminal 2
grep 'Welcome email sent' var/log/dev.log--priority=200 obmedzí workera na jednu prioritnú frontu; pozrite Priorita.
6. Reštart workerov pri nasadení
<?php
declare(strict_types=1);
namespace App\Command;
use Symfony\Component\Console\Attribute\AsCommand;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Output\OutputInterface;
use Tomaj\Hermes\Shutdown\ShutdownInterface;
#[AsCommand(name: 'hermes:restart', description: 'Stop every running Hermes worker after its current message')]
final class HermesRestartCommand extends Command
{
public function __construct(private readonly ShutdownInterface $shutdown)
{
parent::__construct();
}
#[\Override]
protected function execute(InputInterface $input, OutputInterface $output): int
{
$this->shutdown->shutdown();
$output->writeln('Workers will stop after their current message. Supervisor or systemd starts new ones.');
return Command::SUCCESS;
}
}Po cache:clear spustite bin/console hermes:restart. Workery sa zastavia po aktuálnej správe a Supervisor spustí nové:
; /etc/supervisor/conf.d/hermes-worker.conf
; Two workers. hermes:restart stops them after their current message, and Supervisor starts new ones.
; With ext-pcntl, SIGTERM (stopsignal) also lets a worker finish its message; stopwaitsecs is the limit before SIGKILL.
[program:hermes-worker]
command=php bin/console hermes:work
directory=/var/www/app
process_name=%(program_name)s_%(process_num)02d
numprocs=2
user=www-data
autostart=true
autorestart=true
startsecs=1
stopwaitsecs=60
stopsignal=TERM
redirect_stderr=true
stdout_logfile=/var/log/hermes-worker.logsystemd opisuje Nasadenie do produkcie.
7. Otestujte producenta
V testovacom kontajneri nahraďte EmitterInterface mockom. Od Hermesu 6.0 je Emitter final a nedá sa zamockovať.
<?php
declare(strict_types=1);
namespace App\Tests\Controller;
use Symfony\Bundle\FrameworkBundle\Test\WebTestCase;
use Tomaj\Hermes\EmitterInterface;
use Tomaj\Hermes\MessageInterface;
final class UserControllerTest extends WebTestCase
{
public function testRegistrationEmitsUserRegistered(): void
{
$client = static::createClient();
// Mock the interface: Emitter is final from Hermes 6.0.
$emitter = $this->createMock(EmitterInterface::class);
$emitter->expects(self::once())
->method('emit')
->with(self::callback(
static fn(MessageInterface $message): bool => $message->getType() === 'user.registered'
&& $message->getPayload() === ['email' => 'jane@example.com'],
))
->willReturnSelf();
static::getContainer()->set(EmitterInterface::class, $emitter);
$client->jsonRequest('POST', '/users', ['email' => 'jane@example.com']);
self::assertResponseStatusCodeSame(201);
}
public function testInvalidEmailEmitsNothing(): void
{
$client = static::createClient();
$emitter = $this->createMock(EmitterInterface::class);
$emitter->expects(self::never())->method('emit');
static::getContainer()->set(EmitterInterface::class, $emitter);
$client->jsonRequest('POST', '/users', ['email' => 'not-an-email']);
self::assertResponseStatusCodeSame(422);
}
}Emitovanie po commite do databázy
Ak controller vloží riadok v transakcii a emitne o ňom správu, nasmerujte EmitterInterface na BufferedEmitter, aby worker nikdy nehľadal riadok, ktorý ešte nie je commitnutý. Doctrine DBAL nemá callbacky po commite, preto zavolajte flush() po návrate z transactional() a discard(), keď vyhodí výnimku:
Tomaj\Hermes\Emitter: ~
Tomaj\Hermes\EmitterInterface:
class: Tomaj\Hermes\BufferedEmitter
arguments:
- '@Tomaj\Hermes\Emitter'
- ['@doctrine.dbal.default_connection', 'isTransactionActive']try {
$connection->transactional(function () use ($emitter): void {
// INSERT ...
$emitter->emit(new Message('user.registered', ['id' => $id]));
});
} catch (\Throwable $e) {
$emitter->discard();
throw $e;
}
$emitter->flush();isTransactionActive() zabezpečí, že mimo transakcie emitter odosiela hneď. Predikát a volania flush()/discard() okolo transactional() boli overené s Doctrine DBAL 4.5.0; definícia služby nebola spustená v Symfony kerneli.
Náhodná zložka opakovania a služba Random\Randomizer
Od Hermesu 6.0 Dispatcher prijíma voliteľný ?Random\Randomizer $randomizer (pozrite Integrácia do frameworkov). Ak vaša aplikácia registruje službu Random\Randomizer, autowiring ju odovzdá dispatcheru a náhodná zložka opakovania pochádza z tohto enginu. Ak chcete zachovať predvolený bezpečný engine, na Hermese 6.0 alebo novšom pridajte toto do služby dispatchera:
Tomaj\Hermes\DispatcherInterface:
class: Tomaj\Hermes\Dispatcher
arguments:
$randomizer: nullNa Hermese 5.x kontajner zlyhá s has no argument named "$randomizer", preto to tam vynechajte.
Celý zdrojový kód
Všetky súbory, s install.sh, e2e.sh a programom pre Supervisor, sú v examples/frameworks/symfony na GitHube. Overené so Symfony 7.4.21 a monolog-bundle 4.1.0 na PHP 8.4, so všetkými drivermi.
Časté otázky
Potrebujem Symfony Messenger?
Nie. Toto nastavenie Messenger neinštaluje ani nepoužíva. Oba môžu bežať v jednej aplikácii, ak používajú rôzne fronty; pozrite Integrácia do frameworkov.
Prečo je handler zaregistrovaný v services.yaml a nie tagom?
Jeden riadok calls na typ správy sa ľahko číta a nepotrebuje compiler pass. Pri veľa handleroch funguje aj tagovaný iterátor a malý cyklus vo factory. Pozrite Pojmy.
Worker po nasadení nevidí môj nový kód. Prečo?
Je to dlhobežiaci proces so starým kontajnerom v pamäti. Spustite bin/console hermes:restart. Pozrite Ukončenie a limit správ.
Môže worker bežať s APP_ENV=prod?
Áno, a mal by. Pred spustením zahrejte cache, aby prvý worker nevytváral kontajner, kým ostatné čakajú. Pozrite Nasadenie do produkcie.
Ako nasmerovať driver sqs na LocalStack alebo moto?
Nastavte HERMES_SQS_ENDPOINT=http://127.0.0.1:4566 a fiktívne kľúče AWS. Pozrite Amazon SQS.