الصفحة الرئيسية>المتعلقة بالبرمجة>شفرة المصدر الأخرى
تنزيل

البسيط-التدفق

إطار عمل بسيط من سير العمل مكتوب في GO

[TODO INSERT IMATE هنا]

العرض التوضيحي

Demo.gif

Quickstarts

الخلفية

الواجهة الخلفية مسؤولة عن استمرار حالة سير العمل ، بما في ذلك المهام والأحداث وبيانات تعريف وقت تشغيل سير العمل.

في هذا المثال ، سوف نستخدم PSQL كخلفية.

أولاً ، ابدأ خادم PSQL محليًا 

docker compose -f docker/docker-compose-psql.yaml up -d

بعد ذلك ، inter مثيل خلفي يتصل بخادم PSQL 

 const (
    DbHost     = "localhost"
    DbPort     = 5432
    DbName     = "postgres"
    DbUser     = "user"
    DbPassword = "123456"
)

func InitPSQLBackend ( logger * zap. Logger ) (backend. Backend , error ) {
    hostname , err := os . Hostname ()
    if err != nil {
        return nil , err
    }
    db , err := psql . Connect ( DbHost , DbPort , DbUser , DbPassword , DbName , nil )
    if err != nil {
        return nil , err
    }
    err = psql . PrepareDB ( db ) // auto-create table if not exists
    if err != nil {
        return nil , err
    }
    dataConverter := dataconverter . NewJsonDataConverter ()
    be := psql . NewPSQLBackend ( hostname , 5 * time . Minute , dataConverter , db , logger )
    return be , nil
}
 be , err := examples . InitPSQLBackend ( logger )

أنشطة

النشاط هو وظيفة تستخدم لتنفيذ مكالمات الخدمة ، أو تشغيل الإدخال/الإخراج ، أو العمليات طويلة الأمد ، أو الإجراءات المكلفة التي لا تفضل إعادة التنفيذ 

 type PaymentInput struct {
    Value int64
}

func PaymentActivity ( ctx context. Context , input * PaymentInput ) ( * Void , error ) {
    r := rand . Intn ( 100 )
    if r < 30 { // 30% of failure
        return & Void {}, nil
    } else {
        return nil , errors . New ( "payment failed" )
    }
}

ملاحظة رقم 1 : يجب أن يكون النشاط داخل النشاط غير محدد ، وهو عند تشغيل النشاط مرتين ، فإنه يعطي دائمًا نفس النتيجة

ملاحظة رقم 2 : إذا واجهت خطأ غير متوقع أثناء تنفيذ النشاط ، فما عليك سوى الاتصال panic(...) ، فسيتم إعادة إعادة النشاط لاحقًا

سير العمل

سير العمل هو تنسيق الأنشطة 

 type SubscriptionWorkflowInput struct {
    TotalAmount   int64
    Cycles        int
    CycleDuration time. Duration
}

type SubscriptionWorkflowOutput struct {
    Paid    int64
    Overdue int64
}

func SubscriptionWorkflow ( ctx context. Context , input * SubscriptionWorkflowInput ) ( * SubscriptionWorkflowOutput , error ) {
    startTimestamp := workflow . GetWorkflowExecutionStartedTimestamp ( ctx )
    paymentAmounts := calculatePaymentCycles ( input . TotalAmount , input . Cycles )
    paymentTimings := calculatePaymentTimings ( startTimestamp , input . Cycles , input . CycleDuration )
    //
    var paid int64 = 0
    var overdue int64 = 0
    currentCycle := 0
    for {
        workflow . SetVar ( ctx , "paid" , paid )
        workflow . SetVar ( ctx , "overdue" , overdue )
        workflow . SetVar ( ctx , "currentCycle" , currentCycle )
        if currentCycle >= input . Cycles {
            break
        }
        currentCycleAmount := paymentAmounts [ currentCycle ]
        amountToPay := currentCycleAmount + overdue
        workflow . SetVar ( ctx , "amountToPay" , amountToPay )
        workflow . WaitUntil ( ctx , time . UnixMilli ( paymentTimings [ currentCycle ]))
        _ , err := workflow . CallActivity ( ctx , PaymentActivity , & PaymentInput { Value : amountToPay }). Await ()
        if err != nil {
            overdue += paymentAmounts [ currentCycle ]
            workflow . SetVar ( ctx , fmt . Sprintf ( "cycle_%d_err" , currentCycle ), err . Error ())
        } else {
            paid += amountToPay
            overdue = 0
            workflow . SetVar ( ctx , fmt . Sprintf ( "cycle_%d_paid_amount" , currentCycle ), amountToPay )
        }
        workflow . SetVar ( ctx , "amountToPay" , 0 )
        workflow . SetVar ( ctx , fmt . Sprintf ( "cycle_%d_completed_at" , currentCycle ), workflow . GetCurrentTimestamp ( ctx ))
        currentCycle += 1
    }
    return & SubscriptionWorkflowOutput {
        Paid :    paid ,
        Overdue : overdue ,
    }, nil
}

