simple workflow go
1.0.0
一个简单的工作流程框架
[在这里插入图像]
后端负责持续工作流状态，包括任务，事件，工作流运行时元数据。
在此示例中，我们将使用PSQL作为后端。
首先，本地启动PSQL Server
docker compose -f docker/docker-compose-psql.yaml up -d然后，INIT连接到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 )活动是用于实现服务呼叫，I/O操作，长期运行或不希望重新执行的昂贵的操作的功能
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
}
运行我们的工人实例后，我们可以编写代码以开始工作流程并等待其结果
要安排工作流程，请致电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) ，用于修改运行时变量。之后，我们将使用WorkFlowDebugger通过将这些变量删除来调试当前的运行时状态。
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 )
所有上述代码均来自subpcription_with_debug示例
见示例
