Prioritné fronty

Cieľ: e-mail na obnovu hesla odíde pred potvrdením platby a potvrdenie pred newsletterom, aj keď čakajú tisíce newsletterov. Jedného workera si navyše môžete nechať voľného len na urgentnú prácu.

Spustiteľný projekt

Každý súbor nižšie sa načítava z examples/priority, kde je Makefile a e2e.sh, ktorý spustí celý postup. Naklonujte repozitár, spustite cd examples/priority a potom make install up.

Hlavné súbory

• src/PrintHandler.php
• bin/worker.php
• public/index.php
• docker-compose.yml

Zdrojový kód na GitHube

Súbory

Projekt spúšťa Redis z vlastného docker-compose.yml, ako v návode na uvítací e-mail. Tento recept má vlastný src/Hermes.php, pretože obe strany musia zaregistrovať rovnaké prioritné fronty.

{
    "require": {
        "php": ">=8.4",
        "tomaj/hermes": "dev-master",
        "predis/predis": "^2.3 || ^3.0"
    },
    "autoload": {
        "psr-4": {
            "App\\": "src/"
        }
    },
    "minimum-stability": "dev",
    "prefer-stable": true
}

<?php

declare(strict_types=1);

namespace App;

use Predis\Client;
use Tomaj\Hermes\Dispatcher;
use Tomaj\Hermes\Driver\PredisSetDriver;

final class Hermes
{
    public const LOW = Dispatcher::DEFAULT_PRIORITY;          // 100, the queue the driver always has
    public const NORMAL = Dispatcher::DEFAULT_PRIORITY + 50;  // 150
    public const HIGH = Dispatcher::DEFAULT_PRIORITY + 100;   // 200

    public static function driver(?Client $redis = null): PredisSetDriver
    {
        $driver = new PredisSetDriver($redis ?? new Client(getenv('REDIS_URL') ?: 'tcp://127.0.0.1:6379'), 'hermes', 1);

        // The emitter and every worker must register the same queues.
        // Register them in ascending order: the worker polls them in reverse registration order.
        $driver->setupPriorityQueue('hermes_normal', self::NORMAL);
        $driver->setupPriorityQueue('hermes_high', self::HIGH);

        return $driver;
    }
}

<?php

declare(strict_types=1);

use App\Hermes;
use Tomaj\Hermes\Emitter;
use Tomaj\Hermes\Message;

require __DIR__ . '/../vendor/autoload.php';

$emitter = new Emitter(Hermes::driver());

// A password reset is waiting for a person: high.
$emitter->emit(new Message('email.password_reset', ['email' => 'jane@example.com']), Hermes::HIGH);

// A receipt can take a few seconds: normal.
$emitter->emit(new Message('email.receipt', ['order_id' => 1042]), Hermes::NORMAL);

// The weekly newsletter can wait for an idle worker: low.
foreach (range(1, 3) as $i) {
    $emitter->emit(new Message('email.newsletter', ['subscriber_id' => $i]), Hermes::LOW);
}

echo "Queued 5 messages.\n";

<?php

declare(strict_types=1);

namespace App;

use Tomaj\Hermes\Handler\HandlerInterface;
use Tomaj\Hermes\MessageInterface;

final class PrintHandler implements HandlerInterface
{
    #[\Override]
    public function handle(MessageInterface $message): bool
    {
        // Replace with the real work for each type.
        printf("[%s] %s %s\n", date('H:i:s'), $message->getType(), json_encode($message->getPayload()));

        return true;
    }
}

<?php

declare(strict_types=1);

use App\Hermes;
use App\PrintHandler;
use Tomaj\Hermes\Dispatcher;

require __DIR__ . '/../vendor/autoload.php';

// php bin/worker.php          takes every priority, highest first
// php bin/worker.php high     takes only the high queue
$priorities = ($argv[1] ?? '') === 'high' ? [Hermes::HIGH] : [];

$dispatcher = new Dispatcher(Hermes::driver());
$handler = new PrintHandler();
foreach (['email.password_reset', 'email.receipt', 'email.newsletter'] as $type) {
    $dispatcher->registerHandler($type, $handler);
}
$dispatcher->handle($priorities);

Spustenie

Najprv vložte správy a workera spustite až potom, aby všetkých päť správ čakalo v Redise a videli ste poradie:

composer install
docker compose up -d redis
php public/index.php
php bin/worker.php

[09:16:53] email.password_reset {"email":"jane@example.com"}
[09:16:53] email.receipt {"order_id":1042}
[09:16:53] email.newsletter {"subscriber_id":2}
[09:16:53] email.newsletter {"subscriber_id":3}
[09:16:53] email.newsletter {"subscriber_id":1}

Teraz spustite workera len na urgentnú prácu:

php public/index.php
php bin/worker.php high

Spracuje obnovu hesla a ostatné štyri nechá v Redise pre bežného workera. make e2e skontroluje oba behy.

Úskalia

• Vo verzii 5.x registrujte fronty vo vzostupnom poradí. Worker v 5.x číta fronty v opačnom poradí registrácie, nie podľa čísla. Hermes 6.0 ich zoraďuje podľa čísla, takže na poradí už nezáleží. Driver vždy registruje predvolenú prioritu 100 ako prvú. Tento recept preto používa 100 ako najnižšiu prioritu a v tomto poradí pridáva 150 a 200. Priorita pod 100 by sa čítala pred 100. Pozrite Priorita a issue #50.
• Emitter a worker potrebujú rovnaké fronty. Vloženie s prioritou, ktorú driver nezaregistroval, vyhodí UnknownPriorityException.
• V rámci jednej priority nie je poradie. Tri newslettery vyjdú v náhodnom poradí, pretože množina v Redise žiadne nemá.
• Opakovania idú na predvolenú prioritu. Dispatcher pošle kópiu na opakovanie bez priority, takže neúspešná obnova hesla sa zopakuje so 100, za potvrdeniami platieb.
• Nízka priorita môže čakať navždy. Kým prichádzajú správy s vysokou prioritou, worker, ktorý číta všetky fronty, sa k nízkym nikdy nedostane. Ak newslettery musia odísť, pridajte workera pre nízku prioritu.
• Priority majú Redis drivery a FileSystemDriver. Na RabbitMQ, SQS a Kafka setupPriorityQueue() vyhodí NotSupportedException. Tam použite samostatné fronty a workery.

Celý zdrojový kód

Všetky súbory tohto receptu, s README.md, Makefile a kontrolou e2e.sh, sú v examples/priority na GitHube. Naklonujte repozitár, spustite cd examples/priority a make install up, potom make e2e.

Časté otázky

Prečo je najnižšia priorita 100?

Vo verzii 5.x driver vždy registruje 100 ako prvú a číta v opačnom poradí registrácie, takže priorita pod 100 by sa spustila pred ňou. Od 6.0 sa fronty čítajú podľa čísla (#50) a nižšie priority fungujú podľa očakávania.

Prečo newslettery vychádzajú v náhodnom poradí?

Každá priorita je množina v Redise, ktorá nemá poradie.

Kam idú opakovania?

Na predvolenú prioritu 100, pretože dispatcher posiela kópie na opakovanie bez priority.

Ako si nechať workera voľného na urgentnú prácu?

Spustite ho cez php bin/worker.php high, ktorý volá handle([200]).

Dokáže to RabbitMQ alebo SQS?

Nie. Tam použite samostatné fronty a samostatné workery.