Μετάβαση στο περιεχόμενο

Ροές εργασίας

Προβολή ως Markdown Επεξεργασία στο GitHub Τεκμηρίωση στο 2cc7b2b

Οι ροές εργασίας χωρίζουν τις μακρόχρονες εργασίες σε διακριτά βήματα. Κάθε βήμα αποθηκεύει την έξοδό του σε μια κοινόχρηστη δομή κατάστασης στη βάση δεδομένων. Αν το PHP τερματιστεί στη μέση ενός βήματος, ο worker επαναλαμβάνει το βήμα με ανέπαφη όλη την προηγούμενη κατάσταση.

Τι είναι οι ροές εργασίας;

Μια ροή εργασίας είναι μια διατεταγμένη ακολουθία βημάτων. Κάθε βήμα είναι μια κλάση 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'}

Κάθε βήμα βλέπει ολόκληρη τη συσσωρευμένη κατάσταση. Έτσι είναι εύκολο να μεταφέρετε δεδομένα μεταξύ βημάτων χωρίς εξωτερική αποθήκευση.

Δυνατότητα συνέχισης

Οι ροές εργασίας είναι ανθεκτικές επειδή η κατάσταση αποθηκεύεται στη βάση δεδομένων μετά από κάθε βήμα. Αν ο worker καταρρεύσει κατά το βήμα 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 αποθηκεύει επίσης ένα ντετερμινιστικό hash ορισμού για κάθε εκτέλεση, το idempotency_key() διατηρεί την ανθεκτικότητα της αποστολής σε επαναλήψεις ή διπλά αιτήματα και τα max_transitions(), max_for_each_items() και max_state_bytes() αποτυγχάνουν τις ανεξέλεγκτες εκτελέσεις πριν αυτές ξεφύγουν. Δείτε τις Δικλείδες ασφαλείας.

Τύποι βημάτων

Το Queuety υποστηρίζει έντεκα τύπους βημάτων μέσα σε μια ροή εργασίας:

Συνήθη μοτίβα

Αν δημιουργείτε αυτόνομα ή ημιαυτόνομα συστήματα, ξεκινήστε από την Ενορχήστρωση agent. Δείχνει πώς συνδυάζονται τα for_each(), start_agents(), wait_for_agents(), wait_for_workflow() και wait_for_decision() σε πραγματικές ροές σχεδιασμού/εκτέλεσης και ανθρώπινου ελέγχου.