如何搭建API Gateway->SQS->Lambda架构并实现Lambda返回响应？

我想要搭建一套 API Gateway -> SQS -> Lambda 的架构，实现Lambda处理完成后向客户端返回API Gateway响应。目前已经完成API Gateway到SQS的配置，当SQS有消息时会触发Lambda函数，但不知道如何让Lambda将处理结果返回给客户端。

现有配置代码（CDK） #

SQS配置 #

const sqs = new sqs.Queue(this, 'sqs', {
   visibilityTimeout: cdk.Duration.seconds(60),
   queueName: 'sqs',
 });
sqs.grantSendMessages(customRole);

const sqsIntegration = new apigw.AwsIntegration({
   service: 'sqs',
   integrationHttpMethod: "POST",
   path: `${cdk.Aws.ACCOUNT_ID}/${sqs.queueName}`,
   options: {
     credentialsRole: customRole,
     passthroughBehavior: apigw.PassthroughBehavior.NEVER,
     integrationResponses: [def.IntegrationResponse(200)],
     requestParameters: {
       'integration.request.header.Content-Type': `'application/x-www-form-urlencoded'`,
     },
     requestTemplates: {
       "application/json": `Action=SendMessage&MessageBody={"route": "$input.params('route')", "seat": "$input.params('seat')", "body": "$util.parseJson($input.body)"}`
     }
   }
});

Lambda配置 #

const lambdaFn = new lambda.Function(this, 'lambdaFn', {
   functionName: 'lambdaFn',
   handler: 'lambdaFn',
   memorySize: 1024,
   role: customRole,
   runtime: lambda.Runtime.GO_1_X,
   timeout: cdk.Duration.seconds(60),
   code: lambda.Code.fromAsset('cmd/lambdaFn')
});

lambdaFn.addEventSource(new SqsEventSource(sqs, {
   enabled: true
}));

API Gateway方法配置 #

apigw.addMethod('POST', sqsIntegration, {
   requestParameters: {
     'method.request.querystring.seat': true,
     'method.request.querystring.route': true
   },
   methodResponses: [
     def.MethodResponse(200, ApiResponseModel),
     def.MethodResponse(400, ApiResponseModel),
     def.MethodResponse(415, ApiResponseModel),
   ],
   requestValidatorOptions: {
     validateRequestBody: false,
     validateRequestParameters: true,
     requestValidatorName: 'api-sqs-validator',
   }
});

当前架构中，API Gateway将消息发送到SQS后会立即返回响应给客户端，Lambda处理SQS消息是异步执行的，因此客户端无法直接获取Lambda的处理结果。要实现Lambda向客户端返回响应，需要调整架构或引入异步响应机制，以下是两种可行方案：

方案一：异步请求+轮询获取结果 #

步骤1：添加DynamoDB存储结果 #

创建DynamoDB表，用于存储请求ID和Lambda的处理结果，支持客户端通过请求ID轮询状态。

const resultTable = new dynamodb.Table(this, 'RequestResultTable', {
  partitionKey: { name: 'requestId', type: dynamodb.AttributeType.STRING },
  billingMode: dynamodb.BillingMode.PAY_PER_REQUEST,
  tableName: 'RequestResultTable'
});

// 授予Lambda读写DynamoDB的权限
resultTable.grantReadWriteData(lambdaFn);

步骤2：修改API Gateway集成，携带请求ID #

调整SQS集成的请求模板，生成唯一requestId并传入SQS消息，同时将requestId返回给客户端：

// 修改SQS集成的requestTemplates
requestTemplates: {
  "application/json": `Action=SendMessage&MessageBody={"requestId": "$context.requestId", "route": "$input.params('route')", "seat": "$input.params('seat')", "body": "$util.parseJson($input.body)"}`
}

// 修改integrationResponses，返回请求ID和待处理状态
integrationResponses: [
  def.IntegrationResponse(202, {
    responseTemplates: {
      "application/json": `{"requestId": "$context.requestId", "status": "pending"}`
    }
  })
]

步骤3：修改Lambda逻辑，写入处理结果 #

Lambda处理SQS消息时，完成业务逻辑后将结果写入DynamoDB关联requestId：

package main

import (
  "context"
  "encoding/json"
  "github.com/aws/aws-lambda-go/events"
  "github.com/aws/aws-lambda-go/lambda"
  "github.com/aws/aws-sdk-go-v2/aws"
  "github.com/aws/aws-sdk-go-v2/config"
  "github.com/aws/aws-sdk-go-v2/service/dynamodb"
  "github.com/aws/aws-sdk-go-v2/service/dynamodb/types"
)

type SQSMessageBody struct {
  RequestId string          `json:"requestId"`
  Route     string          `json:"route"`
  Seat      string          `json:"seat"`
  Body      json.RawMessage `json:"body"`
}

type ProcessingResult struct {
  Status string      `json:"status"`
  Result interface{} `json:"result,omitempty"`
  Error  string      `json:"error,omitempty"`
}

