Priority queues

Goal: a password reset email goes out before a receipt, and a receipt before a newsletter, even when thousands of newsletters are waiting. One worker can also be kept free for urgent work only.

Runnable project

Every file below is imported from examples/priority, which has a Makefile and an e2e.sh that runs the whole flow. Clone the repository, cd examples/priority, and run make install up.

Key files

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

View source on GitHub

Files

The project starts Redis from its own docker-compose.yml, as in the welcome email tutorial. This recipe has its own src/Hermes.php, because both sides must register the same priority queues.

{
    "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);

Run it

Emit first and start the worker afterwards, so all five messages wait in Redis and you can see the order:

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}

Now run a worker for urgent work only:

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

It handles the password reset and leaves the other four in Redis for a normal worker. make e2e checks both runs.

Pitfalls

• On 5.x, register queues in ascending order. The 5.x worker polls queues in reverse registration order, not by number. Hermes 6.0 sorts them by number, so the order no longer matters. The driver always registers the default priority 100 first. This recipe therefore uses 100 as the lowest priority and adds 150 and 200 in that order. A priority below 100 would be polled before 100. See Priority and issue #50.
• Emitter and worker need the same queues. Emitting with a priority that the driver did not register throws UnknownPriorityException.
• No order inside one priority. The three newsletters come out in random order, because a Redis set has none.
• Retries go to the default priority. The dispatcher sends a retry copy without a priority, so a failed password reset is retried at 100, behind receipts.
• Low priority can wait forever. While high messages keep arriving, a worker that reads all queues never reaches the low ones. Add a worker for low work if newsletters must go out.
• Only the Redis drivers have priorities. On RabbitMQ and SQS, setupPriorityQueue() throws NotSupportedException. Use separate queues and workers there.

Full source

Every file of this recipe, with its README.md, Makefile and the e2e.sh check, is in examples/priority on GitHub. Clone the repository, cd examples/priority, and run make install up, then make e2e.

FAQ

Why is the lowest priority 100?

On 5.x the driver always registers 100 first and polls in reverse registration order, so a priority below 100 would run before it. From 6.0 the queues are polled by number (#50) and lower priorities work as expected.

Why do newsletters come out in random order?

Each priority is a Redis set, which has no order.

Where do retries land?

On the default priority 100, because the dispatcher sends retry copies without a priority.

How do I keep a worker free for urgent work?

Start it with php bin/worker.php high, which calls handle([200]).

Can RabbitMQ or SQS do this?

No. Use separate queues and separate workers there.