PHP API
Усім в Queuety можна скористатися через статичний клас Queuety\Queuety. Це єдина точка входу для запуску завдань, побудови робочих процесів, керування машинами станів і чергами, а також доступу до внутрішніх підсистем.
use Queuety\Queuety;
Самостійний доступ до ядра
Середовище виконання Queuety, що використовує базу даних, можна ініціалізувати поза WordPress.
Цей самостійний режим призначено для ядра оркестрації:
- завдання та воркери (для пулів воркерів також потрібні константи
DB_HOST,DB_NAME,DB_USERіDB_PASSWORD, див. нижче) - робочі процеси й сигнали
- машини станів і події
- артефакти, журнали, метрики та розклади
Це не замінює адаптер WordPress. Queuety::on_action(),
replace_wp_cron(), ініціалізація плагіна та справжній шлях реєстрації wp queuety ...
і далі потребують WordPress.
Queuety::init( Connection $conn )
Ініціалізуйте Queuety за допомогою прямого з’єднання з базою даних.
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 );
Необов’язковий аргумент table_prefix змінює частину Queuety у назвах таблиць середовища виконання для кожного з’єднання:
$conn = new Connection(
host: '127.0.0.1:3306',
dbname: 'wordpress',
user: 'root',
password: 'secret',
prefix: 'wp_',
table_prefix: 'themequeue_',
);
Так таблиці, як-от queuety_jobs, зіставляються з wp_themequeue_jobs.
Queuety::ensure_schema()
Переконайтеся, що таблиці середовища виконання Queuety існують для активного з’єднання.
Queuety::init( $conn );
Queuety::ensure_schema();
Повну модель і межі WordPress описано в розділі Самостійне використання.
Завдання
Queuety::dispatch( $handler, $payload )
Запустити завдання. Приймає рядок із назвою обробника або екземпляр Contracts\Job. Повертає об’єкт-побудовник PendingJob для послідовного налаштування параметрів.
// 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();
Якщо передано екземпляр Contracts\Job, серіалізатор отримує FQCN як назву обробника, а відкриті властивості — як дані. Параметр $payload ігнорується, якщо $handler — це екземпляр Job.
Методи PendingJob
| Метод | Опис |
|---|---|
on_queue( string $queue ) |
Задати цільову чергу |
with_priority( Priority $priority ) |
Задати рівень пріоритету |
delay( int $seconds ) |
Час затримки перед доступністю завдання |
max_attempts( int $max ) |
Максимальна кількість спроб повтору |
rate_limit( int $max, int $window ) |
Обмеження частоти: максимум виконань за проміжок у секундах |
unique() |
Запобігти повторному запуску того самого завдання |
after( int $job_id ) |
Дочекатися завершення іншого завдання |
concurrency_group( string $group, int $limit ) |
Додати завдання до спільної групи допуску воркерів |
cost_units( int $units ) |
Додати до запуску метадані про відносну вартість виконання |
id() |
Примусово запустити й повернути ID завдання |
concurrency_group() і cost_units() є частиною ширшої моделі Керування ресурсами, тож один запуск може брати участь в обмеженнях паралельності за кількістю, зважених бюджетах черги чи провайдера та евристиках допуску воркерів.
Queuety::dispatch_sync( $handler, $payload )
Синхронно виконати завдання в поточному процесі, не додаючи його до черги.
// 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' ] );
Метод handle() завдання викликається безпосередньо. Рядок у базі даних не створюється, воркер не залучається, проміжне програмне забезпечення не застосовується.
Queuety::batch( $jobs )
Запустити кілька завдань одним багаторядковим оператором INSERT.
$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 )
Створити BatchBuilder для запуску групи завдань із функціями зворотного виклику та відстеженням перебігу. Докладніше див. у розділі Пакетна обробка.
$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
| Метод | Повертає | Опис |
|---|---|---|
name( string $name ) |
self |
Задати назву пакета |
on_queue( string $queue ) |
self |
Задати чергу для всіх завдань пакета |
then( string $handler_class ) |
self |
Обробник викликається після успішного завершення пакета |
catch( string $handler_class ) |
self |
Обробник викликається, якщо перше завдання пакета завершується помилкою |
finally( string $handler_class ) |
self |
Обробник викликається після завершення пакета (успішного або з помилкою) |
allow_failures() |
self |
Дозволити виклик функції зворотного виклику then навіть у разі помилок |
dispatch() |
Batch |
Створити запис пакета й усі завдання; повертає об’єкт-значення Batch |
Queuety::find_batch( $id )
Знайти пакет за ID. Повертає об’єкт-значення Batch або null.
$batch = Queuety::find_batch( $batch_id );
if ( null !== $batch ) {
echo $batch->progress() . '% complete';
}
Керування воркерами
Queuety::run_worker( $queue = 'default', $once = false )
Безпосередньо запустити один цикл воркера.
Queuety::run_worker( 'default' );
Queuety::run_worker( 'critical,default', true );
Queuety::run_worker_pool( $workers, $queue = 'default' )
Безпосередньо запустити пул воркерів фіксованого розміру. Потрібні розширення pcntl і posix. Для дочірніх з’єднань пул зчитує константи WordPress DB_HOST, DB_NAME, DB_USER і DB_PASSWORD (лише префікси таблиць беруться з ініціалізованого з’єднання), тому визначте їх, якщо запускаєте пул поза WordPress (#5).
Queuety::run_worker_pool( 4, 'providers' );
Queuety::run_auto_scaling_worker_pool( $min_workers, $max_workers, $queue = 'default' )
Безпосередньо запустити адаптивний пул воркерів. Вимоги ті самі, що й для run_worker_pool().
Queuety::run_auto_scaling_worker_pool( 2, 6, 'providers' );
Об’єкт-значення Batch
| Метод/властивість | Тип | Опис |
|---|---|---|
$id |
int |
ID пакета |
$name |
string|null |
Необов’язкова назва пакета |
$total_jobs |
int |
Загальна кількість завдань у пакеті |
$pending_jobs |
int |
Завдання, що ще очікують виконання |
$failed_jobs |
int |
Кількість завдань із помилками |
$failed_job_ids |
array |
ID завдань із помилками |
$created_at |
DateTimeImmutable |
Час створення пакета |
$finished_at |
DateTimeImmutable|null |
Час завершення пакета |
$cancelled_at |
DateTimeImmutable|null |
Час скасування пакета |
progress() |
int |
Відсоток виконання (0–100) |
finished() |
bool |
Чи завершився пакет |
has_failures() |
bool |
Чи завершилося якесь завдання помилкою |
cancelled() |
bool |
Чи було скасовано пакет |
Queuety::chain( $jobs )
Створити ChainBuilder для послідовного виконання завдань. Кожне завдання залежить від успішного завершення попереднього.
$first_job_id = Queuety::chain( [
new FetchDataJob( $url ),
new ProcessDataJob(),
new NotifyCompleteJob(),
] )
->on_queue( 'pipeline' )
->catch( ChainFailedHandler::class )
->dispatch();
Методи ChainBuilder
| Метод | Повертає | Опис |
|---|---|---|
on_queue( string $queue ) |
self |
Задати чергу для всіх завдань у ланцюжку |
catch( string $handler_class ) |
self |
Обробник викликається, коли будь-яке завдання в ланцюжку назавжди переміщується до сховища |
dispatch() |
int |
Створити всі завдання з послідовними залежностями; повертає ID першого завдання |
Queuety::retry( $job_id )
Повторити конкретне завдання. Скидає його статус на pending.
Queuety::retry( 42 );
Queuety::retry_buried()
Повторити всі відкладені назавжди завдання. Повертає кількість повторно запущених завдань.
$count = Queuety::retry_buried();
Queuety::purge( $older_than_days )
Очистити завершені завдання. Повертає кількість очищених завдань.
$count = Queuety::purge(); // uses QUEUETY_RETENTION_DAYS
$count = Queuety::purge( 30 ); // purge jobs older than 30 days
Черги
Queuety::stats( $queue )
Отримати кількість завдань у кожному статусі.
$stats = Queuety::stats(); // all queues
$stats = Queuety::stats( 'emails' ); // specific queue
// ['pending' => 12, 'processing' => 3, 'completed' => 150, 'failed' => 2, 'buried' => 1]
Queuety::buried( $queue )
Отримати всі відкладені назавжди завдання.
$jobs = Queuety::buried();
$jobs = Queuety::buried( 'emails' );
Queuety::pause( $queue )
Призупинити чергу, щоб воркери її пропускали.
Queuety::pause( 'emails' );
Queuety::resume( $queue )
Відновити призупинену чергу.
Queuety::resume( 'emails' );
Queuety::is_paused( $queue )
Перевірити, чи призупинено чергу.
if ( Queuety::is_paused( 'emails' ) ) {
// Queue is paused
}
Робочі процеси
Queuety::workflow( $name )
Почати створення робочого процесу. Повертає WorkflowBuilder.
$workflow_id = Queuety::workflow( 'generate_report' )
->then( FetchDataHandler::class )
->then( CallLLMHandler::class )
->then( FormatOutputHandler::class )
->dispatch( [ 'user_id' => 42 ] );
Методи WorkflowBuilder
| Метод | Опис |
|---|---|
then( string $handler_class, ?string $name = null ) |
Додати послідовний крок. Необов’язкова назва для умовного розгалуження. |
parallel( array $handler_classes ) |
Додати групу паралельних кроків. |
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 ) |
Додати динамічний крок циклу для кожного елемента, який створює окреме завдання гілки для кожного елемента в стані робочого процесу. |
repeat_while( string $target_step, ?string $state_key = null, mixed $expected = true, ?string $name = null, ?string $condition_class = null, ?int $max_iterations = null ) |
Повернутися до попереднього іменованого кроку, доки збігається відкритий ключ стану або клас RepeatCondition. |
repeat_until( string $target_step, ?string $state_key = null, mixed $expected = true, ?string $name = null, ?string $condition_class = null, ?int $max_iterations = null ) |
Повернутися до попереднього іменованого кроку, доки не збігатиметься відкритий ключ стану або клас RepeatCondition. |
run_workflow( string $workflow_name, WorkflowBuilder $sub_builder ) |
Додати крок запуску робочого процесу. |
delay( int $seconds = 0, int $minutes = 0, int $hours = 0, int $days = 0 ) |
Додати крок тривалої затримки. Усі тривалості підсумовуються. |
wait_for_signal( string $name, ?string $result_key = null, ?string $step_name = null, array $match_payload = [], ?string $correlation_key = null ) |
Додати крок очікування сигналу. Необов’язкові match_payload та correlation_key дають змогу ігнорувати сторонні події. |
wait_for_signals( array $signal_names, WaitMode $mode = WaitMode::All, ?string $result_key = null, ?string $name = null, array $match_payload = [], ?string $correlation_key = null ) |
Очікувати на всі або будь-які з кількох сигналів, з необов’язковою перевіркою даних і кореляцією. |
wait_for_approval( string $signal_name = 'approval', ?string $result_key = 'approval', ?string $name = null, array $match_payload = [], ?string $correlation_key = null ) |
Очікувати на сигнал схвалення від людини. |
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 ) |
Очікувати на схвалення або відхилення та зберігати структуровані дані рішення. |
wait_for_input( string $signal_name = 'input', ?string $result_key = 'input', ?string $name = null, array $match_payload = [], ?string $correlation_key = null ) |
Очікувати на сигнал введення даних від людини. |
wait_for_workflow( int|string $workflow, ?string $result_key = null, ?string $name = null ) |
Очікувати завершення одного робочого процесу за його ID або за ID, збереженим у відкритому стані. |
wait_for_workflows( array|string $workflows, WaitMode $mode = WaitMode::All, ?string $result_key = null, ?string $name = null, ?int $quorum = null ) |
Очікувати завершення всіх, будь-яких або кворуму з кількох робочих процесів. |
wait_for_workflow_group( string $group_key, WaitMode $mode = WaitMode::All, ?int $quorum = null, ?string $result_key = null, ?string $name = null ) |
Очікувати на раніше запущену групу робочих процесів за внутрішнім ключем групи. |
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 ) |
Запускати окремий робочий процес верхнього рівня для кожного елемента середовища виконання; за бажанням позначити запущену групу довговічним ключем групи. |
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 ) |
Псевдонім 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 ) |
Псевдонім 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 ) |
Псевдонім wait_for_workflow_group() для агентських сценаріїв. |
version( string $version ) |
Позначити визначення робочого процесу версією рівня застосунку для перегляду й експорту. |
idempotency_key( string $key ) |
Зробити запуск робочого процесу довговічним за ключем, наданим викликачем. Повторний запуск із тим самим ключем повертає початковий ID робочого процесу. |
max_transitions( int $max ) |
Завершити робочий процес помилкою, якщо завершиться більше ніж $max переходів між кроками. |
max_for_each_items( int $max ) |
Завершити робочий процес помилкою, якщо крок for_each() спробує розгорнути більше ніж $max елементів. |
max_state_bytes( int $bytes ) |
Завершити робочий процес помилкою, якщо відкритий стан перевищить $bytes. |
max_cost_units( int $max ) |
Завершити робочий процес помилкою, якщо завершені кроки використають понад $max одиниць вартості. |
max_started_workflows( int $max ) |
Завершити робочий процес помилкою, перш ніж запустити понад $max дочірніх робочих процесів. |
compensate_with( string $handler_class ) |
Додати обробник компенсації до останнього доданого кроку. |
compensate_on_failure() |
Автоматично запускати компенсації завершених кроків у разі помилки робочого процесу. |
on_queue( string $queue ) |
Задати чергу для всіх кроків. |
with_priority( Priority $priority ) |
Задати пріоритет для всіх кроків. |
max_attempts( int $max ) |
Задати максимальну кількість повторних спроб для кожного кроку. |
on_cancel( string $handler_class ) |
Зареєструвати клас обробника очищення для скасування. Докладніше: Скасування. |
prune_state_after( int $steps = 2 ) |
Увімкнути автоматичне очищення стану. Видаляє результати кроків, старіші за порогове значення. Докладніше: Очищення стану. |
must_complete_within( int $seconds = 0, int $minutes = 0, int $hours = 0, int $days = 0 ) |
Задати крайній термін для робочого процесу. Усі тривалості підсумовуються. Докладніше: Крайні терміни. |
on_deadline( string $handler_class ) |
Зареєструвати обробник, який викликається після перевищення крайнього терміну робочого процесу. Докладніше: Крайні терміни. |
dispatch( array $initial_payload = [] ) |
Запустити робочий процес. Повертає його ID. |
Queuety::workflow_status( $workflow_id )
Отримати статус робочого процесу. Повертає об’єкт WorkflowState або null.
$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']
WorkflowState також містить:
waiting_for: назви сигналів або ID робочих процесів, які наразі блокують виконанняcurrent_step_name: налаштовану назву поточного кроку, якщо вона єwait_mode: поточний режим очікування (all,anyабоquorum) під час блокуванняwait_details: додаткові дані для перегляду, наприклад зіставлені та незавершені блокувальникиdefinition_version: значення, задане черезversion(), якщо воно єdefinition_hash: детермінований хеш визначення робочого процесуidempotency_key: довговічний ключ запуску, якщо він єbudget: налаштовані обмеження та поточне використання, якщо ввімкнено захисні механізмиartifact_count: кількість збережених артефактів робочого процесуartifact_keys: ключі збережених артефактів для швидкого перегляду
Queuety::retry_workflow( $workflow_id )
Повторити запуск робочого процесу з помилкою, починаючи з кроку, на якому вона сталася.
Queuety::retry_workflow( $workflow_id );
Викликає RuntimeException, якщо після помилки для робочого процесу вже виконано компенсацію.
Queuety::pause_workflow( $workflow_id )
Призупинити робочий процес, що виконується. Уже поставлена в чергу робота для поточного кроку може завершитися, але наступні кроки не ставитимуться в чергу, доки ви не відновите робочий процес.
Queuety::pause_workflow( $workflow_id );
Queuety::resume_workflow( $workflow_id )
Відновити призупинений робочий процес.
Queuety::resume_workflow( $workflow_id );
Queuety::cancel_workflow( $workflow_id )
Скасувати робочий процес і запустити всі зареєстровані обробники очищення. Задати статус cancelled і перемістити всі очікувані завдання та завдання в обробці для цього робочого процесу до відкладених.
Queuety::cancel_workflow( $workflow_id );
Викликає RuntimeException, якщо робочий процес уже має статус completed або cancelled. Докладніше див. у розділі Скасування.
Queuety::rewind_workflow( $workflow_id, $to_step )
Повернути робочий процес до попереднього кроку та повторно запустити його звідти. Стан відновлюється з журналу подій state_after для вказаного кроку, а наступний крок ставиться в чергу.
Queuety::rewind_workflow( $workflow_id, to_step: 1 );
Викликає RuntimeException, якщо для цільового кроку немає записаного стану. Потрібен Журнал подій робочого процесу. Докладніше див. у розділі Подорож у часі.
Queuety::fork_workflow( $workflow_id )
Створити незалежну копію робочого процесу, що виконується, з його поточним станом. Повертає ID нового робочого процесу.
$forked_id = Queuety::fork_workflow( $workflow_id );
Розгалужений робочий процес отримує власний ID і виконується незалежно від початкового. Докладніше: Створення гілки.
Queuety::export_workflow( $workflow_id )
Експортувати повну історію виконання робочого процесу як масив, серіалізований у JSON. Містить рядок робочого процесу, усі завдання, події робочого процесу та записи журналу.
$data = Queuety::export_workflow( $workflow_id );
Докладніше про структуру JSON та варіанти використання див. у розділі Експорт та повторний запуск.
Queuety::replay_workflow( $data )
Повторно запустити експортований робочий процес у поточному середовищі. Створює новий робочий процес із експортованих даних і повертає його ID.
$new_id = Queuety::replay_workflow( $data );
Докладніше див. у розділі Експорт та повторний запуск.
Queuety::replace_wp_cron()
Замінити вбудовану систему WordPress cron планувальником Queuety. Перехоплює wp_schedule_event(), wp_unschedule_event() та wp_get_scheduled_event() і спрямовує повторювані події через робочий процес Queuety.
Queuety::replace_wp_cron();
Працює лише за умови завантаження WordPress. Докладніше: WP-Cron Заміна.
Queuety::restore_wp_cron()
Відновити стандартну поведінку WordPress cron, видаливши перехоплювачі мосту Queuety cron.
Queuety::restore_wp_cron();
Докладніше про обмеження див. у розділі WP-Cron Заміна.
Queuety::signal( $workflow_id, $name, $data )
Надіслати сигнал робочому процесу. Якщо робочий процес наразі очікує на цей сигнал, він відновиться негайно. Інакше сигнал буде збережено та оброблено, коли робочий процес дійде до відповідного кроку wait_for_signal.
Queuety::signal( $workflow_id, 'approved' );
// With data merged into workflow state
Queuety::signal( $workflow_id, 'approved', [
'approved_by' => 'admin@example.com',
] );
Докладніше: Сигнали робочого процесу.
Queuety::approve_workflow( $workflow_id, $data = [], $signal_name = 'approval' )
Семантична обгортка навколо signal() для потоків схвалення.
Queuety::approve_workflow( $workflow_id, [
'reviewer' => 'editor@example.com',
] );
Queuety::reject_workflow( $workflow_id, $data = [], $signal_name = 'rejected' )
Семантична обгортка навколо signal() для потоків відхилення.
Queuety::reject_workflow( $workflow_id, [
'reason' => 'needs citations',
] );
Queuety::submit_workflow_input( $workflow_id, $data = [], $signal_name = 'input' )
Семантична обгортка навколо signal() для структурованого введення даних людиною.
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 = [] )
Зберегти один довговічний артефакт для робочого процесу.
Queuety::put_artifact(
$workflow_id,
'research_brief',
[ 'summary' => 'done' ],
'json',
metadata: [ 'source' => 'agent' ],
);
Queuety::put_current_artifact( $artifact_key, $content, $kind = 'json', $metadata = [] )
Зберегти один артефакт для кроку робочого процесу, який виконується зараз.
Queuety::put_current_artifact(
'draft',
[ 'status' => 'ready' ],
'json',
);
Queuety::workflow_artifact( $workflow_id, $artifact_key )
Отримати один збережений артефакт робочого процесу.
$artifact = Queuety::workflow_artifact( $workflow_id, 'draft' );
Queuety::workflow_artifacts( $workflow_id, $include_content = false )
Переглянути список артефактів робочого процесу.
$artifacts = Queuety::workflow_artifacts( $workflow_id );
Queuety::delete_workflow_artifact( $workflow_id, $artifact_key )
Видалити один артефакт робочого процесу.
Queuety::delete_workflow_artifact( $workflow_id, 'draft' );
Автомати станів
Queuety::machine( $name )
Почати створення автомата станів. Повертає StateMachineBuilder.
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
| Метод | Опис |
|---|---|
initial( string $state_name ) |
Явно задати початковий стан. |
state( string $state_name, ?StateMachineStatus $terminal_status = null ) |
Почати налаштування одного стану. |
action( string|array $action_class ) |
Зареєструвати дію входу в поточний стан, яку буде поставлено в чергу. Передайте назву класу або структуроване визначення з class, payload та config; payload отримує StateAction::handle() як $payload. |
on( string $event, string $target, string|array|null $guard_class = null, ?string $name = null ) |
Зареєструвати перехід за однією подією з поточного стану. Для умови переходу приймається та сама структурована форма, що й для action(). |
on_queue( string $queue ) |
Задати чергу для дій зі станами, які ставляться в чергу. |
with_priority( Priority $priority ) |
Задати пріоритет для дій зі станами, які ставляться в чергу. |
max_attempts( int $max_attempts ) |
Задати ліміт повторних спроб для дій зі станами, які ставляться в чергу. |
version( string $version ) |
Позначити визначення автомата версією рівня застосунку. |
idempotency_key( string $key ) |
Зробити запуск довговічним за ключем, наданим викликачем. Повторний запуск із тим самим ключем повертає початковий ID автомата. |
dispatch( array $initial_state = [] ) |
Запустити автомат і повернути його ID. |
Класи дій входу в стан також можуть мати необов’язковий метод config() із тими самими ключами політики ресурсів, що підтримуються кроками робочого процесу: concurrency_group, concurrency_limit та cost_units.
Queuety::dispatch_state_machine_definition( $definition, $initial_state = [], $options = [] )
Запустити серіалізований пакет визначення автомата станів. У більшості коду застосунку варто використовувати machine()->dispatch(). Цей низькорівневий метод потрібен, щоб адаптери могли запускати автомат без прямої залежності від конструктора.
$machine_id = Queuety::dispatch_state_machine_definition(
$definition,
[ 'thread_id' => 42 ],
[ 'idempotency_key' => 'thread:42' ],
);
Queuety::machine_status( $machine_id )
Отримати поточний статус автомата. Повертає об’єкт StateMachineState або null.
$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
StateMachineState також містить:
definition_versiondefinition_hashidempotency_keyerror_messagecurrent_actionterminal_status
Queuety::machine_event( $machine_id, $event_name, $payload = [] )
Надіслати одну зовнішню подію автомату, що очікує.
Queuety::machine_event(
$machine_id,
'approve',
[
'approved' => true,
'reviewer' => 'editor@example.com',
]
);
Дані доступні для умов переходів і об’єднуються з відкритим станом автомата, коли подію прийнято.
Queuety::list_machines( $limit = 50, $status = null )
Переглянути збережені рядки автоматів для перевірки.
$machines = Queuety::list_machines();
$waiting = Queuety::list_machines( status: 'waiting_event' );
Кожен рядок містить ID автомата, назву, статус, поточний стан, версію визначення, хеш визначення, ключ ідемпотентності та часові позначки життєвого циклу.
Queuety::machine_timeline( $machine_id, $limit = 100, $offset = 0 )
Отримати довговічну хронологію подій одного автомата.
$timeline = Queuety::machine_timeline( $machine_id );
$page_two = Queuety::machine_timeline( $machine_id, 100, 100 );
Кожен рядок події містить: 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 та created_at.
Queuety::machine_trace( $machine_id )
Отримати нормалізований пакет трасування для налагоджувача автоматів станів UI.
$trace = Queuety::machine_trace( $machine_id );
Трасування містить machine, згруповані states, необроблені events, корельовану дію jobs, logs, artifacts та chunks.
Queuety::machine_state_at( $machine_id, $event_id )
Отримати відкритий стан після конкретної події трасування автомата або null, якщо його не знайдено.
$state = Queuety::machine_state_at( $machine_id, $event_id );
Модель життєвого циклу та типові способи використання описано в розділі Автомати станів.
Шаблони робочих процесів
Queuety::define_workflow( $name )
Створити конструктор робочого процесу для використання як шаблону.
$builder = Queuety::define_workflow( 'onboarding' )
->then( CreateAccountHandler::class )
->then( SendWelcomeEmailHandler::class );
Queuety::register_workflow_template( $builder )
Зареєструвати шаблон робочого процесу.
Queuety::register_workflow_template( $builder );
Queuety::run_workflow( $name, $payload, $options = [] )
Запустити зареєстрований шаблон робочого процесу за назвою. Повертає ID робочого процесу.
$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, якщо шаблон не зареєстровано.
Queuety::dispatch_workflow_definition( $definition, $payload, $options = [] )
Запустити робочий процес із пакета визначення середовища виконання.
У більшості коду застосунку варто використовувати run_workflow() або workflow()->dispatch(). Цей низькорівневий метод потрібен, щоб адаптери повторного використання могли запускати серіалізоване визначення робочого процесу без прямої залежності від внутрішніх методів менеджера робочих процесів.
$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, яка запускає робочий процес.
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}",
);
$workflow може бути одним із таких варіантів:
- назва зареєстрованого шаблону робочого процесу
- вбудований
WorkflowBuilder
map, when та idempotency_key виконуються синхронно в поточному запиті WordPress. Лише кроки робочого процесу виконуються асинхронно після запуску.
Повний посібник див. у розділі Тригери дій.
Queuety::workflow_templates()
Отримати реєстр шаблонів робочих процесів.
$registry = Queuety::workflow_templates();
$template = $registry->get( 'onboarding' ); // WorkflowTemplate or null
Queuety::load_workflows( $directory, $recursive )
Завантажити й зареєструвати всі файли робочих процесів із каталогу. Кожен файл .php має повертати WorkflowBuilder. Повертає кількість зареєстрованих робочих процесів.
$count = Queuety::load_workflows( __DIR__ . '/workflows/' );
$count = Queuety::load_workflows( __DIR__ . '/workflows/', recursive: true );
Queuety::load_workflow_file( $file_path )
Завантажити й зареєструвати один файл робочого процесу. Повертає WorkflowTemplate або null, якщо файл не повертає конструктор.
$template = Queuety::load_workflow_file( __DIR__ . '/workflows/onboard-user.php' );
Планування
Queuety::schedule( $handler, $payload )
Створити повторюваний розклад. Повертає конструктор PendingSchedule.
Queuety::schedule( 'cleanup_handler' )
->every( '1 hour' )
->on_queue( 'maintenance' );
Queuety::schedule( 'nightly_report' )
->cron( '0 3 * * *' )
->on_queue( 'reports' );
Методи PendingSchedule
| Метод | Опис |
|---|---|
every( string $interval ) |
Задати вираз інтервалу (e.g, 1 hour, 30 minutes) |
cron( string $expression ) |
Задати вираз cron із 5 полів |
overlap( OverlapPolicy $policy ) |
Задати політику перекриття (за замовчуванням: Allow). Докладніше: Планування. |
on_queue( string $queue ) |
Задати цільову чергу (за замовчуванням: default) |
id() |
Примусово запустити й повернути ID розкладу |
Queuety::scheduler()
Отримати внутрішній екземпляр Scheduler.
$scheduler = Queuety::scheduler();
$schedules = $scheduler->list();
$scheduler->remove( 'handler_name' );
$count = $scheduler->tick(); // manually trigger
Реєстрація обробників
Queuety::register( $name, $class )
Зареєструвати клас обробника під назвою.
Queuety::register( 'send_email', SendEmailHandler::class );
Queuety::discover_handlers( $directory, $namespace )
Автоматично знайти й зареєструвати класи обробників із каталогу. Повертає кількість зареєстрованих обробників.
$count = Queuety::discover_handlers( __DIR__ . '/handlers', 'MyPlugin\\Handlers' );
Queuety::registry()
Отримати реєстр обробників.
$registry = Queuety::registry();
Спостережуваність
Queuety::logger()
Отримати екземпляр Logger для запитів до записів журналу.
$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()
Отримати екземпляр Metrics.
$stats = Queuety::metrics()->handler_stats( 60 ); // last 60 minutes
Queuety::webhook_notifier()
Отримати екземпляр WebhookNotifier.
$notifier = Queuety::webhook_notifier();
$id = $notifier->register( 'job.buried', 'https://hooks.slack.com/...' );
$webhooks = $notifier->list();
$notifier->remove( $id );
Queuety::rate_limiter()
Отримати екземпляр RateLimiter.
$limiter = Queuety::rate_limiter();
$limiter->register( 'call_openai', 60, 60 );
Кеш
Queuety::cache()
Отримати екземпляр активного бекенду кешу.
$cache = Queuety::cache();
$cache->set( 'my_key', 'my_value', 60 );
$value = $cache->get( 'my_key' );
Queuety::set_cache( $cache )
Перевизначити бекенд кешу. Викличте це перед Queuety::init(), щоб використовувати власну реалізацію.
Queuety::set_cache( new RedisCache() );
Queuety::init( $connection );
QUEUETY_CACHE_TTL
Стандартний TTL кешу в секундах (за замовчуванням: 5). Змініть цю константу в wp-config.php, щоб змінити стандартний TTL для внутрішніх операцій кешу.
define( 'QUEUETY_CACHE_TTL', 10 );
Докладніше про бекенди й автоматичне визначення див. у розділі Кешування.
Журнал подій автоматів станів
Queuety::machine_events()
Отримати екземпляр StateMachineEventLog.
$log = Queuety::machine_events();
Журнал подій робочих процесів
Queuety::workflow_events()
Отримати екземпляр WorkflowEventLog.
$log = Queuety::workflow_events();
Queuety::workflow_timeline( $workflow_id, $limit = 100, $offset = 0 )
Отримати повну хронологію подій робочого процесу. Повертає масив рядків подій, упорядкованих за ID.
$events = Queuety::workflow_timeline( $workflow_id );
$next = Queuety::workflow_timeline( $workflow_id, 100, 100 );
Кожен рядок події містить: 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 (скопійовано з error.message), duration_ms і created_at.
Queuety::workflow_trace( $workflow_id )
Отримайте нормалізований пакет трасування для налагоджувача робочих процесів UI.
$trace = Queuety::workflow_trace( $workflow_id );
Трасування містить workflow, згруповані steps, необроблені events, jobs, logs, artifacts, chunks, signals і wait_dependencies.
Queuety::trace_input( $input )
Задайте вхідні дані рівня застосунку для поточного кроку робочого процесу, що виконується.
Queuety::trace_output( $output )
Задайте вихідні дані рівня застосунку для поточного кроку робочого процесу, що виконується.
Queuety::trace_context( array $context )
Об’єднайте контекст рівня застосунку з поточним кроком робочого процесу, що виконується.
Queuety::workflow_state_at( $workflow_id, $step_index )
Отримайте публічний стан після завершення або відновлення певного кроку робочого процесу або null, якщо його не знайдено.
$state = Queuety::workflow_state_at( $workflow_id, 2 );
Перегляньте Журнал подій робочого процесу, щоб ознайомитися з прикладами використання та командами CLI.
Внутрішні екземпляри
Ці методи повертають внутрішні екземпляри для розширених сценаріїв використання.
| Метод | Повертає | Опис |
|---|---|---|
Queuety::queue() |
Queue |
Операції з чергою (отримання, звільнення, поховання завдань) |
Queuety::worker() |
Worker |
Керування процесами воркерів |
Queuety::workflow_manager() |
Workflow |
Оркестрація робочих процесів |
Queuety::machines() |
StateMachine |
Керування життєвим циклом автомата станів |
Queuety::connection() |
Connection |
Підключення до бази даних |
Queuety::batch_manager() |
BatchManager |
Керування життєвим циклом пакетів завдань (скасування, очищення) |
Queuety::chunk_store() |
ChunkStore |
Збереження фрагментів для потокових кроків |
Queuety::machine_events() |
StateMachineEventLog |
Події трасування автомата станів і нормалізовані пакети трасування |
Queuety::workflow_events() |
WorkflowEventLog |
Події трасування робочого процесу і нормалізовані пакети трасування |
Queuety::resource_manager() |
ResourceManager |
Профілі допуску воркерів, зважені ліміти та перевірка системної пам’яті |
Queuety::cache() |
Cache |
Екземпляр бекенду кешу |
Тестування
Queuety::fake()
Замініть чергу на фіктивну чергу в пам’яті для тестування. Повертає екземпляр QueueFake, який реєструє відправлені завдання, не звертаючись до бази даних. create_batch() також реєструє метадані фіктивних пакетів завдань; ланцюжки перевіряються за окремими завданнями, які вони відправляють.
$fake = Queuety::fake();
SendEmailJob::dispatch( 'user@example.com', 'Hello', 'Hi!' );
$fake->assert_pushed( SendEmailJob::class );
Queuety::reset()
Скиньте весь внутрішній стан, зокрема фіктивну чергу. Викликайте в tearDown().
Queuety::reset();
Методи QueueFake
| Метод | Опис |
|---|---|
assert_pushed( string $class, ?Closure $callback ) |
Перевірити, що завдання певного класу було додано до черги. Необов’язкова функція зворотного виклику фільтрує дані завдання. |
assert_pushed_times( string $class, int $count ) |
Перевірити, що завдання додали до черги рівно N разів. |
assert_not_pushed( string $class ) |
Перевірити, що завдання жодного разу не додавали до черги. |
assert_nothing_pushed() |
Перевірити, що до черги взагалі не додавали завдань. |
assert_batched( ?Closure $callback ) |
Перевірити, що пакет завдань було відправлено. Необов’язкова функція зворотного виклику фільтрує дані пакета. |
pushed( string $class ) |
Отримати всі зареєстровані відправлення для класу. Повертає масив. |
batches() |
Отримати всі зареєстровані пакети завдань. Повертає масив. |
reset() |
Очистити всі зареєстровані завдання та пакети завдань. |
Див. Тестування, щоб переглянути повні приклади.
Інтерфейси
Handler
Інтерфейс для простих обробників завдань.
interface Handler {
public function handle( array $payload ): void;
public function config(): array;
}
config() може повертати queue, max_attempts, backoff, rate_limit, concurrency_group, concurrency_limit і cost_units.
Step
Інтерфейс для обробників кроків робочого процесу.
interface Step {
public function handle( array $state ): array;
public function config(): array;
}
config() може повертати max_attempts, backoff, concurrency_group, concurrency_limit і cost_units.
Contracts\ForEachHandler
Інтерфейс для обробників динамічних гілок циклу for-each.
namespace Queuety\Contracts;
interface ForEachHandler {
public function handle_item( array $state, mixed $item, int $index ): array;
public function config(): array;
}
config() може повертати max_attempts, backoff, rate_limit, concurrency_group, concurrency_limit і cost_units.
Contracts\ForEachReducer
Необов’язковий редуктор для результатів for-each.
namespace Queuety\Contracts;
interface ForEachReducer {
public function reduce( array $state, array $for_each ): array;
}
Contracts\Compensation
Інтерфейс для обробників компенсації кроків.
namespace Queuety\Contracts;
interface Compensation {
public function handle( array $state ): void;
}
Contracts\RepeatCondition
Інтерфейс для предикатів циклу, які використовують repeat_until() і repeat_while(), коли порівняння за ключем стану надто обмежене.
namespace Queuety\Contracts;
interface RepeatCondition {
public function matches( array $state ): bool;
}
Contracts\StateAction
Інтерфейс для дій, що додаються до черги під час входу в стан.
namespace Queuety\Contracts;
interface StateAction {
public function handle( array $state, ?string $event = null, array $event_payload = [], array $payload = [] ): array|string;
}
$payload отримує структуровані payload з визначення дії, якщо автомат було визначено з ними. Поверніть оновлення публічного стану. Додайте _event і за потреби _event_payload, якщо дія має негайно спричинити перехід.
Дії стану можуть також визначати необов’язковий метод config(). Queuety зчитує з нього concurrency_group, concurrency_limit і cost_units, коли дію входу додають до черги.
Contracts\StateGuard
Інтерфейс для умов переходу автомата станів.
namespace Queuety\Contracts;
interface StateGuard {
public function allows( array $state, array $event_payload, string $event, array $payload = [] ): bool;
}
Contracts\Job
Інтерфейс для самодостатніх класів завдань, які можна відправляти. Класи завдань інкапсулюють корисне навантаження у відкритих властивостях, а логіку виконання — у handle().
namespace Queuety\Contracts;
interface Job {
public function handle(): void;
}
Класи завдань можуть за потреби визначати метод middleware(), який повертає масив екземплярів Contracts\Middleware.
Contracts\Cache
Інтерфейс для бекендів кешу. Вбудовані реалізації та власні бекенди описано в розділі Кешування.
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;
}
| Метод | Повертає | Опис |
|---|---|---|
get( $key ) |
mixed |
Отримати значення. Повертає null, якщо значення не знайдено або строк його дії минув. |
set( $key, $value, $ttl ) |
void |
Зберегти значення. TTL 0 означає, що строк дії не обмежений. |
delete( $key ) |
void |
Видалити значення. |
has( $key ) |
bool |
Перевірити, чи існує ключ і чи не минув строк його дії. |
add( $key, $value, $ttl ) |
bool |
Атомарно встановити, лише якщо значення ще не існує. Повертає true, якщо значення було встановлено. |
flush() |
void |
Видалити всі елементи з кешу. |
WorkflowEventLog
Записує переходи між кроками робочого процесу з повним контекстом трасування. Доступ через Queuety::workflow_events().
| Метод | Повертає | Опис |
|---|---|---|
record_event( array $event ) |
void |
Записати необроблену подію трасування |
record_step_started( ... ) |
void |
Записати подію step_started із вхідними даними та станом до виконання |
record_step_completed( ... ) |
void |
Записати подію step_completed із вхідними даними, вихідними даними, станом до та після виконання |
record_step_failed( ... ) |
void |
Записати подію step_failed із контекстом трасування та структурованими відомостями про помилку |
get_timeline( int $workflow_id, ?int $limit = 100, int $offset = 0 ) |
array |
Отримати події трасування робочого процесу за порядком, із можливістю посторінкового перегляду |
get_trace( int $workflow_id ) |
array |
Отримати нормалізований пакет трасування |
get_state_at_step( int $workflow_id, int $step_index ) |
array|null |
Отримати публічний стан після завершення або відновлення певного кроку |
prune( int $older_than_days ) |
int |
Видалити події, старші за вказану кількість днів. Повертає кількість видалених рядків. |
StateMachineEventLog
Записує події життєвого циклу автомата станів, дій, переходів і умов із тими самими полями трасування, що й події робочого процесу.
| Метод | Повертає | Опис |
|---|---|---|
record_event( array $event ) |
void |
Записати необроблену подію трасування автомата |
get_timeline( int $machine_id, ?int $limit = 100, int $offset = 0 ) |
array |
Отримати події трасування автомата за порядком, із можливістю посторінкового перегляду |
get_trace( int $machine_id ) |
array |
Отримати нормалізований пакет трасування автомата |
get_state_at_event( int $machine_id, int $event_id ) |
array|null |
Отримати публічний стан після певної події трасування автомата |
prune( int $older_than_days ) |
int |
Видалити події, старші за вказану кількість днів. Повертає кількість видалених рядків. |
Contracts\StreamingStep
Інтерфейс для обробників кроків робочого процесу, які створюють потік фрагментів. Кожне видане значення негайно зберігається в базі даних. Див. Потокові кроки.
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;
}
| Метод | Опис |
|---|---|
stream( $state, $existing_chunks ) |
Генератор, що видає рядкові фрагменти. Кожне видання спричиняє негайний запис у БД. Під час повторної спроби $existing_chunks містить раніше збережені фрагменти. |
on_complete( $chunks, $state ) |
Викликається після завершення потоку. Повертає дані для об’єднання зі станом робочого процесу. |
config() |
Необов’язкова конфігурація. Підтримує max_attempts, backoff, concurrency_group, concurrency_limit і cost_units. |
ChunkStore
Керує збереженням фрагментів для потокових кроків. Доступ через Queuety::chunk_store().
| Метод | Повертає | Опис |
|---|---|---|
get_chunks( int $job_id ) |
string[] |
Отримати всі фрагменти завдання, упорядковані за індексом фрагмента |
append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index ) |
void |
Додати один фрагмент |
clear_chunks( int $job_id ) |
void |
Видалити всі фрагменти завдання |
chunk_count( int $job_id ) |
int |
Порахувати фрагменти завдання |
get_accumulated( int $job_id ) |
string |
Об’єднати всі фрагменти в один рядок |
Heartbeat
Статичний допоміжний засіб для надсилання сигналів активності з обробників кроків і потокових кроків. Див. Сигнали активності.
| Метод | Опис |
|---|---|
Heartbeat::init( int $job_id, Connection $conn ) |
Задати контекст завдання. Викликається автоматично воркером. |
Heartbeat::beat( array $progress = [] ) |
Надіслати сигнал активності. Оновлює reserved_at і за потреби зберігає дані про перебіг у heartbeat_data. |
Heartbeat::clear() |
Очистити контекст завдання. Викликається автоматично воркером. |
Heartbeat::current_job_id() |
Отримати поточний ID завдання (для тестування). |
Contracts\Middleware
Інтерфейс для проміжного програмного забезпечення завдань. Воно обгортає виконання завдань у конвеєр, подібний до цибулини.
namespace Queuety\Contracts;
interface Middleware {
public function handle( object $job, \Closure $next ): void;
}
Трейти
Dispatchable
Надає статичні методи відправлення для класів завдань, що реалізують Contracts\Job. Аргументи конструктора передаються для створення екземпляра завдання.
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' );
| Метод | Повертає | Опис |
|---|---|---|
dispatch( ...$args ) |
PendingJob |
Створити екземпляр і відправити його |
dispatch_if( bool $condition, ...$args ) |
PendingJob|null |
Відправити, лише якщо $condition має значення true |
dispatch_unless( bool $condition, ...$args ) |
PendingJob|null |
Відправити, лише якщо $condition має значення false |
with_chain( array $jobs ) |
ChainBuilder |
Розпочати ланцюжок із цього класу завдань |
// 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();
Властивості завдання
Класи завдань можуть оголошувати відкриті властивості, які воркер зчитує за допомогою рефлексії, щоб перевизначити параметри повторних спроб і тайм-ауту.
| Властивість | Тип | Опис |
|---|---|---|
$tries |
int |
Максимальна кількість спроб, після якої завдання переміщується до відкладених |
$timeout |
int |
Максимальний час виконання в секундах (потрібне pcntl) |
$backoff |
array |
Зростаючі інтервали очікування в секундах для кожної повторної спроби |
class ImportProductsJob implements Job {
use Dispatchable;
public int $tries = 5;
public int $timeout = 120;
public array $backoff = [ 10, 30, 60, 120 ];
// ...
}
Хук failed()
Коли після вичерпання всіх повторних спроб завдання переміщується до відкладених, воркер викликає метод failed() екземпляра завдання, якщо він визначений.
public function failed( \Throwable $exception ): void {
error_log( "Job failed: {$exception->getMessage()}" );
}
Вбудоване проміжне програмне забезпечення
Усі класи проміжного програмного забезпечення реалізують Contracts\Middleware і містяться в просторі імен Queuety\Middleware.
| Клас | Опис |
|---|---|
RateLimited( int $max, int $window ) |
Обмежити частоту до $max виконань за $window секунд |
Timeout( int $seconds ) |
Завершити завдання через $seconds секунд (потрібне pcntl) |
UniqueJob( ?string $key ) |
Запобігти паралельному виконанню за допомогою блокування бази даних |
WithoutOverlapping( string $key, int $release_after ) |
Запобігти одночасному виконанню з автоматичним завершенням строку блокування |
ThrottlesExceptions( int $max_attempts, int $decay_minutes ) |
Обмежити кількість винятків, щоб запобігти лавині завдань у разі збою зовнішньої служби |
Відомості про використання та приклади див. у розділі Проміжне програмне забезпечення.
Переліки
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';
}
Докладніше про роботу кожної політики див. у розділі Планування.
Атрибути
#[QueuetyHandler]
PHP 8 атрибут для автоматичної реєстрації. Див. атрибути PHP.
use Queuety\Attributes\QueuetyHandler;
#[QueuetyHandler( name: 'send_email', queue: 'emails', max_attempts: 5 )]
class SendEmailHandler implements Handler {
// ...
}
Міст CLI
Queuety::cli_command()
Повертає опубліковану назву кореневої команди WP-CLI.
Коренева команда за замовчуванням — queuety. Визначте QUEUETY_CLI_COMMAND до запуску Queuety, якщо команду потрібно опублікувати під іншою назвою.
Queuety::cli_command_map()
Повертає серіалізований каталог Queuety CLI для тестових середовищ, яким потрібно знати доступний набір команд без зчитування класів WP-CLI.
Кожен запис містить:
- стабільний ID операції
- відносний шлях CLI
- клас і метод обробника WP-CLI
- можливі цілі виконання
- адаптер, що нормалізує необроблені аргументи CLI у план виконання
Queuety::resolve_cli_command( $path, $args, $assoc_args )
Перетворює розібраний виклик CLI на план виконання, який має запустити тестове середовище.
$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'] );
Плани виконання навмисно враховують транспорт:
- команди зіставляються з публічними викликаними об’єктами PHP із нормалізованими аргументами
- у каталозі заздалегідь наведено всі можливі цілі виконання, а
callableіtransportвизначають гілку, обрану для розібраного введення
Це стандартний варіант використання Queuety для тестових середовищ агентів: стабільний ID операції, публічна ціль API і нормалізація, контрольована адаптером, для специфічних лише для CLI аспектів, як-от типові параметри черги, декодування корисного навантаження JSON, розгалуження пулу воркерів і обробка файлів експорту робочого процесу.