能否在同一Step Functions工作流中发送并接收SQS消息处理QLDB事务？

完全可以在同一个Step Functions工作流里完成SQS消息发送并等待该消息被处理完成，不用拆分工作流。核心思路是结合SQS FIFO队列的顺序性特性，搭配Step Functions的SendMessage.waitForTaskToken模式，再让处理消息的Lambda回调任务令牌即可。

具体实现步骤 #

1. 工作流中添加带令牌等待的SQS发送步骤 #

在Step Functions状态机里，用AWS SDK集成的sqs:sendMessage.waitForTaskToken动作发送消息，把Step Functions的任务令牌（$$.Task.Token）塞进消息体或消息属性里，同时指定MessageGroupId为用户ID，保证同一用户的消息按顺序排队。

示例状态定义片段：

"SendToSQS": {
  "Type": "Task",
  "Resource": "arn:aws:states:::sqs:sendMessage.waitForTaskToken",
  "Parameters": {
    "QueueUrl": "你的SQS FIFO队列URL",
    "MessageBody": {
      "userId": "$.userId",
      "taskToken": "$$.Task.Token",
      "transactionData": "$.transactionData"
    },
    "MessageGroupId": "$.userId"
  },
  "Next": "TransactionCompleted",
  "Catch": [
    {
      "ErrorEquals": ["States.ALL"],
      "Next": "HandleFailure"
    }
  ]
}

2. 处理SQS消息的Lambda逻辑 #

Lambda从SQS取出消息后，先解析出任务令牌和事务数据，执行QLDB事务逻辑，完成后调用Step Functions的SendTaskSuccess API，把令牌传回去，让工作流继续往下走。

示例Python代码片段：

import boto3
import json

sfn_client = boto3.client('stepfunctions')
qldb_client = boto3.client('qldb')

def lambda_handler(event, context):
    # 解析SQS消息
    record = event['Records'][0]
    message_body = json.loads(record['body'])
    task_token = message_body['taskToken']
    user_id = message_body['userId']
    transaction_data = message_body['transactionData']

    # 执行QLDB事务逻辑
    try:
        # 这里写你的QLDB事务代码，比如执行语句、提交事务等
        qldb_client.execute_statement(
            LedgerName='你的账本名称',
            Statement='INSERT INTO 表名 VALUE ?',
            Parameters=[{'StringValue': json.dumps(transaction_data)}]
        )
        # 回调Step Functions，标记任务成功
        sfn_client.send_task_success(
            taskToken=task_token,
            output=json.dumps({"status": "success", "userId": user_id})
        )
    except Exception as e:
        # 出错时回调标记失败
        sfn_client.send_task_failure(
            taskToken=task_token,
            error='QLDBTransactionFailed',
            cause=str(e)
        )
    return {"statusCode": 200}

3. 工作流后续处理 #

当Step Functions收到SendTaskSuccess的回调后，会自动进入Next指定的步骤（比如TransactionCompleted），完成整个流程；如果出错，会进入Catch分支处理异常。

关键注意事项 #

• SQS FIFO配置：开启ContentBasedDeduplication或者手动设置MessageDeduplicationId，避免重复消息导致工作流重复等待。
• 令牌有效期：任务令牌默认有效期1年，可按需调整，但要确保QLDB事务处理时间不超过这个期限。
• 异常处理：一定要给任务步骤加Catch分支，处理发送失败、Lambda执行出错等情况，避免工作流一直处于挂起状态。

内容的提问来源于stack exchange，提问作者Vingtoft