انتقل إلى المحتوى

مسارات العمل

عرض بتنسيق 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() في تدفقات عمل المخطِّط والمنفّذ ومراجعة البشر الفعلية.