FileSystem

Tomaj\Hermes\Driver\FileSystemDriver drží frontu v lokálnom adresári. Každá správa je jeden JSON súbor. Nepotrebuje broker ani PHP rozšírenie, takže sa hodí na jeden server, malú aplikáciu, CLI nástroj alebo vývojový stroj, kde nechcete spúšťať Redis. Podporuje priority, odložené správy a opakovanie a spracúva správy v poradí FIFO.

public function __construct(
    string $directory,
    string $queue = 'hermes',
    int $refreshInterval = 1,
    int $visibilityTimeout = 300
)

• $directory sa pri chýbaní vytvorí pri prvom send() alebo wait(). Priečinky dostanú režim 0775 mínus umask.
• $queue je názov priečinka fronty pre Dispatcher::DEFAULT_PRIORITY. Názvy front môžu obsahovať písmená, číslice, ., _ a - a nesmú začínať .. Iné názvy vyhodia InvalidArgumentException.
• $refreshInterval je, koľko sekúnd wait() spí, keď nie je splatná žiadna správa. 0 polluje bez spánku.
• $visibilityTimeout je, koľko sekúnd môže zostať prevzatá správa nedokončená, kým sa doručí znova. Nastavte ju nad váš najpomalší handler. 0 nikdy neprinesie opätovné doručenie.

Záporné $refreshInterval alebo $visibilityTimeout vyhodí InvalidArgumentException.

Vloženie a spracovanie

Toto je examples/filesystem, worker, ktorý spúšťa exporty reportov. Stačí composer install.

Spustiteľný projekt

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

Hlavné súbory

• src/Hermes.php
• src/ExportHandler.php
• bin/emit.php
• bin/worker.php
• bin/shutdown.php
• e2e.sh

Zdrojový kód na GitHube

Obe strany skladajú driver na jednom mieste, aby sa emitter a worker zhodli na adresári a názvoch front:

<?php

declare(strict_types=1);

namespace App;

use Tomaj\Hermes\Dispatcher;
use Tomaj\Hermes\Driver\FileSystemDriver;

final class Hermes
{
    public const int PRIORITY_HIGH = Dispatcher::DEFAULT_PRIORITY + 10;

    public static function directory(): string
    {
        return getenv('HERMES_DIR') ?: __DIR__ . '/../var/hermes';
    }

    public static function driver(): FileSystemDriver
    {
        // <dir>/<queue>/{tmp,new,processing,failed}, created on first use
        $driver = new FileSystemDriver(self::directory(), 'exports', refreshInterval: 1, visibilityTimeout: 300);
        $driver->setupPriorityQueue('exports_high', self::PRIORITY_HIGH);

        return $driver;
    }
}

Emitter zapíše súbor a vráti sa:

<?php

declare(strict_types=1);

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

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

// php bin/emit.php <report> [delay seconds] [high]
$report = $argv[1] ?? 'sales';
$delay = (int) ($argv[2] ?? 0);
$priority = ($argv[3] ?? '') === 'high' ? Hermes::PRIORITY_HIGH : Dispatcher::DEFAULT_PRIORITY;

$emitter = new Emitter(Hermes::driver());
$emitter->emit(new Message('report.export', ['report' => $report], executeAt: $delay > 0 ? microtime(true) + $delay : null), $priority);

echo "queued {$report}\n";

Worker používa SharedFileShutdown, ktorý tiež potrebuje len súbor:

<?php

declare(strict_types=1);

use App\ExportHandler;
use App\Hermes;
use Tomaj\Hermes\Dispatcher;
use Tomaj\Hermes\Shutdown\SharedFileShutdown;

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

$driver = Hermes::driver();
// php bin/shutdown.php touches this file; workers started before that exit
$driver->setShutdown(new SharedFileShutdown(Hermes::directory() . '/shutdown'));

$dispatcher = new Dispatcher($driver);
$dispatcher->registerHandler('report.export', new ExportHandler(Hermes::directory() . '/exports.log'));
$dispatcher->handle();

<?php

declare(strict_types=1);

use App\Hermes;
use Tomaj\Hermes\Shutdown\SharedFileShutdown;

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

@mkdir(Hermes::directory(), 0o775, true);
new SharedFileShutdown(Hermes::directory() . '/shutdown')->shutdown();
echo "workers asked to stop\n";

Rozloženie adresára

Každá fronta je priečinok v $directory so štyrmi podpriečinkami:

