simple workflow go

Простой задача

Простая структура рабочего процесса, написанная в Go

[Todo вставьте изображение здесь]

Демо

QuickStarts

Бэкэнд

Бэкэнд отвечает за постоянное состояние рабочего процесса, включая задачи, события, метаданные выполнения рабочего процесса.

В этом примере мы будем использовать PSQL в качестве бэкэнда.

Во -первых, начните с сервером PSQL локально

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

Затем инициируйте бэкэнд -экземпляр, который подключается к PSQL Server

 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 несут ответственность за выполнение деятельности и кодов рабочих процессов

Activityworker

 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
}

Результат

Запуск рабочий процесс

После работы с нашим экземпляром работника мы можем написать коды, чтобы начать рабочие процессы и ждать их результата

Начните рабочий процесс

Чтобы запланировать рабочий процесс, позвоните ScheduleWorkflow , а затем заполните необходимые параметры, чтобы запустить рабочий процесс

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

Результат

Отладка работающего рабочего процесса

Во -первых, чтобы отлаживать бегущий рабочий процесс, мы должны поместить несколько переменных времени выполнения в рабочем процессе.

Мы будем использовать метод 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 )

Результат

В ожидании результата рабочего процесса

Чтобы дождаться завершения выполнения рабочего процесса и получить его результат, вызовите метод AwaitWorkflowResult

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

Результат

Все вышеупомянутые код были взяты из примера подписки_with_debug

Примеры

Смотрите примеры