simple workflow go

Fluxo de trabalho simples

Uma estrutura simples de fluxo de trabalho escrita em Go

[TODO Insira a imagem aqui]

Demonstração

Investidores rápidos

Back -end

O back -end é responsável pelo estado persistente do fluxo de trabalho, incluindo tarefas, eventos, metadados de tempo de execução do fluxo de trabalho.

Neste exemplo, usaremos o PSQL como back -end.

Primeiro, inicie o servidor PSQL localmente

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

Então, inicie uma instância de back -end que se conecte ao servidor 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 )

Atividades

A atividade é uma função usada para implementar chamadas de serviço, operação de E/S, operações de longa duração ou ações caras que não preferem ser reexecutadas

 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" )
    }
}

Nota nº 1 : o código interno da atividade deve ser não determinístico, que é quando a execução da atividade duas vezes, sempre produz o mesmo resultado

Nota nº 2 : Se você sofrer um erro inesperado ao executar a atividade, basta ligar para panic(...) , a atividade será julgada posteriormente

Fluxo de trabalho

Fluxo de trabalho é a orquestração de atividades

 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
}

Nota nº 1 : Não coloque nenhuma operações caras (operações de IO, chamadas de serviço externo etc.) no código do fluxo de trabalho, coloque -as no código de atividade em vez disso

Trabalhadores

Os trabalhadores, incluindo ActivityWorker e WorkflowWorker são responsáveis ​​pela execução de códigos de atividade e fluxo de trabalho

ATIVILYWORKER

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

WorkflowWorker

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

Juntando tudo

Juntando todas as peças, podemos implementar nosso programa de trabalhadores

 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
}

Resultado

Iniciando um fluxo de trabalho

Depois de ter a instância do trabalhador em execução, podemos escrever códigos para iniciar fluxos de trabalho e aguardar o resultado deles

Inicie o fluxo de trabalho

Para agendar um fluxo de trabalho, ligue ScheduleWorkflow e preencha os parâmetros necessários para iniciar o fluxo de trabalho

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

Resultado

Debugando um fluxo de trabalho em execução

Primeiro, para depurar um fluxo de trabalho em execução, temos que colocar várias variáveis ​​de tempo de execução dentro do fluxo de trabalho.

Usaremos o método SetVar[T any](ctx context.Context, name string, value T) que é usado para modificar uma variável de tempo de execução. Depois disso, usaremos o WorkflowDebugger para depurar o estado de execução atual, destacando essas variáveis.

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

Resultado

Aguarde o resultado do fluxo de trabalho

Para aguardar a execução do fluxo de trabalho concluir e obter seu resultado, a chamada AwaitWorkflowResult método

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

Resultado

Todo o código acima foi retirado do exemplo de Subscription_with_debug

Exemplos

Veja exemplos