var/hermes/
├── exports/            fronta pre DEFAULT_PRIORITY
│   ├── tmp/            send() najprv píše sem
│   ├── new/            čakajúce správy
│   ├── processing/     prevzaté workerom
│   └── failed/         nepodarilo sa deserializovať
└── exports_high/       jeden priečinok na každé setupPriorityQueue()

send() zapíše JSON do tmp/ a premenuje ho do new/. Worker preto nikdy nečíta napoly zapísaný súbor. Keď zápis alebo premenovanie zlyhá, send() vyhodí FileSystemDriverException.

Worker prevezme správu premenovaním z new/ do processing/. Po návrate callbacku sa súbor zmaže. Inak:

• Callback vyhodí výnimku. Súbor ide späť do new/ a výnimka zastaví wait(), rovnako ako pri ostatných driveroch. Chyby handlerov, ktoré dispatcher zachytí skôr, než dôjdu k driveru, pozrite v Opakovaní.
• Worker zomrie. Súbor zostane v processing/. Prevzatie ho premenuje na <čas prevzatia>~<názov>, takže čas prevzatia nastaví to isté atomické premenovanie. Po $visibilityTimeout sekundách ho ktorýkoľvek worker vráti do new/ pod pôvodným názvom.
• JSON sa nedá deserializovať. Súbor ide do failed/ a wait() vyhodí SerializeException. Tieto súbory skontrolujte alebo zmažte ručne.

Doručenie je at-least-once. Handler, ktorý beží dlhšie ako $visibilityTimeout, môže bežať dvakrát, preto robte handlery idempotentné alebo timeout zdvihnite.

Poradie, oneskorenie a viacerí workeri

Názov súboru začína časom splatnosti správy a potom náhodnou príponou:

1700000000.250000-3f9c1a2b7d4e5f60.json

Čas je executeAt, alebo čas vytvorenia, keď žiadny nie je. Je doplnený nulami, takže zoradenie názvov zoradí podľa splatnosti. wait() číta new/ v tom poradí, čo robí každú frontu FIFO. Zastaví sa pri prvom súbore, ktorý ešte nie je splatný, takže odložená správa čaká v new/ bez samostatného plánovača.

rename() je atomický na jednom lokálnom súborovom systéme. Keď dvaja workeri premenujú ten istý súbor, jeden vyhrá a druhý ide na ďalší. Na adresári môžete spustiť toľko workerov, koľko chcete.

Priorita funguje ako pri Redis driveroch: najvyššie číslo sa polluje prvé. Worker spustený s handle([...]) polluje len tieto priority a pri priorite bez fronty vyhodí UnknownPriorityException.

Obmedzenia

• Jeden server. rename() je atomický len na lokálnom súborovom systéme. Nezdieľajte adresár cez NFS ani iný sieťový súborový systém. Štyri priečinky fronty sú podpriečinky jedného adresára, takže premenovanie nikdy neprechádza medzi súborovými systémami.
• Jeden súbor na správu. Desiatky tisíc čakajúcich správ spôsobia, že každý poll číta veľký adresár. Na vysoký objem použite Redis alebo RabbitMQ.
• Pollovanie. Nečinný worker sa zobudí každých $refreshInterval sekúnd.
• Zálohy. Fronta sú obyčajné súbory. Správa v processing/ je na ceste. Neobnovujte ju, kým workeri bežia.

Časté otázky

Kedy mám použiť FileSystem driver?

Keď emitter a worker bežia na tom istom serveri a nechcete spúšťať broker: malá aplikácia, CLI nástroj, náhrada cronu alebo lokálny vývoj. Porovnajte ho s ostatnými drivermi na stránke Drivery.

Môžu viacerí workeri zdieľať jeden adresár?

Áno, na jednom serveri. Každý worker prevezme správu atomickým rename(), takže správa sa spracuje raz. Nie na NFS, pozrite Obmedzenia.

Čo sa stane, keď worker spadne?

Správa zostane v processing/ a po $visibilityTimeout sekundách sa vráti do new/, takže ju spracuje iný worker. Pozrite Rozloženie adresára.

Sú podporované odložené správy a opakovanie?

Áno. Názov súboru začína časom splatnosti a wait() preskakuje súbory, ktoré ešte nie sú splatné. Opakovania z RetryTrait sú tiež odložené správy, takže fungujú bez ďalšieho nastavenia. Pozrite Odložené spracovanie.

Ako zastavím workery?

Použite SharedFileShutdown so súborom vedľa fronty, ako examples/filesystem/bin/shutdown.php. S ext-pcntl funguje aj SIGTERM.

Existuje úplný príklad?

Áno. examples/filesystem spúšťa exporty reportov a jeho make e2e overí poradie priorít, odloženú správu a ukončenie.