SQS - 步骤功能管道失败

Question

我是步骤函数的新手，并且有一个管道来轮询 SQS 队列并根据收到的消息启动步骤函数执行。目的是一次处理一条消息。但是，Step Function 一开始就遇到问题，并伴有以下异常。我不知道为什么

"ResultPath": "$.ExecutionDetails"

不起作用。

States.ReferencePathConflict
Unable to apply step "ExecutionDetails" to input [{"messageId":"xyz","receiptHandle":"abc","body":"{\"id\": 1, \"price\": 100}","attributes":{...}]

输入信息：

{"id": 1, "price": 100}

在下图中，每次处理和删除消息时，都会执行

GetQueueAttributes

来检查

ApproximateNumberOfMessages

。如果队列中有另一条消息 (

ApproximateNumberOfMessages > 0

)，则会返回到

ReceiveMessage

状态。

状态机定义：

{
  "StartAt": "ListExecutions",
  "States": {
    "ListExecutions": {
      "Type": "Task",
      "Next": "Check Status",
      "Parameters": {
        "StateMachineArn.$": "$$.StateMachine.Id",
        "StatusFilter": "RUNNING"
      },
      "Resource": "arn:aws:states:::aws-sdk:sfn:listExecutions",
      "ResultSelector": {
        "StateMachineArn.$": "$$.StateMachine.Id",
        "uuid.$": "States.UUID()",
        "runningExecutionsCount.$": "States.ArrayLength($.Executions)"
      },
      "ResultPath": "$.ExecutionDetails"
    },
    "Check Status": {
      "Type": "Choice",
      "Choices": [
        {
          "Variable": "$.ExecutionDetails.runningExecutionsCount",
          "NumericEquals": 1,
          "Next": "ReceiveMessage"
        }
      ],
      "Default": "Pass"
    },
    "ReceiveMessage": {
      "Type": "Task",
      "Next": "Dummy Processing",
      "Parameters": {
        "MaxNumberOfMessages": 1,
        "QueueUrl": "https://sqs.us-east-1.amazonaws.com/xxx/test-sqs.fifo"
      },
      "Resource": "arn:aws:states:::aws-sdk:sqs:receiveMessage",
      "ResultSelector": {
        "body.$": "States.StringToJson($.Messages[0].Body)",
        "ReceiptHandle.$": "$.Messages[0].ReceiptHand"
      }
    },
    "Pass": {
      "Type": "Pass",
      "End": true
    },
    "Dummy Processing": {
      "Type": "Pass",
      "Next": "DeleteMessage"
    },
    "DeleteMessage": {
      "Type": "Task",
      "Parameters": {
        "QueueUrl": "https://sqs.us-east-1.amazonaws.com/xxx/test-sqs.fifo",
        "ReceiptHandle.$": "$.ReceiptHandle"
      },
      "Resource": "arn:aws:states:::aws-sdk:sqs:deleteMessage",
      "Next": "GetQueueAttributes"
    },
    "GetQueueAttributes": {
      "Type": "Task",
      "Parameters": {
        "AttributeNames": [
          "ApproximateNumberOfMessages"
        ],
        "QueueUrl": "https://sqs.us-east-1.amazonaws.com/xxx/test-sqs.fifo"
      },
      "Resource": "arn:aws:states:::aws-sdk:sqs:getQueueAttributes",
      "Next": "Choice"
    },
    "Choice": {
      "Type": "Choice",
      "Choices": [
        {
          "Variable": "$.Attributes.ApproximateNumberOfMessages",
          "StringEquals": "0",
          "Next": "Done"
        }
      ],
      "Default": "ReceiveMessage"
    },
    "Done": {
      "Type": "Pass",
      "End": true
    }
  }
}

Answer 1

您的 Step Function 接收 SQS 消息数组 作为输入。您可以在错误消息中看到这一点。因此，在

ResultPath

键中，

$

指的是数组。错误消息告诉您

"ResultPath": "$.ExecutionDetails"

没有意义，因为

$

是一个数组，而不是一个对象。

您需要从数组输入中提取消息项。如果您可以指望输入中始终只有一条消息，则可以在

Pass

状态的 Parameters 键中添加 "States.ArrayGetItem($, 0)" 内部函数来提取唯一的消息项。如果您的执行可能会收到多条消息，则需要添加一个 Map 状态来处理每一项。