ملاحظة رقم 1 : لا تضع أي عمليات مكلفة (عمليات IO ، مكالمات الخدمة الخارجية ، إلخ) على رمز سير العمل ، ضعها في رمز النشاط بدلاً من ذلك

العمال

العمال ، بما في ذلك ActivityWorker وعامل WorkflowWorker مسؤولون عن تنفيذ رموز النشاط ورموز سير العمل

عمل النشاط 

 aw , err := worker . NewActivityWorkersBuilder ().
    WithName ( "demo activity worker" ).
    WithBackend ( be ).
    WithLogger ( logger ).
    RegisterActivities (
        PaymentActivity ,
    ).
    WithActivityWorkerOpts (
        activity_worker . WithTaskProcessorMaxBackoffInterval ( 1 * time . Minute ),
    ).
    Build ()

عامل العمل 

 ww , err := worker . NewWorkflowWorkersBuilder ().
    WithName ( "demo workflow worker" ).
    WithBackend ( be ).
    WithLogger ( logger ).
    RegisterWorkflows (
        SubscriptionWorkflow ,
    ). Build ()

وضع كل شيء معا

وضع جميع القطع معًا ، يمكننا تنفيذ برنامج العمال لدينا 

 func main () {
    ctx := context . Background ()
    logger , err := examples . GetLogger ()
    if err != nil {
        panic ( err )
    }
    be , err := examples . InitPSQLBackend ( logger )
    if err != nil {
        panic ( err )
    }
    aw , err := worker . NewActivityWorkersBuilder ().
        WithName ( "demo activity worker" ).
        WithBackend ( be ).
        WithLogger ( logger ).
        RegisterActivities (
            PaymentActivity ,
        ).
        WithActivityWorkerOpts (
            activity_worker . WithTaskProcessorMaxBackoffInterval ( 1 * time . Minute ),
        ).
        Build ()
    if err != nil {
    panic ( err )
    }
    ww , err := worker . NewWorkflowWorkersBuilder ().
        WithName ( "demo workflow worker" ).
        WithBackend ( be ).
        WithLogger ( logger ).
        RegisterWorkflows (
            SubscriptionWorkflow ,
        ). Build ()
    if err != nil {
        panic ( err )
    }
    aw . Start ( ctx )
    defer aw . Stop ( ctx )
    ww . Start ( ctx )
    defer ww . Stop ( ctx )
    //
    sigs := make ( chan os. Signal , 1 )
    signal . Notify ( sigs , syscall . SIGINT , syscall . SIGTERM )
    <- sigs
}

نتيجة

start_worker.gif

بدء سير العمل

بعد تشغيل مثيل العامل الخاص بنا ، يمكننا كتابة رموز لبدء سير العمل وانتظار نتائجها

بدء سير العمل

لجدولة سير العمل ، ومكالمات ScheduleWorkflow ، ثم ملء المعلمات اللازمة لبدء سير العمل 

 err := client . ScheduleWorkflow ( ctx , be , SubscriptionWorkflow , & SubscriptionWorkflowInput {
    TotalAmount : totalAmount ,
    Cycles :      cycles ,
}, client. WorkflowScheduleOptions {
    WorkflowID : workflowID ,
    Version :    "1" ,
})

نتيجة

جدول العمل

تصحيح سير العمل الجري

أولاً ، لتصحيح سير العمل الجاري ، يتعين علينا وضع العديد من متغيرات وقت التشغيل داخل سير العمل.

سنستخدم Method SetVar[T any](ctx context.Context, name string, value T) التي يتم استخدامها لتعديل متغير وقت التشغيل. بعد ذلك ، سوف نستخدم سير العمل لتصحيح حالة وقت التشغيل الحالية من خلال إخراج تلك المتغيرات. 

 dbg := debug . NewWorkflowDebugger ( be )
vars , err := dbg . QueryUserDefinedVars ( SubscriptionWorkflow , workflowID )
if err != nil {
    panic ( err )
}
PrettyPrint ( vars )

نتيجة

Debug-Workflow.gif

في انتظار نتيجة سير العمل

لانتظار تنفيذ سير العمل لإكمالها AwaitWorkflowResult 

 workflowResult , workflowErr , err := client . AwaitWorkflowResult ( ctx , be , SubscriptionWorkflow , workflowID )

نتيجة

await-workflow.gif

تم أخذ جميع الكود أعلاه من الاشتراك _with_debug مثال

أمثلة

انظر الأمثلة

يوسع
معلومات إضافية
تطبيقات ذات صلة
نوصي لك
أخبار ذات صلة الكل