func handler(ctx context.Context, sqsEvent events.SQSEvent) error {
  cfg, err := config.LoadDefaultConfig(ctx)
  if err != nil {
    return err
  }
  dbClient := dynamodb.NewFromConfig(cfg)

  for _, msg := range sqsEvent.Records {
    var body SQSMessageBody
    if err := json.Unmarshal([]byte(msg.Body), &body); err != nil {
      continue
    }

    // 执行业务逻辑
    result, err := processRequest(body.Route, body.Seat, body.Body)
    var processingResult ProcessingResult
    if err != nil {
      processingResult = ProcessingResult{Status: "failed", Error: err.Error()}
    } else {
      processingResult = ProcessingResult{Status: "success", Result: result}
    }

    // 写入DynamoDB
    resultJson, _ := json.Marshal(processingResult)
    _, err = dbClient.PutItem(ctx, &dynamodb.PutItemInput{
      TableName: aws.String("RequestResultTable"),
      Item: map[string]types.AttributeValue{
        "requestId": &types.AttributeValueMemberS{Value: body.RequestId},
        "result":    &types.AttributeValueMemberS{Value: string(resultJson)},
      },
    })
  }
  return nil
}

func processRequest(route, seat string, body json.RawMessage) (interface{}, error) {
  // 替换为实际业务逻辑
  return map[string]string{"route": route, "seat": seat}, nil
}

func main() {
  lambda.Start(handler)
}

步骤4：添加轮询API端点 #

创建查询结果的Lambda，并在API Gateway添加GET方法，支持客户端通过requestId查询结果：

// 创建查询Lambda
const queryResultLambda = new lambda.Function(this, 'QueryResultLambda', {
  functionName: 'QueryResultLambda',
  handler: 'queryResult',
  memorySize: 512,
  role: customRole,
  runtime: lambda.Runtime.GO_1_X,
  timeout: cdk.Duration.seconds(10),
  code: lambda.Code.fromAsset('cmd/queryResult')
});

resultTable.grantReadData(queryResultLambda);

// 添加GET方法到API Gateway
apigw.addMethod('GET', new apigw.LambdaIntegration(queryResultLambda), {
  requestParameters: {
    'method.request.querystring.requestId': true
  },
  methodResponses: [
    def.MethodResponse(200, ApiResponseModel),
    def.MethodResponse(404, ApiResponseModel),
  ],
  requestValidatorOptions: {
    validateRequestParameters: true,
    requestValidatorName: 'api-result-validator',
  }
});

查询Lambda的Go代码：

package main

import (
  "context"
  "encoding/json"
  "github.com/aws/aws-lambda-go/events"
  "github.com/aws/aws-lambda-go/lambda"
  "github.com/aws/aws-sdk-go-v2/aws"
  "github.com/aws/aws-sdk-go-v2/config"
  "github.com/aws/aws-sdk-go-v2/service/dynamodb"
  "github.com/aws/aws-sdk-go-v2/service/dynamodb/types"
)

func handler(ctx context.Context, request events.APIGatewayProxyRequest) (events.APIGatewayProxyResponse, error) {
  requestId := request.QueryStringParameters["requestId"]
  if requestId == "" {
    return events.APIGatewayProxyResponse{StatusCode: 400, Body: `{"error": "requestId is required"}`}, nil
  }

  cfg, err := config.LoadDefaultConfig(ctx)
  if err != nil {
    return events.APIGatewayProxyResponse{StatusCode: 500, Body: `{"error": "internal server error"}`}, err
  }
  dbClient := dynamodb.NewFromConfig(cfg)

  result, err := dbClient.GetItem(ctx, &dynamodb.GetItemInput{
    TableName: aws.String("RequestResultTable"),
    Key: map[string]types.AttributeValue{
      "requestId": &types.AttributeValueMemberS{Value: requestId},
    },
  })
  if err != nil {
    return events.APIGatewayProxyResponse{StatusCode: 500, Body: `{"error": "internal server error"}`}, err
  }

  if result.Item == nil {
    return events.APIGatewayProxyResponse{StatusCode: 404, Body: `{"status": "pending"}`}, nil
  }

  resultStr := result.Item["result"].(*types.AttributeValueMemberS).Value
  return events.APIGatewayProxyResponse{StatusCode: 200, Body: resultStr, Headers: map[string]string{"Content-Type": "application/json"}}, nil
}

func main() {
  lambda.Start(handler)
}

方案二：WebSocket API主动推送结果 #

如果需要实时向客户端推送结果，可使用API Gateway WebSocket API：

步骤1：创建WebSocket API #

const webSocketApi = new apigw.WebSocketApi(this, 'WebSocketApi', {
  connectRouteOptions: {
    integration: new apigw.LambdaIntegration(connectLambda)
  },
  disconnectRouteOptions: {
    integration: new apigw.LambdaIntegration(disconnectLambda)
  }
});

const webSocketStage = new apigw.WebSocketStage(this, 'WebSocketStage', {
  webSocketApi,
  stageName: 'prod',
  autoDeploy: true
});

步骤2：携带连接ID到SQS消息 #

客户端建立WebSocket连接后会获得connectionId，将该ID作为参数传入SQS消息（可通过客户端请求携带或API Gateway模板自动注入）。

步骤3：Lambda处理后推送结果 #

Lambda处理完消息后，调用API Gateway的PostToConnection接口将结果推送给客户端：

// 代码片段
client := apigatewaymanagementapi.NewFromConfig(cfg, func(o *apigatewaymanagementapi.Options) {
  o.BaseEndpoint = aws.String("https://" + apiId + ".execute-api." + region + ".amazonaws.com/prod")
})

_, err = client.PostToConnection(ctx, &apigatewaymanagementapi.PostToConnectionInput{
  ConnectionId: aws.String(connectionId),
  Data:         []byte(resultJson),
})

• 方案一适合非实时场景，实现简单、兼容性强，需处理消息重复、结果过期等问题。
• 方案二适合实时推送场景，但要求客户端支持WebSocket连接。

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