simple workflow go

go-workflow-go

Un cadre de workflow simple écrit en Go

[Image d'insertion de TODO ici]

Démo

Starts

Backend

Backend est responsable de la persistance de l'État de flux de travail, y compris des tâches, des événements, des métadonnées d'exécution du flux de travail.

Dans cet exemple, nous utiliserons PSQL comme backend.

Tout d'abord, démarrez PSQL Server localement

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

Ensuite, init une instance backend qui se connecte au serveur 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 )

Activités

L'activité est une fonction qui a utilisé pour implémenter les appels de service, le fonctionnement des E / S, les opérations de longue date ou les actions coûteuses qui ne sont pas préférées

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

Remarque n ° 1 : Le code à l'intérieur de l'activité doit être non déterministe, c'est-à-dire lors de l'exécution de l'activité deux fois, elle donne toujours le même résultat

Remarque n ° 2 : Si vous ressentez une erreur inattendue lors de l'exécution de l'activité, appelez simplement panic(...) , l'activité sera réassée plus tard

Flux de travail

Le flux de travail est l'orchestration des activités

 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
}

Remarque n ° 1 : ne mettez pas aucune opération coûteuse (opérations IO, appels de service externes, etc.) sur le code de workflow, mettez-les à la place dans le code d'activité

Ouvriers

Les travailleurs, y compris ActivityWorker et WorkflowWorker sont responsables de l'exécution des codes d'activité et de flux de travail

Travailleur de l'activité

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

Travailleur de travail

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

Assembler

En rassemblant toutes les pièces, nous pouvons mettre en œuvre notre programme de travailleurs

 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
}

Résultat

Démarrer un workflow

Après avoir fait fonctionner notre instance de travail, nous pouvons écrire des codes pour commencer les workflows et attendre leur résultat

Commencer le workflow

Pour planifier un flux de travail, appelez ScheduleWorkflow , puis remplissez les paramètres nécessaires pour démarrer le workflow

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

Résultat

Déboguer un flux de travail en cours d'exécution

Tout d'abord, pour déboguer un flux de travail en cours, nous devons mettre plusieurs variables d'exécution à l'intérieur du flux de travail.

Nous utiliserons la méthode SetVar[T any](ctx context.Context, name string, value T) qui est utilisée pour modifier une variable d'exécution. Après cela, nous utiliserons le WorkflowDebugger pour déboguer l'état d'exécution actuel en réalisant ces variables.

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

Résultat

Attendre le résultat du flux de travail

Pour attendre une exécution de workflow pour terminer et obtenir son résultat, appelez la méthode AwaitWorkflowResult

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

Résultat

Tout le code ci-dessus a été pris à partir de l'abonnement_with_debug Exemple

Exemples

Voir des exemples