PHP API
A Queuety minden funkciója elérhető az Queuety\Queuety statikus osztályon keresztül. Ez az egyetlen belépési pont feladatok indításához, munkafolyamatok összeállításához, állapotgépek vezérléséhez, várólisták kezeléséhez és a belső alrendszerek eléréséhez.
use Queuety\Queuety;
Önálló hozzáférés az alapfunkciókhoz
A Queuety adatbázis-alapú futtatókörnyezete a WordPress keretrendszeren kívül is inicializálható.
Ez az önálló működési mód a vezénylési alapfunkciókhoz való:
- feladatok és munkások (a munkáskészletekhez a
DB_HOST,DB_NAME,DB_USERésDB_PASSWORDkonstansokra is szükség van, lásd alább) - munkafolyamatok és jelzések
- állapotgépek és események
- artefaktumok, naplók, mérőszámok és ütemezések
Ez nem helyettesíti a WordPress adaptert. A Queuety::on_action(), a
replace_wp_cron(), a bővítmény inicializálása és a valódi wp queuety ...
regisztrációs útvonal továbbra is WordPress-t igényel.
Queuety::init( Connection $conn )
A Queuety inicializálása közvetlen adatbázis-kapcsolattal.
use Queuety\Connection;
use Queuety\Queuety;
$conn = new Connection(
host: '127.0.0.1:3306',
dbname: 'wordpress',
user: 'root',
password: 'secret',
prefix: 'wp_',
);
Queuety::init( $conn );
A nem kötelező table_prefix argumentum kapcsolatonként módosítja a futásidejű táblanevek Queuety részét:
$conn = new Connection(
host: '127.0.0.1:3306',
dbname: 'wordpress',
user: 'root',
password: 'secret',
prefix: 'wp_',
table_prefix: 'themequeue_',
);
Így például az queuety_jobs tábla neve wp_themequeue_jobs lesz.
Queuety::ensure_schema()
Ellenőrizd, hogy az aktív kapcsolathoz tartozó Queuety futásidejű táblák léteznek-e.
Queuety::init( $conn );
Queuety::ensure_schema();
A teljes modellről és a WordPress határról lásd az Önálló használat című részt.
Feladatok
Queuety::dispatch( $handler, $payload )
Feladat indítása. Handlernévként karakterláncot vagy Contracts\Job példányt fogad. A beállítások láncolásához egy PendingJob építőt ad vissza.
// Handler name + payload (original API)
Queuety::dispatch( 'send_email', [ 'to' => 'user@example.com' ] );
// Contracts\Job instance (v0.6.0+)
Queuety::dispatch( new SendEmailJob( to: 'user@example.com' ) );
// With options
Queuety::dispatch( 'send_email', [ 'to' => 'user@example.com' ] )
->on_queue( 'emails' )
->with_priority( Priority::High )
->delay( 300 )
->max_attempts( 5 )
->rate_limit( 10, 60 )
->unique()
->after( $other_job_id );
// Get the job ID
$job_id = Queuety::dispatch( 'send_email', $payload )->id();
Ha Contracts\Job példányt adsz át, a szerializáló a teljes minősítésű osztálynevet használja handlernévként, a nyilvános tulajdonságokat pedig a hasznos adatként. A $payload paraméter figyelmen kívül marad, ha a $handler egy Job példánya.
PendingJob metódusai
| Metódus | Leírás |
|---|---|
on_queue( string $queue ) |
Cél várólista beállítása |
with_priority( Priority $priority ) |
Prioritási szint beállítása |
delay( int $seconds ) |
Késleltetés, mielőtt a feladat elérhetővé válik |
max_attempts( int $max ) |
Újrapróbálkozások maximális száma |
rate_limit( int $max, int $window ) |
Sebességkorlát: végrehajtások maximális száma az időablak másodperceiben |
unique() |
Ismételt indítások megakadályozása |
after( int $job_id ) |
Várakozás egy másik feladat befejezésére |
concurrency_group( string $group, int $limit ) |
Feladat hozzáadása egy megosztott munkásfelvételi csoporthoz |
cost_units( int $units ) |
A végrehajtás viszonylagos költségét leíró metaadat csatolása az indításhoz |
id() |
Kényszerített indítás, a feladatazonosító visszaadásával |
A concurrency_group() és a cost_units() a tágabb Erőforrás-kezelési modell része, így ugyanaz az indítás beleszámíthat a darabszámalapú párhuzamossági korlátokba, a súlyozott várólista- vagy szolgáltatói keretekbe, valamint a munkások felvételi heurisztikáiba.
Queuety::dispatch_sync( $handler, $payload )
Feladat szinkron végrehajtása az aktuális folyamatban, várólistára helyezés nélkül.
// With a Job instance
Queuety::dispatch_sync( new SendEmailJob( to: 'user@example.com', subject: 'Hi', body: 'Hello' ) );
// With a handler name and payload
Queuety::dispatch_sync( 'send_email', [ 'to' => 'user@example.com' ] );
A feladat handle() metódusa közvetlenül hívódik meg. Nem jön létre adatbázissor, nincs szükség munkásra, és a köztes szoftverek sem futnak le.
Queuety::batch( $jobs )
Több feladat indítása egyetlen több soros INSERT utasítással.
$job_ids = Queuety::batch( [
[ 'handler' => 'send_email', 'payload' => [ 'to' => 'a@example.com' ], 'queue' => 'emails' ],
[ 'handler' => 'send_email', 'payload' => [ 'to' => 'b@example.com' ], 'priority' => 2 ],
] );
Queuety::create_batch( $jobs )
Hozz létre egy BatchBuilder példányt feladatcsoport indításához, visszahívásokkal és előrehaladás-követéssel. A részletekért lásd a Csoportosítás részt.
$batch = Queuety::create_batch( [
new ImportUsersJob( $chunk_1 ),
new ImportUsersJob( $chunk_2 ),
] )
->name( 'Import users' )
->then( ImportCompleteHandler::class )
->catch( ImportFailedHandler::class )
->finally( ImportCleanupHandler::class )
->allow_failures()
->on_queue( 'imports' )
->dispatch();
BatchBuilder metódusai
| Metódus | Visszatérési érték | Leírás |
|---|---|---|
name( string $name ) |
self |
Csoportnév megadása |
on_queue( string $queue ) |
self |
Várólista beállítása a csoport összes feladatához |
then( string $handler_class ) |
self |
Sikeres csoportbefejezéskor meghívott handler |
catch( string $handler_class ) |
self |
A csoport első sikertelen feladatakor meghívott handler |
finally( string $handler_class ) |
self |
A csoport befejezésekor (siker vagy hiba esetén) meghívott handler |
allow_failures() |
self |
A then visszahívás engedélyezése hibák esetén is |
dispatch() |
Batch |
Csoportsor és összes feladat létrehozása, a Batch értékobjektum visszaadásával |
Queuety::find_batch( $id )
Csoport keresése azonosító alapján. Batch értékobjektumot vagy null értéket ad vissza.
$batch = Queuety::find_batch( $batch_id );
if ( null !== $batch ) {
echo $batch->progress() . '% complete';
}
Munkások vezérlése
Queuety::run_worker( $queue = 'default', $once = false )
Egy munkásciklus közvetlen futtatása.
Queuety::run_worker( 'default' );
Queuety::run_worker( 'critical,default', true );
Queuety::run_worker_pool( $workers, $queue = 'default' )
Rögzített méretű munkáskészlet közvetlen futtatása. Szükség van a pcntl és a posix bővítményre. A készlet a gyermekkapcsolatokhoz a WordPress, DB_HOST, DB_NAME, DB_USER és DB_PASSWORD konstansokat olvassa be (csak a táblaprefixek származnak az inicializált kapcsolatból), ezért munkáskészlet WordPress-n kívüli futtatásakor ezeket is állítsd be (#5).
Queuety::run_worker_pool( 4, 'providers' );
Queuety::run_auto_scaling_worker_pool( $min_workers, $max_workers, $queue = 'default' )
Adaptív munkáskészlet közvetlen futtatása. Ugyanazok a követelmények érvényesek, mint a run_worker_pool() esetén.
Queuety::run_auto_scaling_worker_pool( 2, 6, 'providers' );
Batch értékobjektum
| Metódus/tulajdonság | Típus | Leírás |
|---|---|---|
$id |
int |
Csoportazonosító |
$name |
string|null |
Nem kötelező csoportnév |
$total_jobs |
int |
A csoport feladatainak teljes száma |
$pending_jobs |
int |
Még függőben lévő feladatok |
$failed_jobs |
int |
Sikertelen feladatok száma |
$failed_job_ids |
array |
A sikertelen feladatok azonosítói |
$created_at |
DateTimeImmutable |
A csoport létrehozásának időpontja |
$finished_at |
DateTimeImmutable|null |
A csoport befejezésének időpontja |
$cancelled_at |
DateTimeImmutable|null |
A csoport törlésének időpontja |
progress() |
int |
Befejezettség százalékos aránya (0-100) |
finished() |
bool |
Befejeződött-e a csoport |
has_failures() |
bool |
Sikertelen volt-e bármely feladat |
cancelled() |
bool |
Törölték-e a csoportot |
Queuety::chain( $jobs )
Hozz létre egy ChainBuilder példányt a feladatok egymás utáni végrehajtásához. Minden feladat az előző sikeres befejezésétől függ.
$first_job_id = Queuety::chain( [
new FetchDataJob( $url ),
new ProcessDataJob(),
new NotifyCompleteJob(),
] )
->on_queue( 'pipeline' )
->catch( ChainFailedHandler::class )
->dispatch();
ChainBuilder metódusai
| Metódus | Visszatérési érték | Leírás |
|---|---|---|
on_queue( string $queue ) |
self |
Várólista beállítása a lánc összes feladatához |
catch( string $handler_class ) |
self |
Handler meghívása, ha a lánc bármely feladata véglegesen eltemetett állapotba kerül |
dispatch() |
int |
Összes feladat létrehozása egymás utáni függőségekkel, az első feladatazonosító visszaadásával |
Queuety::retry( $job_id )
Adott feladat újrapróbálása. Állapota visszaáll pending értékre.
Queuety::retry( 42 );
Queuety::retry_buried()
Az összes eltemetett feladat újrapróbálása. Visszaadja az újrapróbált feladatok számát.
$count = Queuety::retry_buried();
Queuety::purge( $older_than_days )
A befejezett feladatok törlése. Visszaadja a törölt feladatok számát.
$count = Queuety::purge(); // uses QUEUETY_RETENTION_DAYS
$count = Queuety::purge( 30 ); // purge jobs older than 30 days
Várólisták
Queuety::stats( $queue )
Feladatszámok lekérése állapot szerint csoportosítva.
$stats = Queuety::stats(); // all queues
$stats = Queuety::stats( 'emails' ); // specific queue
// ['pending' => 12, 'processing' => 3, 'completed' => 150, 'failed' => 2, 'buried' => 1]
Queuety::buried( $queue )
Az összes eltemetett feladat lekérése.
$jobs = Queuety::buried();
$jobs = Queuety::buried( 'emails' );
Queuety::pause( $queue )
Szüneteltesse a sort, hogy a munkások kihagyják.
Queuety::pause( 'emails' );
Queuety::resume( $queue )
Folytassa a szüneteltetett sort.
Queuety::resume( 'emails' );
Queuety::is_paused( $queue )
Ellenőrizze, hogy szünetel-e egy sor.
if ( Queuety::is_paused( 'emails' ) ) {
// Queue is paused
}
Munkafolyamatok
Queuety::workflow( $name )
Kezdje el egy munkafolyamat felépítését. Egy WorkflowBuilder értéket ad vissza.
$workflow_id = Queuety::workflow( 'generate_report' )
->then( FetchDataHandler::class )
->then( CallLLMHandler::class )
->then( FormatOutputHandler::class )
->dispatch( [ 'user_id' => 42 ] );
WorkflowBuilder metódusai
| Metódus | Leírás |
|---|---|
then( string $handler_class, ?string $name = null ) |
Szekvenciális lépés hozzáadása. A feltételes elágazáshoz opcionális nevet adhat meg. |
parallel( array $handler_classes ) |
Párhuzamos lépéscsoport hozzáadása. |
for_each( string $items_key, string $handler_class, ?string $result_key = null, ForEachMode $mode = ForEachMode::All, ?int $quorum = null, ?string $reducer_class = null, ?string $name = null ) |
Dinamikus, elemenkénti lépés hozzáadása, amely a munkafolyamat állapotában található minden elemhez egy elágazó feladatot hoz létre. |
repeat_while( string $target_step, ?string $state_key = null, mixed $expected = true, ?string $name = null, ?string $condition_class = null, ?int $max_iterations = null ) |
Visszaugrás egy korábbi, névvel megadott lépésre, amíg egy nyilvános állapotkulcs vagy egy RepeatCondition osztály illeszkedik. |
repeat_until( string $target_step, ?string $state_key = null, mixed $expected = true, ?string $name = null, ?string $condition_class = null, ?int $max_iterations = null ) |
Visszaugrás egy korábbi, névvel megadott lépésre addig, amíg egy nyilvános állapotkulcs vagy egy RepeatCondition osztály illeszkedik. |
run_workflow( string $workflow_name, WorkflowBuilder $sub_builder ) |
Munkafolyamatot futtató lépés hozzáadása. |
delay( int $seconds = 0, int $minutes = 0, int $hours = 0, int $days = 0 ) |
Tartós késleltetési lépés hozzáadása. Az összes időtartam összeadódik. |
wait_for_signal( string $name, ?string $result_key = null, ?string $step_name = null, array $match_payload = [], ?string $correlation_key = null ) |
Jelzésre váró lépés hozzáadása. Opcionális match_payload és correlation_key segítségével figyelmen kívül hagyhatja a nem kapcsolódó eseményeket. |
wait_for_signals( array $signal_names, WaitMode $mode = WaitMode::All, ?string $result_key = null, ?string $name = null, array $match_payload = [], ?string $correlation_key = null ) |
Várakozás több jelzés mindegyikére vagy bármelyikére, opcionális adattartalom-egyeztetéssel és korrelációval. |
wait_for_approval( string $signal_name = 'approval', ?string $result_key = 'approval', ?string $name = null, array $match_payload = [], ?string $correlation_key = null ) |
Várakozás emberi jóváhagyást jelző eseményre. |
wait_for_decision( string $approve_signal = 'approved', string $reject_signal = 'rejected', ?string $result_key = 'decision', ?string $name = null, array $match_payload = [], ?string $correlation_key = null ) |
Várakozás jóváhagyásra vagy elutasításra, és strukturált döntési adattartalom tárolása. |
wait_for_input( string $signal_name = 'input', ?string $result_key = 'input', ?string $name = null, array $match_payload = [], ?string $correlation_key = null ) |
Várakozás emberi bevitelt jelző eseményre. |
wait_for_workflow( int|string $workflow, ?string $result_key = null, ?string $name = null ) |
Várakozás egy munkafolyamat-azonosítóval megadott munkafolyamat befejeződésére, vagy a nyilvános állapotban tárolt munkafolyamat-azonosítóra. |
wait_for_workflows( array|string $workflows, WaitMode $mode = WaitMode::All, ?string $result_key = null, ?string $name = null, ?int $quorum = null ) |
Várakozás több munkafolyamat mindegyikének, bármelyikének vagy egy határozatképes számú munkafolyamatnak a befejeződésére. |
wait_for_workflow_group( string $group_key, WaitMode $mode = WaitMode::All, ?int $quorum = null, ?string $result_key = null, ?string $name = null ) |
Várakozás egy korábban elindított munkafolyamat-csoportra a belső csoportkulcs alapján. |
start_workflows( string $items_key, WorkflowBuilder $workflow_builder, string $result_key = 'started_workflow_ids', string $payload_key = 'item', bool $inherit_state = true, ?string $name = null, ?string $group_key = null ) |
Felső szintű munkafolyamat indítása minden futásidejű elemhez, opcionálisan tartós csoportkulccsal megjelölve az elindított halmazt. |
start_agents( string $items_key, WorkflowBuilder $workflow_builder, string $result_key = 'agent_workflow_ids', string $payload_key = 'agent_task', bool $inherit_state = true, ?string $name = null, ?string $group_key = null ) |
Ügynökökre szabott alias a következőhöz: start_workflows(). |
wait_for_agents( array|string $workflows = 'agent_workflow_ids', WaitMode $mode = WaitMode::All, ?string $result_key = 'agent_results', ?string $name = null, ?int $quorum = null ) |
Ügynökökre szabott alias a következőhöz: wait_for_workflows(). |
wait_for_agent_group( string $group_key = 'agents', WaitMode $mode = WaitMode::All, ?int $quorum = null, ?string $result_key = 'agent_results', ?string $name = null ) |
Ügynökökre szabott alias a következőhöz: wait_for_workflow_group(). |
version( string $version ) |
A munkafolyamat-definíció megjelölése alkalmazásszintű verzióval ellenőrzéshez és exportáláshoz. |
idempotency_key( string $key ) |
Tartóssá teszi a munkafolyamat indítását a hívó által megadott kulcshoz. Az azonos kulccsal történő ismételt indítás az eredeti munkafolyamat-azonosítót adja vissza. |
max_transitions( int $max ) |
A munkafolyamat meghiúsítása, ha több mint $max lépésátmenet fejeződik be. |
max_for_each_items( int $max ) |
A munkafolyamat meghiúsítása, ha egy for_each() lépés több mint $max elemet próbál létrehozni. |
max_state_bytes( int $bytes ) |
A munkafolyamat meghiúsítása, ha a nyilvános állapot mérete meghaladja a $bytes értéket. |
max_cost_units( int $max ) |
A munkafolyamat meghiúsítása, ha a befejezett lépések több mint $max költségegységet használnak fel. |
max_started_workflows( int $max ) |
A munkafolyamat meghiúsítása, mielőtt több mint $max gyermek-munkafolyamatot indítana. |
compensate_with( string $handler_class ) |
Kompenzációkezelő csatolása a legutóbb hozzáadott lépéshez. |
compensate_on_failure() |
A befejezett lépések kompenzációinak automatikus futtatása a munkafolyamat meghiúsulásakor. |
on_queue( string $queue ) |
Sor beállítása az összes lépéshez. |
with_priority( Priority $priority ) |
Prioritás beállítása az összes lépéshez. |
max_attempts( int $max ) |
Lépésenkénti maximális újrapróbálkozási szám beállítása. |
on_cancel( string $handler_class ) |
Takarítókezelő osztály regisztrálása megszakításhoz. Lásd: Megszakítás. |
prune_state_after( int $steps = 2 ) |
Automatikus állapottisztítás engedélyezése. Eltávolítja a küszöbnél régebbi lépések kimeneteit. Lásd: Állapottisztítás. |
must_complete_within( int $seconds = 0, int $minutes = 0, int $hours = 0, int $days = 0 ) |
Határidő beállítása a munkafolyamathoz. Az összes időtartam összeadódik. Lásd: Határidők. |
on_deadline( string $handler_class ) |
Kezelő regisztrálása, amely a munkafolyamat határidejének túllépésekor fut le. Lásd: Határidők. |
dispatch( array $initial_payload = [] ) |
Munkafolyamat indítása. Visszaadja a munkafolyamat-azonosítót. |
Queuety::workflow_status( $workflow_id )
Munkafolyamat állapotának lekérése. Egy WorkflowState objektumot vagy null értéket ad vissza.
$state = Queuety::workflow_status( $workflow_id );
echo $state->name; // 'generate_report'
echo $state->status->value; // 'running'
echo $state->current_step; // 1
echo $state->total_steps; // 3
print_r( $state->state ); // accumulated public state array
echo $state->wait_type; // 'signal' or 'workflow' while blocked
echo $state->wait_mode; // 'all', 'any', or 'quorum'
echo $state->definition_version; // 'research.v2'
echo $state->definition_hash; // '4c7a...'
echo $state->idempotency_key; // 'brief:42:research'
echo $state->artifact_count; // 2
print_r( $state->budget ); // configured workflow budget + current usage
print_r( $state->wait_details ); // matched/remaining blockers, result key, correlation info
print_r( $state->artifact_keys ); // ['draft', 'citations']
A WorkflowState emellett a következőket is tartalmazza:
waiting_for: a futást jelenleg blokkoló jelzésnevek vagy munkafolyamat-azonosítókcurrent_step_name: az aktuális lépés beállított neve, ha van ilyenwait_mode: az aktuális várakozási mód (all,anyvagyquorum) blokkolás eseténwait_details: további vizsgálati adatok, például az egyező és a még fennálló blokkolókdefinition_version: aversion()által beállított érték, ha van ilyendefinition_hash: a munkafolyamat-definíció determinisztikus hash-értékeidempotency_key: a tartós indítási kulcs, ha van ilyenbudget: beállított korlátok és aktuális használat, ha a védelmi korlátok engedélyezve vannakartifact_count: a munkafolyamathoz tárolt artefaktumok számaartifact_keys: a gyors ellenőrzéshez tárolt artefaktumkulcsok
Queuety::retry_workflow( $workflow_id )
Sikertelen munkafolyamat újrapróbálása a meghiúsult lépéstől.
Queuety::retry_workflow( $workflow_id );
RuntimeException hibát dob, ha a munkafolyamatot egy hiba után már kompenzálták.
Queuety::pause_workflow( $workflow_id )
Futó munkafolyamat szüneteltetése. Az aktuális lépéshez már sorba állított munka még befejeződhet, de a későbbi lépések nem kerülnek sorba, amíg nem folytatja a munkafolyamatot.
Queuety::pause_workflow( $workflow_id );
Queuety::resume_workflow( $workflow_id )
Szüneteltetett munkafolyamat folytatása.
Queuety::resume_workflow( $workflow_id );
Queuety::cancel_workflow( $workflow_id )
Munkafolyamat megszakítása és a regisztrált takarítókezelők futtatása. Az állapotot cancelled értékre állítja, és a munkafolyamathoz tartozó összes függőben lévő és feldolgozás alatt álló feladatot eltemeti.
Queuety::cancel_workflow( $workflow_id );
RuntimeException hibát dob, ha a munkafolyamat már completed vagy cancelled állapotú. Részletekért lásd: Megszakítás.
Queuety::rewind_workflow( $workflow_id, $to_step )
Munkafolyamat visszaállítása egy korábbi lépésre, majd újrafuttatása ettől a ponttól. Az állapot a megadott lépéshez tartozó eseménynapló state_after bejegyzéséből áll vissza, és a következő lépés sorba kerül.
Queuety::rewind_workflow( $workflow_id, to_step: 1 );
RuntimeException hibát dob, ha a cél-lépéshez nem található rögzített állapot. Ehhez szükséges a Munkafolyamat-eseménynapló. Részletekért lásd: Időutazás.
Queuety::fork_workflow( $workflow_id )
Futó munkafolyamat elágaztatása független másolattá az aktuális állapotában. Visszaadja az új munkafolyamat-azonosítót.
$forked_id = Queuety::fork_workflow( $workflow_id );
Az elágaztatott munkafolyamat saját azonosítót kap, és az eredetitől függetlenül működik. Részletekért lásd: Elágaztatás.
Queuety::export_workflow( $workflow_id )
A munkafolyamat teljes végrehajtási előzményeinek exportálása JSON formátumban szerializálható tömbként. Tartalmazza a munkafolyamat-sor rekordját, az összes feladatot, a munkafolyamat-eseményeket és a naplóbejegyzéseket.
$data = Queuety::export_workflow( $workflow_id );
A JSON felépítéséért és felhasználási eseteiért lásd: Exportálás és visszajátszás.
Queuety::replay_workflow( $data )
Exportált munkafolyamat visszajátszása az aktuális környezetben. Új munkafolyamatot hoz létre az exportált adatokból, és visszaadja az új munkafolyamat-azonosítót.
$new_id = Queuety::replay_workflow( $data );
Részletekért lásd: Exportálás és visszajátszás.
Queuety::replace_wp_cron()
A beépített WordPress cron rendszer lecserélése a Queuety ütemezőjére. Elfogja a wp_schedule_event(), wp_unschedule_event() és wp_get_scheduled_event() eseményeket, és az ismétlődő eseményeket a Queuety munkáshoz irányítja.
Queuety::replace_wp_cron();
Csak akkor működik, ha a WordPress be van töltve. Részletekért lásd: WP-Cron cseréje.
Queuety::restore_wp_cron()
A WordPress alapértelmezett cron viselkedésének visszaállítása a Queuety cron áthidaló hookok eltávolításával.
Queuety::restore_wp_cron();
A korlátozásokért lásd: WP-Cron cseréje.
Queuety::signal( $workflow_id, $name, $data )
Jelzés küldése egy munkafolyamatnak. Ha a munkafolyamat éppen erre a jelzésre vár, azonnal folytatódik. Ellenkező esetben a jelzés tárolódik, és a munkafolyamat a megfelelő wait_for_signal lépéshez érve veszi fel.
Queuety::signal( $workflow_id, 'approved' );
// With data merged into workflow state
Queuety::signal( $workflow_id, 'approved', [
'approved_by' => 'admin@example.com',
] );
Részletekért lásd: Munkafolyamat-jelzések.
Queuety::approve_workflow( $workflow_id, $data = [], $signal_name = 'approval' )
Szemantikai burkoló a signal() körül, jóváhagyási folyamatokhoz.
Queuety::approve_workflow( $workflow_id, [
'reviewer' => 'editor@example.com',
] );
Queuety::reject_workflow( $workflow_id, $data = [], $signal_name = 'rejected' )
Szemantikai burkoló a signal() körül, elutasítási folyamatokhoz.
Queuety::reject_workflow( $workflow_id, [
'reason' => 'needs citations',
] );
Queuety::submit_workflow_input( $workflow_id, $data = [], $signal_name = 'input' )
Szemantikai burkoló a signal() körül, strukturált emberi bevitelhez.
Queuety::submit_workflow_input( $workflow_id, [
'notes' => 'Ship after legal review',
] );
Queuety::put_artifact( $workflow_id, $artifact_key, $content, $kind = 'json', $step_index = null, $metadata = [] )
Egy tartós artefaktum tárolása egy munkafolyamathoz.
Queuety::put_artifact(
$workflow_id,
'research_brief',
[ 'summary' => 'done' ],
'json',
metadata: [ 'source' => 'agent' ],
);
Queuety::put_current_artifact( $artifact_key, $content, $kind = 'json', $metadata = [] )
Egy artefaktum tárolása az éppen végrehajtott munkafolyamat-lépéshez.
Queuety::put_current_artifact(
'draft',
[ 'status' => 'ready' ],
'json',
);
Queuety::workflow_artifact( $workflow_id, $artifact_key )
Egy tárolt artefaktum lekérése egy munkafolyamathoz.
$artifact = Queuety::workflow_artifact( $workflow_id, 'draft' );
Queuety::workflow_artifacts( $workflow_id, $include_content = false )
Egy munkafolyamat artefaktumainak listázása.
$artifacts = Queuety::workflow_artifacts( $workflow_id );
Queuety::delete_workflow_artifact( $workflow_id, $artifact_key )
Egy munkafolyamat artefaktumának törlése.
Queuety::delete_workflow_artifact( $workflow_id, 'draft' );
Állapotgépek
Queuety::machine( $name )
Kezdje el egy állapotgép felépítését. Egy StateMachineBuilder értéket ad vissza.
use Queuety\Enums\StateMachineStatus;
$machine_id = Queuety::machine( 'agent_session' )
->state( 'awaiting_user' )
->on( 'user_message', 'planning' )
->state( 'planning' )
->action( PlanSessionAction::class )
->on( 'planned', 'completed' )
->state( 'completed', StateMachineStatus::Completed )
->dispatch( [ 'thread_id' => 42 ] );
StateMachineBuilder metódusai
| Metódus | Leírás |
|---|---|
initial( string $state_name ) |
A kezdeti állapot explicit beállítása. |
state( string $state_name, ?StateMachineStatus $terminal_status = null ) |
Egy állapot beállításának megkezdése. |
action( string|array $action_class ) |
Sorba állított belépési művelet regisztrálása az aktuális állapothoz. Adjon meg egy osztálynevet vagy strukturált definíciót class, payload és config értékekkel; a payload a StateAction::handle() értéket $payload formájában kapja meg. |
on( string $event, string $target, string|array|null $guard_class = null, ?string $name = null ) |
Eseményátmenet regisztrálása az aktuális állapotból. Az őrfeltétel ugyanazt a strukturált formát fogadja el, mint a action(). |
on_queue( string $queue ) |
Sor beállítása a sorba állított állapotműveletekhez. |
with_priority( Priority $priority ) |
Prioritás beállítása a sorba állított állapotműveletekhez. |
max_attempts( int $max_attempts ) |
Újrapróbálkozási korlát beállítása a sorba állított állapotműveletekhez. |
version( string $version ) |
A gép definíciójának megjelölése alkalmazásszintű verzióval. |
idempotency_key( string $key ) |
Tartóssá teszi az indítást a hívó által megadott kulcshoz. Az azonos kulccsal történő ismételt indítás az eredeti gépazonosítót adja vissza. |
dispatch( array $initial_state = [] ) |
Gép indítása és a gépazonosító visszaadása. |
Az állapotba belépő műveletek osztályai opcionális config() metódust is megadhatnak, ugyanazokkal az erőforrás-házirendkulcsokkal, amelyeket a munkafolyamat-lépések is támogatnak: concurrency_group, concurrency_limit és cost_units.
Queuety::dispatch_state_machine_definition( $definition, $initial_state = [], $options = [] )
Szerializált állapotgép-definíciós csomag indítása. A legtöbb alkalmazáskódban a machine()->dispatch() használata ajánlott. Ez az alacsonyabb szintű metódus lehetővé teszi, hogy az adapterek közvetlen függőség nélkül indítsanak gépet az építőtől.
$machine_id = Queuety::dispatch_state_machine_definition(
$definition,
[ 'thread_id' => 42 ],
[ 'idempotency_key' => 'thread:42' ],
);
Queuety::machine_status( $machine_id )
Az aktuális gépállapot lekérése. Egy StateMachineState objektumot vagy null értéket ad vissza.
$status = Queuety::machine_status( $machine_id );
echo $status->name; // 'agent_session'
echo $status->status->value; // 'waiting_event'
echo $status->current_state; // 'awaiting_review'
print_r( $status->state ); // public machine state
print_r( $status->available_events ); // valid incoming events in the current state
A StateMachineState emellett a következőket is tartalmazza:
definition_versiondefinition_hashidempotency_keyerror_messagecurrent_actionterminal_status
Queuety::machine_event( $machine_id, $event_name, $payload = [] )
Egy külső esemény küldése egy várakozó gépnek.
Queuety::machine_event(
$machine_id,
'approve',
[
'approved' => true,
'reviewer' => 'editor@example.com',
]
);
Az adattartalom elérhető az átmenetek őrfeltételei számára, és az esemény elfogadásakor összeolvad a gép nyilvános állapotával.
Queuety::list_machines( $limit = 50, $status = null )
A megőrzött gépsor-rekordok listázása ellenőrzéshez.
$machines = Queuety::list_machines();
$waiting = Queuety::list_machines( status: 'waiting_event' );
Minden sor tartalmazza a gépazonosítót, a nevet, az állapotot, az aktuális állapotot, a definíció verzióját, a definíció hash-értékét, az idempotenciakulcsot és az életciklus időbélyegeit.
Queuety::machine_timeline( $machine_id, $limit = 100, $offset = 0 )
Egy gép tartós esemény-idővonalának lekérése.
$timeline = Queuety::machine_timeline( $machine_id );
$page_two = Queuety::machine_timeline( $machine_id, 100, 100 );
Minden eseménysor a következőket tartalmazza: id, machine_id, job_id, state_name, handler, event, event_name, transition_name, queue, attempt, input, output, state_before, state_after, context, artifacts, chunks, error, duration_ms és created_at.
Queuety::machine_trace( $machine_id )
Normalizált nyomkövetési csomag lekérése egy állapotgép-debuggerhez UI.
$trace = Queuety::machine_trace( $machine_id );
A nyomkövetés tartalmazza a következőket: machine, csoportosított states, nyers events, korrelált művelet jobs, logs, artifacts és chunks.
Queuety::machine_state_at( $machine_id, $event_id )
A nyilvános állapot lekérése egy adott gépnyomkövetési esemény után, vagy null, ha nem található.
$state = Queuety::machine_state_at( $machine_id, $event_id );
Az életciklus modelljéért és a használati mintákért lásd: Állapotgépek.
Munkafolyamat-sablonok
Queuety::define_workflow( $name )
Munkafolyamat-építő létrehozása sablonként való használatra.
$builder = Queuety::define_workflow( 'onboarding' )
->then( CreateAccountHandler::class )
->then( SendWelcomeEmailHandler::class );
Queuety::register_workflow_template( $builder )
Munkafolyamat-sablon regisztrálása.
Queuety::register_workflow_template( $builder );
Queuety::run_workflow( $name, $payload, $options = [] )
Regisztrált munkafolyamat-sablon indítása név alapján. Visszaadja a munkafolyamat-azonosítót.
$workflow_id = Queuety::run_workflow( 'onboarding', [ 'email' => 'user@example.com' ] );
$workflow_id = Queuety::run_workflow(
'onboarding',
[ 'email' => 'user@example.com' ],
[ 'idempotency_key' => 'onboarding:user@example.com' ],
);
RuntimeException hibát dob, ha a sablon nincs regisztrálva.
Queuety::dispatch_workflow_definition( $definition, $payload, $options = [] )
Munkafolyamat indítása egy futásidejű definíciós csomagból.
A legtöbb alkalmazáskódban a run_workflow() vagy a workflow()->dispatch() használata ajánlott. Ez az alacsonyabb szintű metódus lehetővé teszi az újrahasználható adaptereknek szerializált munkafolyamat-definíció indítását anélkül, hogy belső munkafolyamat-kezelő metódusoktól függnének.
$workflow_id = Queuety::dispatch_workflow_definition(
$definition,
[ 'email' => 'user@example.com' ],
[ 'idempotency_key' => 'onboarding:user@example.com' ],
);
Queuety::on_action( $hook, $workflow, $map = null, $when = null, $idempotency_key = null, $priority = 10, $accepted_args = null )
WordPress művelet regisztrálása, amely munkafolyamatot indít.
Queuety::on_action(
'save_post',
workflow: 'content_review',
map: static fn ( int $post_id, object $post, bool $update ): array => [
'post_id' => $post_id,
'post_type' => $post->post_type,
'update' => $update,
],
when: static fn ( int $post_id, object $post ): bool => 'post' === $post->post_type,
idempotency_key: static fn ( int $post_id ): string => "save_post:{$post_id}",
);
A $workflow lehet:
- egy regisztrált munkafolyamat-sablon neve
- egy beágyazott
WorkflowBuilder
A map, when és idempotency_key szinkron módon, az aktuális WordPress kérésen belül futnak. Csak maguk a munkafolyamat-lépések futnak aszinkron módon az indítás után.
A teljes útmutatóért lásd: Műveletindítók.
Queuety::workflow_templates()
A munkafolyamat-sablonok nyilvántartásának lekérése.
$registry = Queuety::workflow_templates();
$template = $registry->get( 'onboarding' ); // WorkflowTemplate or null
Queuety::load_workflows( $directory, $recursive )
Az összes munkafolyamat-fájl betöltése és regisztrálása egy könyvtárból. Minden .php fájlnak WorkflowBuilder értéket kell visszaadnia. Visszaadja a regisztrált munkafolyamatok számát.
$count = Queuety::load_workflows( __DIR__ . '/workflows/' );
$count = Queuety::load_workflows( __DIR__ . '/workflows/', recursive: true );
Queuety::load_workflow_file( $file_path )
Egyetlen munkafolyamat-fájl betöltése és regisztrálása. Visszaadja a WorkflowTemplate értéket, vagy a null értéket, ha a fájl nem ad vissza építőt.
$template = Queuety::load_workflow_file( __DIR__ . '/workflows/onboard-user.php' );
Ütemezés
Queuety::schedule( $handler, $payload )
Ismétlődő ütemezés létrehozása. Egy PendingSchedule építőt ad vissza.
Queuety::schedule( 'cleanup_handler' )
->every( '1 hour' )
->on_queue( 'maintenance' );
Queuety::schedule( 'nightly_report' )
->cron( '0 3 * * *' )
->on_queue( 'reports' );
PendingSchedule metódusai
| Metódus | Leírás |
|---|---|
every( string $interval ) |
Intervallumkifejezés beállítása (e.g. 1 hour, 30 minutes) |
cron( string $expression ) |
Ötmezős cron kifejezés beállítása |
overlap( OverlapPolicy $policy ) |
Az átfedési házirend beállítása (alapértelmezés: Allow). Lásd: Ütemezés. |
on_queue( string $queue ) |
Cél-sor beállítása (alapértelmezés: default) |
id() |
Indítás kényszerítése és az ütemezésazonosító visszaadása |
Queuety::scheduler()
A belső Scheduler példány lekérése.
$scheduler = Queuety::scheduler();
$schedules = $scheduler->list();
$scheduler->remove( 'handler_name' );
$count = $scheduler->tick(); // manually trigger
Kezelők regisztrálása
Queuety::register( $name, $class )
Kezelőosztály regisztrálása egy név alatt.
Queuety::register( 'send_email', SendEmailHandler::class );
Queuety::discover_handlers( $directory, $namespace )
Kezelőosztályok automatikus felderítése és regisztrálása egy könyvtárból. Visszaadja a regisztrált kezelők számát.
$count = Queuety::discover_handlers( __DIR__ . '/handlers', 'MyPlugin\\Handlers' );
Queuety::registry()
A kezelőnyilvántartás lekérése.
$registry = Queuety::registry();
Megfigyelhetőség
Queuety::logger()
A naplóbejegyzések lekérdezéséhez használt Logger példány lekérése.
$logger = Queuety::logger();
$logs = $logger->for_job( 42 );
$logs = $logger->for_workflow( 7 );
$logs = $logger->for_handler( 'send_email', 50 );
$logs = $logger->for_event( LogEvent::Failed, 50 );
$logs = $logger->since( new \DateTimeImmutable( '-1 hour' ), 100 );
$count = $logger->purge( 30 );
Queuety::metrics()
A Metrics példány lekérése.
$stats = Queuety::metrics()->handler_stats( 60 ); // last 60 minutes
Queuety::webhook_notifier()
A WebhookNotifier példány lekérése.
$notifier = Queuety::webhook_notifier();
$id = $notifier->register( 'job.buried', 'https://hooks.slack.com/...' );
$webhooks = $notifier->list();
$notifier->remove( $id );
Queuety::rate_limiter()
A RateLimiter példány lekérése.
$limiter = Queuety::rate_limiter();
$limiter->register( 'call_openai', 60, 60 );
Gyorsítótár
Queuety::cache()
Az aktív gyorsítótár-háttérpéldány lekérése.
$cache = Queuety::cache();
$cache->set( 'my_key', 'my_value', 60 );
$value = $cache->get( 'my_key' );
Queuety::set_cache( $cache )
A gyorsítótár-háttér felülbírálása. Egyéni megvalósítás használatához ezt a Queuety::init() előtt hívja meg.
Queuety::set_cache( new RedisCache() );
Queuety::init( $connection );
QUEUETY_CACHE_TTL
Alapértelmezett gyorsítótár-élettartam másodpercben (alapértelmezés: 5). A belső gyorsítótárműveletek alapértelmezett élettartamának módosításához állítsa be ezt a konstansot a wp-config.php fájlban.
define( 'QUEUETY_CACHE_TTL', 10 );
A háttérrendszerekre és az automatikus felismerésre vonatkozó részletekért lásd: Gyorsítótárazás.
Állapotgép eseménynaplója
Queuety::machine_events()
A StateMachineEventLog példány lekérése.
$log = Queuety::machine_events();
Munkafolyamat eseménynaplója
Queuety::workflow_events()
A WorkflowEventLog példány lekérése.
$log = Queuety::workflow_events();
Queuety::workflow_timeline( $workflow_id, $limit = 100, $offset = 0 )
Egy munkafolyamat teljes esemény-idővonalának lekérése. Az eseménysorok azonosító szerinti sorrendben található tömbjét adja vissza.
$events = Queuety::workflow_timeline( $workflow_id );
$next = Queuety::workflow_timeline( $workflow_id, 100, 100 );
Minden eseménysor tartalmazza: id, workflow_id, job_id, parent_event_id, step_index, step_name, step_type, handler, event, queue, attempt, input, output, state_before, state_after, context, artifacts, chunks, error, error_message (a error.message elemből másolva), duration_ms és created_at.
Queuety::workflow_trace( $workflow_id )
Lekérheti a munkafolyamat-hibakereső UI számára normalizált nyomkövetési csomagot.
$trace = Queuety::workflow_trace( $workflow_id );
A nyomkövetés tartalmazza a következőket: workflow, csoportosítva steps, nyers events, jobs, logs, artifacts, chunks, signals és wait_dependencies.
Queuety::trace_input( $input )
Alkalmazásszintű bemenet beállítása az éppen végrehajtás alatt álló munkafolyamat-lépéshez.
Queuety::trace_output( $output )
Alkalmazásszintű kimenet beállítása az éppen végrehajtás alatt álló munkafolyamat-lépéshez.
Queuety::trace_context( array $context )
Alkalmazásszintű környezet egyesítése az éppen végrehajtás alatt álló munkafolyamat-lépéssel.
Queuety::workflow_state_at( $workflow_id, $step_index )
A nyilvános állapot lekérése egy adott munkafolyamat-lépés befejeződése vagy folytatása után, illetve null, ha nem található.
$state = Queuety::workflow_state_at( $workflow_id, 2 );
A felhasználási esetekért és a CLI parancsokért lásd a Munkafolyamat-eseménynaplót.
Belső példányok
Ezek a metódusok belső példányokat adnak vissza speciális felhasználási esetekhez.
| Metódus | Visszatérési érték | Leírás |
|---|---|---|
Queuety::queue() |
Queue |
Várólista-műveletek (igénylés, felszabadítás, eltemetés) |
Queuety::worker() |
Worker |
Munkafolyamat vezérlése |
Queuety::workflow_manager() |
Workflow |
Munkafolyamatok összehangolása |
Queuety::machines() |
StateMachine |
Állapotgép-életciklus kezelése |
Queuety::connection() |
Connection |
Adatbázis-kapcsolat |
Queuety::batch_manager() |
BatchManager |
Köteg-életciklus kezelése (megszakítás, ritkítás) |
Queuety::chunk_store() |
ChunkStore |
Darabok megőrzése adatfolyamot kezelő lépésekhez |
Queuety::machine_events() |
StateMachineEventLog |
Állapotgép-nyomkövetési események és normalizált nyomkövetési csomagok |
Queuety::workflow_events() |
WorkflowEventLog |
Munkafolyamat-nyomkövetési események és normalizált nyomkövetési csomagok |
Queuety::resource_manager() |
ResourceManager |
Munkavégző-felvételi profilok, súlyozott keretek és rendszermemória-vizsgálat |
Queuety::cache() |
Cache |
Gyorsítótár-háttérrendszer példánya |
Tesztelés
Queuety::fake()
Teszteléshez cserélje le a várólistát memóriában működő álimplementációra. Egy QueueFake példányt ad vissza, amely adatbázis-műveletek nélkül rögzíti az elküldött feladatokat. A create_batch() az ál-köteg metaadatait is rögzíti; a láncokat az általuk elküldött egyes feladatokkal ellenőrzi.
$fake = Queuety::fake();
SendEmailJob::dispatch( 'user@example.com', 'Hello', 'Hi!' );
$fake->assert_pushed( SendEmailJob::class );
Queuety::reset()
Minden belső állapot alaphelyzetbe állítása, beleértve a várólista álimplementációját is. Hívja meg a tearDown() metódust.
Queuety::reset();
A QueueFake metódusai
| Metódus | Leírás |
|---|---|
assert_pushed( string $class, ?Closure $callback ) |
Ellenőrzi, hogy egy feladatosztályt elküldtek-e. A választható visszahívás a feladat adatai alapján szűr. |
assert_pushed_times( string $class, int $count ) |
Ellenőrzi, hogy egy feladatot pontosan N alkalommal küldtek-e el. |
assert_not_pushed( string $class ) |
Ellenőrzi, hogy egy feladatot sem küldtek-e el. |
assert_nothing_pushed() |
Ellenőrzi, hogy egyáltalán nem küldtek-e el feladatot. |
assert_batched( ?Closure $callback ) |
Ellenőrzi, hogy elküldtek-e egy köteget. A választható visszahívás a köteg adatai alapján szűr. |
pushed( string $class ) |
Lekéri az osztályhoz rögzített összes elküldést. Tömböt ad vissza. |
batches() |
Lekéri az összes rögzített köteget. Tömböt ad vissza. |
reset() |
Törli az összes rögzített feladatot és köteget. |
A teljes példákért lásd a Tesztelés című részt.
Interfészek
Handler
Egyszerű feladatkezelők interfésze.
interface Handler {
public function handle( array $payload ): void;
public function config(): array;
}
A config() a következőket adhatja vissza: queue, max_attempts, backoff, rate_limit, concurrency_group, concurrency_limit és cost_units.
Step
Munkafolyamat-lépések kezelőinek interfésze.
interface Step {
public function handle( array $state ): array;
public function config(): array;
}
A config() a következőket adhatja vissza: max_attempts, backoff, concurrency_group, concurrency_limit és cost_units.
Contracts\ForEachHandler
Dinamikus, minden ágra kiterjedő kezelők interfésze.
namespace Queuety\Contracts;
interface ForEachHandler {
public function handle_item( array $state, mixed $item, int $index ): array;
public function config(): array;
}
A config() a következőket adhatja vissza: max_attempts, backoff, rate_limit, concurrency_group, concurrency_limit és cost_units.
Contracts\ForEachReducer
Választható reduktor az egyes ágak eredményeihez.
namespace Queuety\Contracts;
interface ForEachReducer {
public function reduce( array $state, array $for_each ): array;
}
Contracts\Compensation
A lépéskompenzációs kezelők interfésze.
namespace Queuety\Contracts;
interface Compensation {
public function handle( array $state ): void;
}
Contracts\RepeatCondition
Hurokfeltételek interfésze; a repeat_until() és repeat_while() ezt használja, ha egy állapotkulcs összehasonlítása nem elegendő.
namespace Queuety\Contracts;
interface RepeatCondition {
public function matches( array $state ): bool;
}
Contracts\StateAction
Várólistára helyezett állapotbelépési műveletek interfésze.
namespace Queuety\Contracts;
interface StateAction {
public function handle( array $state, ?string $event = null, array $event_payload = [], array $payload = [] ): array|string;
}
A $payload az állapotdefinícióban megadott strukturált payload értéket adja át, ha a gépet ilyennel határozták meg. Adja vissza a nyilvános állapot frissítéseit. Ha a műveletnek azonnali állapotátmenetet kell kiváltania, adja meg a _event értéket, és opcionálisan a _event_payload értéket is.
Az állapotműveletek opcionális config() metódust is definiálhatnak. A Queuety a várólistába helyezett belépési műveletkor kiolvassa belőle a concurrency_group, concurrency_limit és cost_units értéket.
Contracts\StateGuard
Állapotgép-átmeneteket ellenőrző feltételek interfésze.
namespace Queuety\Contracts;
interface StateGuard {
public function allows( array $state, array $event_payload, string $event, array $payload = [] ): bool;
}
Contracts\Job
Önállóan elküldhető feladatosztályok interfésze. A feladatosztályok nyilvános tulajdonságokban tárolják a hasznos adatokat, a végrehajtási logikát pedig a handle() metódusban.
namespace Queuety\Contracts;
interface Job {
public function handle(): void;
}
A feladatosztályok opcionálisan definiálhatnak egy middleware() metódust, amely Contracts\Middleware példányok tömbjét adja vissza.
Contracts\Cache
Gyorsítótár-háttérrendszerek interfésze. A beépített megvalósításokért és az egyéni háttérrendszerekért lásd a Gyorsítótárazás című részt.
namespace Queuety\Contracts;
interface Cache {
public function get( string $key ): mixed;
public function set( string $key, mixed $value, int $ttl = 0 ): void;
public function delete( string $key ): void;
public function has( string $key ): bool;
public function add( string $key, mixed $value, int $ttl = 0 ): bool;
public function flush(): void;
}
| Metódus | Visszatérési érték | Leírás |
|---|---|---|
get( $key ) |
mixed |
Érték lekérése. null értéket ad vissza, ha nem található vagy lejárt. |
set( $key, $value, $ttl ) |
void |
Érték tárolása. A 0 TTL azt jelenti, hogy az érték nem jár le. |
delete( $key ) |
void |
Érték eltávolítása. |
has( $key ) |
bool |
Ellenőrzi, hogy létezik-e a kulcs, és nem járt-e le. |
add( $key, $value, $ttl ) |
bool |
Atomi beállítás, ha még nem létezik. true értéket ad vissza, ha az érték be lett állítva. |
flush() |
void |
Minden elem eltávolítása a gyorsítótárból. |
WorkflowEventLog
A munkafolyamat-lépések átmeneteit teljes nyomkövetési környezettel rögzíti. Elérése a Queuety::workflow_events() segítségével.
| Metódus | Visszatérési érték | Leírás |
|---|---|---|
record_event( array $event ) |
void |
Nyers nyomkövetési esemény rögzítése |
record_step_started( ... ) |
void |
step_started esemény rögzítése a bemenettel és a végrehajtás előtti állapottal |
record_step_completed( ... ) |
void |
step_completed esemény rögzítése a bemenettel, a kimenettel, a végrehajtás előtti és utáni állapottal |
record_step_failed( ... ) |
void |
step_failed esemény rögzítése nyomkövetési környezettel és strukturált hibarészletekkel |
get_timeline( int $workflow_id, ?int $limit = 100, int $offset = 0 ) |
array |
Munkafolyamat-nyomkövetési események lekérése sorrendben, választható lapozással |
get_trace( int $workflow_id ) |
array |
Normalizált nyomkövetési csomag lekérése |
get_state_at_step( int $workflow_id, int $step_index ) |
array|null |
Nyilvános állapot lekérése egy adott lépés befejeződése vagy folytatása után |
prune( int $older_than_days ) |
int |
A megadott számú napnál régebbi események törlése. A törölt sorok számát adja vissza. |
StateMachineEventLog
Az állapotgép életciklusának, műveleteinek, átmeneteinek és ellenőrző feltételeinek eseményeit rögzíti, a munkafolyamat-eseményekével megegyező nyomkövetési mezőkkel.
| Metódus | Visszatérési érték | Leírás |
|---|---|---|
record_event( array $event ) |
void |
Nyers gépnyomkövetési esemény rögzítése |
get_timeline( int $machine_id, ?int $limit = 100, int $offset = 0 ) |
array |
Gépnyomkövetési események lekérése sorrendben, választható lapozással |
get_trace( int $machine_id ) |
array |
Normalizált gépnyomkövetési csomag lekérése |
get_state_at_event( int $machine_id, int $event_id ) |
array|null |
Nyilvános állapot lekérése egy adott gépnyomkövetési esemény után |
prune( int $older_than_days ) |
int |
A megadott számú napnál régebbi események törlése. A törölt sorok számát adja vissza. |
Contracts\StreamingStep
Adatfolyamot kezelő lépések interfésze, amelyek darabokból álló adatfolyamot hoznak létre. Minden kiadott érték azonnal az adatbázisba kerül. Lásd az Adatfolyamot kezelő lépések című részt.
namespace Queuety\Contracts;
interface StreamingStep {
public function stream( array $state, array $existing_chunks = [] ): \Generator;
public function on_complete( array $chunks, array $state ): array;
public function config(): array;
}
| Metódus | Leírás |
|---|---|
stream( $state, $existing_chunks ) |
Generátor, amely karakterlánc-darabokat ad ki. Minden kiadás azonnali adatbázis-írást indít. Újrapróbálkozáskor a $existing_chunks a korábban megőrzött darabokat tartalmazza. |
on_complete( $chunks, $state ) |
Az adatfolyam befejezése után hívódik meg. A munkafolyamat állapotával egyesítendő adatokat ad vissza. |
config() |
Választható konfiguráció. A következőket támogatja: max_attempts, backoff, concurrency_group, concurrency_limit és cost_units. |
ChunkStore
Az adatfolyamot kezelő lépések darabjainak megőrzését kezeli. Elérése a Queuety::chunk_store() segítségével.
| Metódus | Visszatérési érték | Leírás |
|---|---|---|
get_chunks( int $job_id ) |
string[] |
Egy feladat összes darabjának lekérése, darabindex szerint rendezve |
append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index ) |
void |
Egyetlen darab hozzáfűzése |
clear_chunks( int $job_id ) |
void |
Egy feladat összes darabjának törlése |
chunk_count( int $job_id ) |
int |
Egy feladat darabjainak megszámlálása |
get_accumulated( int $job_id ) |
string |
Az összes darab összefűzése egy karakterlánccá |
Heartbeat
Statikus segédeszköz aktivitási életjelek küldéséhez lépéskezelőkből és adatfolyamot kezelő lépésekből. Lásd az Életjelek című részt.
| Metódus | Leírás |
|---|---|
Heartbeat::init( int $job_id, Connection $conn ) |
A feladat környezetének beállítása. A munkavégző automatikusan hívja meg. |
Heartbeat::beat( array $progress = [] ) |
Életjelet küld. Frissíti a reserved_at értékét, és opcionálisan előrehaladási adatokat tárol a heartbeat_data helyen. |
Heartbeat::clear() |
Törli a feladat környezetét. A munkavégző automatikusan hívja meg. |
Heartbeat::current_job_id() |
Lekéri az aktuális feladatazonosítót (teszteléshez). |
Contracts\Middleware
Feladat-köztesréteg interfésze. A köztesréteg hagymaszerű folyamatban veszi körül a feladat végrehajtását.
namespace Queuety\Contracts;
interface Middleware {
public function handle( object $job, \Closure $next ): void;
}
Trait-ek
Dispatchable
Statikus elküldési metódusokat biztosít a Contracts\Job interfészt megvalósító feladatosztályokban. A konstruktor argumentumai átadódnak a feladatpéldány létrehozásához.
use Queuety\Contracts\Job;
use Queuety\Dispatchable;
class SendEmailJob implements Job {
use Dispatchable;
// ...
}
// Dispatches a new SendEmailJob instance
SendEmailJob::dispatch( 'user@example.com', 'Subject', 'Body' );
| Metódus | Visszatérési érték | Leírás |
|---|---|---|
dispatch( ...$args ) |
PendingJob |
Példány létrehozása és elküldése |
dispatch_if( bool $condition, ...$args ) |
PendingJob|null |
Csak akkor küldi el, ha $condition értéke igaz |
dispatch_unless( bool $condition, ...$args ) |
PendingJob|null |
Csak akkor küldi el, ha $condition értéke hamis |
with_chain( array $jobs ) |
ChainBuilder |
Lánc indítása ebből a feladatosztályból |
// Conditional dispatch
SendEmailJob::dispatch_if( $user->wants_email, $user->email, 'Hello', 'Hi!' );
SendEmailJob::dispatch_unless( $maintenance_mode, $user->email, 'Hello', 'Hi!' );
// Chain from a job class
FetchDataJob::with_chain( [
new ProcessDataJob(),
new NotifyCompleteJob(),
] )->dispatch();
A feladat tulajdonságai
A feladatosztályok nyilvános tulajdonságokat deklarálhatnak, amelyeket a munkavégző reflexióval olvas be az újrapróbálkozási és időtúllépési viselkedés felülbírálásához.
| Tulajdonság | Típus | Leírás |
|---|---|---|
$tries |
int |
Az újrapróbálkozások maximális száma, mielőtt a feladatot eltemeti a rendszer |
$timeout |
int |
Maximális végrehajtási idő másodpercben (a pcntl szükséges hozzá) |
$backoff |
array |
Növekvő késleltetési értékek az egyes újrapróbálkozásokhoz, másodpercben |
class ImportProductsJob implements Job {
use Dispatchable;
public int $tries = 5;
public int $timeout = 120;
public array $backoff = [ 10, 30, 60, 120 ];
// ...
}
A failed() hook
Ha az összes újrapróbálkozás kimerítése után egy feladatot eltemet a rendszer, a munkavégző meghívja a feladatpéldány failed() metódusát, ha az definiálva van.
public function failed( \Throwable $exception ): void {
error_log( "Job failed: {$exception->getMessage()}" );
}
Beépített köztesrétegek
Minden köztesréteg-osztály megvalósítja a Contracts\Middleware interfészt, és a Queuety\Middleware névtérben található.
| Osztály | Leírás |
|---|---|
RateLimited( int $max, int $window ) |
Végrehajtásonként $max korláttal szabályozza a futtatások számát $window másodpercenként |
Timeout( int $seconds ) |
$seconds másodperc után leállítja a feladatot (a pcntl szükséges hozzá) |
UniqueJob( ?string $key ) |
Adatbázis-zárral akadályozza meg az egyidejű végrehajtást |
WithoutOverlapping( string $key, int $release_after ) |
Automatikus zárexplicitálással akadályozza meg az átfedő végrehajtást |
ThrottlesExceptions( int $max_attempts, int $decay_minutes ) |
Korlátozza a kivételeket, hogy külső szolgáltatás leállásakor elkerülje a feladatok tömeges felhalmozódását |
A használat részleteiért és a példákért lásd a Köztesrétegek című részt.
Felsorolások
Priority
enum Priority: int {
case Low = 0;
case Normal = 1;
case High = 2;
case Urgent = 3;
}
JobStatus
enum JobStatus: string {
case Pending = 'pending';
case Processing = 'processing';
case Completed = 'completed';
case Failed = 'failed';
case Buried = 'buried';
}
WorkflowStatus
enum WorkflowStatus: string {
case Running = 'running';
case Completed = 'completed';
case Failed = 'failed';
case Paused = 'paused';
case WaitingForSignal = 'waiting_for_signal';
case WaitingForWorkflows = 'waiting_for_workflows';
case Cancelled = 'cancelled';
}
StateMachineStatus
enum StateMachineStatus: string {
case Running = 'running';
case WaitingEvent = 'waiting_event';
case Completed = 'completed';
case Failed = 'failed';
case Paused = 'paused';
case Cancelled = 'cancelled';
}
BackoffStrategy
enum BackoffStrategy: string {
case Fixed = 'fixed';
case Linear = 'linear';
case Exponential = 'exponential';
}
ForEachMode
enum ForEachMode: string {
case All = 'all';
case FirstSuccess = 'first_success';
case Quorum = 'quorum';
}
WaitMode
enum WaitMode: string {
case All = 'all';
case Any = 'any';
case Quorum = 'quorum';
}
LogEvent
enum LogEvent: string {
case Started = 'started';
case Completed = 'completed';
case Failed = 'failed';
case Buried = 'buried';
case Retried = 'retried';
case WorkflowStarted = 'workflow_started';
case WorkflowCompleted = 'workflow_completed';
case WorkflowFailed = 'workflow_failed';
case WorkflowPaused = 'workflow_paused';
case WorkflowResumed = 'workflow_resumed';
case WorkflowCancelled = 'workflow_cancelled';
case WorkflowRewound = 'workflow_rewound';
case WorkflowForked = 'workflow_forked';
case WorkflowDeadlineExceeded = 'workflow_deadline_exceeded';
case Debug = 'debug';
}
OverlapPolicy
enum OverlapPolicy: string {
case Allow = 'allow';
case Skip = 'skip';
case Buffer = 'buffer';
}
Az egyes szabályzatok működésének részleteiért lásd az Ütemezés című részt.
Attribútumok
#[QueuetyHandler]
PHP 8 attribútum az automatikus regisztrációhoz. Lásd az PHP attribútumokat.
use Queuety\Attributes\QueuetyHandler;
#[QueuetyHandler( name: 'send_email', queue: 'emails', max_attempts: 5 )]
class SendEmailHandler implements Handler {
// ...
}
CLI-híd
Queuety::cli_command()
A közzétett WP-CLI gyökérparancs nevének visszaadása.
Az alapértelmezett gyökérparancs a queuety. Ha más néven szeretné közzétenni a parancsot, határozza meg a QUEUETY_CLI_COMMAND értékét a Queuety elindulása előtt.
Queuety::cli_command_map()
Szerializálható Queuety CLI katalógus visszaadása azoknak a tesztkörnyezeteknek, amelyeknek a WP-CLI osztályok vizsgálata nélkül kell megismerniük a szállított parancsfelületet.
Minden bejegyzés tartalmazza:
- egy stabil műveletazonosítót
- a relatív CLI útvonalat
- a WP-CLI kezelőosztályt és metódust
- a lehetséges végrehajtási célokat
- az adaptert, amely a nyers CLI argumentumokat végrehajtási tervvé normalizálja
Queuety::resolve_cli_command( $path, $args, $assoc_args )
Egy elemzett CLI hívást a tesztkörnyezet által futtatandó végrehajtási tervvé alakít.
$plan = Queuety::resolve_cli_command(
[ 'workflow', 'export' ],
[ '42' ],
[ 'output' => '/tmp/workflow.json' ]
);
echo $plan['callable']; // Queuety\Queuety::export_workflow_to_file
print_r( $plan['arguments'] );
A végrehajtási terveknél szándékosan figyelembe vesszük az átviteli módot:
- a parancsok normalizált argumentumokkal rendelkező nyilvános PHP hívható elemekre oldódnak fel
- a katalógus előre felsorolja az összes lehetséges végrehajtási célt, míg a
callableés atransportazonosítja az elemzett bemenet által kiválasztott ágat
Ezt a szabványos Queuety mintát használják az ügynökök tesztkörnyezetei: stabil műveletazonosító, nyilvános API cél és az adapter által végzett normalizálás az olyan, kizárólag CLI-hoz kapcsolódó kérdésekhez, mint a várólista alapértelmezései, a JSON hasznosadat-dekódolása, a munkavégzőkészlet elágaztatása és a munkafolyamat-exportfájlok kezelése.