Към съдържанието

Работни потоци

Преглед като Markdown Редактиране в GitHub Документация за 2cc7b2b

Работните потоци разделят продължителната работа на отделни стъпки. Изходът от всяка стъпка се записва в споделен набор от състояние в базата данни. Ако PHP прекъсне работа по средата на стъпка, работникът я изпълнява отново, като запазва цялото предходно състояние.

Какво представляват работните потоци?

Работният поток е подредена последователност от стъпки. Всяка стъпка е клас PHP, който получава натрупаното състояние от всички предходни стъпки и връща нови данни, които се обединяват със състоянието. Работният поток преминава през стъпките една по една, като всеки преход е обвит в атомарна транзакция MySQL.

PHP
use Queuety\Queuety;

$workflow_id = Queuety::workflow( 'generate_report' )
    ->then( FetchDataHandler::class )
    ->then( CallLLMHandler::class )
    ->then( FormatOutputHandler::class )
    ->dispatch( [ 'user_id' => 42 ] );

Натрупване на състояние

Началният полезен товар, който подавате на dispatch(), става началното състояние. Върнатите ключове от всяка стъпка се обединяват със състоянието. Ключовете, започващи с _, са запазени за контролни стойности като _next_step и не се обединяват с публичното състояние:

Код
dispatch(['user_id' => 42])

Step 0: FetchDataHandler   -> returns ['user_name' => 'Alice']
  state: {user_id: 42, user_name: 'Alice'}

Step 1: CallLLMHandler     -> returns ['llm_response' => '...']
  state: {user_id: 42, user_name: 'Alice', llm_response: '...'}

Step 2: FormatOutputHandler -> returns ['report_url' => '/reports/42.pdf']
  state: {user_id: 42, user_name: 'Alice', llm_response: '...', report_url: '/reports/42.pdf'}

Всяка стъпка вижда цялото натрупано състояние. Така лесно можете да предавате данни между стъпките без външно хранилище.

Възможност за подновяване

Работните потоци са устойчиви, защото състоянието се записва в базата данни след всяка стъпка. Ако работникът се срине по време на стъпка 1, работният поток започва отново от стъпка 1, като изходът от стъпка 0 вече е в състоянието.

Границата между стъпките е една транзакция MySQL, която атомарно:

  1. Актуализира състоянието на работния поток с изхода от стъпката
  2. Маркира задачата на текущата стъпка като завършена
  3. Поставя задачата за следващата стъпка на опашката

Това гарантира, че състоянието никога не се губи дори при сривове, изтичане на времето или фатални грешки на PHP.

Жизнен цикъл на работния поток

Код
dispatch() -> running -> step 0 -> step 1 -> ... -> step N -> completed
                  |                   |
                  |                   +-> failed -> retry -> running (from failed step)
                  |
                  +-> paused -> resumed -> running
                  |
                  +-> waiting_for_signal -> signal received -> running
                  |
                  +-> waiting_for_workflows -> dependency settles -> running
                  |
                  +-> cancelled (via cancel_workflow)
Статус Описание
running Активно обработва стъпки
completed Всички стъпки са завършили успешно
failed Стъпка е неуспешна след изчерпване на повторните опити
paused Ръчно поставен на пауза. Текущата работа в процес може да завърши, но следващите стъпки не се поставят на опашката до подновяване
waiting_for_signal Изчаква външен сигнал чрез Queuety::signal()
waiting_for_workflows Изчаква завършването на един или повече други работни потоци чрез wait_for_workflow() или wait_for_workflows()
cancelled Изрично отменен чрез Queuety::cancel_workflow(). Чакащите задачи се скриват и се изпълнява незадължителен обработчик за почистване.

Управление на работните потоци

Проверка на статуса:

PHP
$state = Queuety::workflow_status( $workflow_id );
echo $state->status->value;   // 'running'
echo $state->current_step;    // 1
echo $state->total_steps;     // 3
print_r( $state->waiting_for ); // ['42'] while waiting on another workflow
print_r( $state->wait_details ); // matched/remaining blockers, result key, correlation info

Поставяне на пауза и подновяване:

PHP
Queuety::pause_workflow( $workflow_id );
Queuety::resume_workflow( $workflow_id );

Повторно изпълнение на неуспешен работен поток от стъпката, при която е възникнала грешката:

PHP
Queuety::retry_workflow( $workflow_id );

Отмяна на работен поток и изпълнение на обработчиците за почистване:

PHP
Queuety::cancel_workflow( $workflow_id );

Вижте Отмяна за подробности относно обработчиците за почистване и какво се случва при отмяна на работен поток.

Опции на работния поток

Задайте опашка, приоритет и поведение при повторни опити за всички стъпки в даден работен поток:

PHP
use Queuety\Enums\Priority;

Queuety::workflow( 'important_pipeline' )
    ->on_queue( 'pipelines' )
    ->with_priority( Priority::High )
    ->max_attempts( 5 )
    ->on_cancel( PipelineCleanupHandler::class )
    ->prune_state_after( 3 )
    ->then( StepA::class )
    ->then( StepB::class )
    ->dispatch( $payload );

Методът on_cancel() регистрира обработчик за почистване, който се изпълнява при отмяна на работния поток. Вижте Отмяна.

Методът prune_state_after() включва автоматично премахване на старите резултати от стъпките, за да ограничи размера на състоянието на работния поток. Вижте Почистване на състоянието.

Използвайте compensate_with(), за да зададете компенсиращ обработчик за всяка стъпка. Използвайте compensate_on_failure(), ако искате компенсациите на завършените стъпки да се изпълняват автоматично след неуспех на работния поток, а не само при изрична отмяна. Вижте Компенсация.

Използвайте защитните механизми на работния поток за по-безопасно оркестриране на продължителни задачи: version() маркира дефиниция с версия на приложението, Queuety съхранява и детерминистичен хеш на дефиницията за всяко изпълнение, idempotency_key() осигурява устойчиво изпращане при повторни опити или дублирани заявки, а max_transitions(), max_for_each_items() и max_state_bytes() прекратяват излизащите извън контрол изпълнения, преди да се разраснат. Вижте Защитни ограничения.

Типове стъпки

Queuety поддържа единадесет типа стъпки в рамките на работен поток:

Често срещани приложения

Ако изграждате автономни или полуавтономни системи, започнете с Оркестриране на агенти. Там е показано как for_each(), start_agents(), wait_for_agents(), wait_for_workflow() и wait_for_decision() работят заедно в реални потоци за планиране и изпълнение, както и за преглед от човек.