适用于 Go 的 AWS 开发工具包版本？
v2.0.0-preview.3
Go ( go version) 的版本？
go1.9.3 达尔文/amd64
你看到了什么问题？
我正在为 Go 中的 Step Functions 编写一个 Activity Worker。
什么时候：
- 活动工作者正在运行，
- 然后，我们从 SFN 控制台开始执行工作流
一切似乎都运行良好。
但是，当：
- 活动工作者停止，
- 然后，我们从 SFN 控制台开始执行工作流，
- 然后工人重新启动，
工作人员似乎在轮询 SFN，但它不执行在停止期间启动的任务。如果我们此时开始一个新的工作流执行（当工作者运行时），工作者成功地执行新任务。在工作人员停止期间执行的工作流没有被工作人员拾取。
编辑：查看执行历史记录，我看到超时状态和以下事件日志：
重现步骤
在这里，如果我的 SFN 状态机：
{
"Comment": "An example using a Task state.",
"StartAt": "getGreeting",
"Version": "1.0",
"TimeoutSeconds": 300,
"States":
{
"getGreeting": {
"Type": "Task",
"Resource": "arn:aws:states:ap-southeast-1:196709014601:activity:get-greeting",
"End": true
}
}
}
这是我的 SFN 工人：
package main
import (
"github.com/aws/aws-sdk-go-v2/aws"
"github.com/aws/aws-sdk-go-v2/aws/endpoints"
"github.com/aws/aws-sdk-go-v2/aws/external"
"github.com/aws/aws-sdk-go-v2/service/sfn"
"fmt"
"encoding/json"
)
type Worker struct {
svc *sfn.SFN
activityARN string
}
type Task struct {
input *string
token *string
}
func New(activityARN string) *Worker {
cfg, err := external.LoadDefaultAWSConfig()
if err != nil {
panic("unable to load SDK config, " + err.Error())
}
// Set the AWS Region that the service clients should use
cfg.Region = endpoints.ApSoutheast1RegionID
// Using the Config value, create the Step Functions client
svc := sfn.New(cfg)
w := &Worker{
svc: svc,
activityARN: activityARN,
}
return w
}
func (w *Worker) getTask() *Task {
req := w.svc.GetActivityTaskRequest(&sfn.GetActivityTaskInput{
ActivityArn: aws.String(w.activityARN),
})
res, err := req.Send()
if err != nil { fmt.Println("failed to get tasks, "+err.Error()) }
return &Task{
input: res.Input,
token: res.TaskToken,
}
}
// Call SendTaskSuccess on success
func (w *Worker) handleSuccess(taskToken *string, json *string) error {
req := w.svc.SendTaskSuccessRequest(&sfn.SendTaskSuccessInput{
TaskToken: taskToken,
Output: json, // JSON string
})
_, err := req.Send()
if err != nil { fmt.Println("failed to send task success result, "+err.Error()) }
return err
}
// Call SendTaskFailure on error
func (w *Worker) handleFailure(taskToken *string, err error) error {
errorMessage := err.Error()
req := w.svc.SendTaskFailureRequest(&sfn.SendTaskFailureInput{
TaskToken: taskToken,
Error: &errorMessage,
})
_, err = req.Send()
if err != nil { fmt.Println("failed to send task failure result, "+err.Error()) }
return err
}
func main() {
activityARN := "arn:aws:states:ap-southeast-1:196709014601:activity:get-greeting"
worker := New(activityARN)
fmt.Println("Starting worker")
for {
// 1. Poll GetActivityTask API for tasks
fmt.Println("Polling for tasks")
task := worker.getTask()
if task.token == nil { continue }
// 2. Do some actual work
fmt.Println("Working")
result, err := work(task.input)
// 3. Notify SFN on success and failure
fmt.Println("Sending results")
if err == nil {
worker.handleSuccess(task.token, result)
} else {
worker.handleFailure(task.token, err)
}
}
}
// Handles marshalling and un-marshalling JSON
func work(jsonInput *string) (*string, error) {
input := &GreetInput{}
json.Unmarshal([]byte(*jsonInput), input)
result, err := Greet(input) // Actual work
if err != nil { return nil, err }
outputBytes, _ := json.Marshal(result)
output := string(outputBytes)
return &output, nil
}
// Actual handler code
type GreetInput struct {
Who string
}
type GreetOutput struct {
Message string
}
func Greet(input *GreetInput) (*GreetOutput, error) {
message := fmt.Sprintf("hello %s", input.Who)
output := &GreetOutput {
Message: message,
}
fmt.Println(message)
return output, nil
}
跑步：
go build worker.go && ./worker