AWS无服务器开发Skill aws-serverless

---

name: aws-serverless description: “专门用于在AWS上构建生产就绪的无服务器应用程序的技能。涵盖Lambda函数、API Gateway、DynamoDB、SQS/SNS事件驱动模式、SAM/CDK部署和冷启动优化。” source: vibeship-spawner-skills (Apache 2.0)

AWS 无服务器

模式

Lambda 处理器模式

具有错误处理的适当Lambda函数结构

何时使用: [‘任何Lambda函数实现’, ‘API处理器、事件处理器、计划任务’]

```javascript
// Node.js Lambda Handler
// handler.js

// 在处理器外初始化（跨调用重用）
const { DynamoDBClient } = require('@aws-sdk/client-dynamodb');
const { DynamoDBDocumentClient, GetCommand } = require('@aws-sdk/lib-dynamodb');

const client = new DynamoDBClient({});
const docClient = DynamoDBDocumentClient.from(client);

// 处理器函数
exports.handler = async (event, context) => {
  // 可选：不等待事件循环清空（Node.js）
  context.callbackWaitsForEmptyEventLoop = false;

  try {
    // 根据事件源解析输入
    const body = typeof event.body === 'string'
      ? JSON.parse(event.body)
      : event.body;

    // 业务逻辑
    const result = await processRequest(body);

    // 返回API Gateway兼容的响应
    return {
      statusCode: 200,
      headers: {
        'Content-Type': 'application/json',
        'Access-Control-Allow-Origin': '*'
      },
      body: JSON.stringify(result)
    };
  } catch (error) {
    console.error('Error:', JSON.stringify({
      error: error.message,
      stack: error.stack,
      requestId: context.awsRequestId
    }));

    return {
      statusCode: error.statusCode || 500,
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({
        error: error.message || 'Internal server error'
      })
    };
  }
};

async function processRequest(data) {
  // 您的业务逻辑在这里
  const result = await docClient.send(new GetCommand({
    TableName: process.env.TABLE_NAME,
    Key: { id: data.id }
  }));
  return result.Item;
}

# Python Lambda Handler
# handler.py

import json
import os
import logging
import boto3
from botocore.exceptions import ClientError

# 在处理器外初始化（跨调用重用）
logger = logging.getLogger()
logger.setLevel(logging.INFO)

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(os.environ['TABLE_NAME'])

def handler(event, context):
    try:
        # 解析 i

API Gateway 集成模式

与Lambda的REST API和HTTP API集成

何时使用: [‘构建由Lambda支持的REST API’, ‘需要函数的HTTP端点’]

```yaml
# template.yaml (SAM)
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31

Globals:
  Function:
    Runtime: nodejs20.x
    Timeout: 30
    MemorySize: 256
    Environment:
      Variables:
        TABLE_NAME: !Ref ItemsTable

Resources:
  # HTTP API（推荐用于简单用例）
  HttpApi:
    Type: AWS::Serverless::HttpApi
    Properties:
      StageName: prod
      CorsConfiguration:
        AllowOrigins:
          - "*"
        AllowMethods:
          - GET
          - POST
          - DELETE
        AllowHeaders:
          - "*"

  # Lambda 函数
  GetItemFunction:
    Type: AWS::Serverless::Function
    Properties:
      Handler: src/handlers/get.handler
      Events:
        GetItem:
          Type: HttpApi
          Properties:
            ApiId: !Ref HttpApi
            Path: /items/{id}
            Method: GET
      Policies:
        - DynamoDBReadPolicy:
            TableName: !Ref ItemsTable

  CreateItemFunction:
    Type: AWS::Serverless::Function
    Properties:
      Handler: src/handlers/create.handler
      Events:
        CreateItem:
          Type: HttpApi
          Properties:
            ApiId: !Ref HttpApi
            Path: /items
            Method: POST
      Policies:
        - DynamoDBCrudPolicy:
            TableName: !Ref ItemsTable

  # DynamoDB 表
  ItemsTable:
    Type: AWS::DynamoDB::Table
    Properties:
      AttributeDefinitions:
        - AttributeName: id
          AttributeType: S
      KeySchema:
        - AttributeName: id
          KeyType: HASH
      BillingMode: PAY_PER_REQUEST

Outputs:
  ApiUrl:
    Value: !Sub "https://${HttpApi}.execute-api.${AWS::Region}.amazonaws.com/prod"

// src/handlers/get.js
const { getItem } = require('../lib/dynamodb');

exports.handler = async (event) => {
  const id = event.pathParameters?.id;

  if (!id) {
    return {
      statusCode: 400,
      body: JSON.stringify({ error: 'Missing id parameter' })
    };
  }

  const item =

事件驱动的 SQS 模式

由SQS触发的Lambda进行可靠异步处理

何时使用: [‘解耦、异步处理’, ‘需要重试逻辑和死信队列’, ‘批量处理消息’]

```yaml
# template.yaml
Resources:
  ProcessorFunction:
    Type: AWS::Serverless::Function
    Properties:
      Handler: src/handlers/processor.handler
      Events:
        SQSEvent:
          Type: SQS
          Properties:
            Queue: !GetAtt ProcessingQueue.Arn
            BatchSize: 10
            FunctionResponseTypes:
              - ReportBatchItemFailures  # 部分批次失败处理

  ProcessingQueue:
    Type: AWS::SQS::Queue
    Properties:
      VisibilityTimeout: 180  # 6倍Lambda超时时间
      RedrivePolicy:
        deadLetterTargetArn: !GetAtt DeadLetterQueue.Arn
        maxReceiveCount: 3

  DeadLetterQueue:
    Type: AWS::SQS::Queue
    Properties:
      MessageRetentionPeriod: 1209600  # 14天

// src/handlers/processor.js
exports.handler = async (event) => {
  const batchItemFailures = [];

  for (const record of event.Records) {
    try {
      const body = JSON.parse(record.body);
      await processMessage(body);
    } catch (error) {
      console.error(`Failed to process message ${record.messageId}:`, error);
      // 报告此项目为失败（将重试）
      batchItemFailures.push({
        itemIdentifier: record.messageId
      });
    }
  }

  // 返回失败项目以重试
  return { batchItemFailures };
};

async function processMessage(message) {
  // 您的处理逻辑
  console.log('Processing:', message);

  // 模拟工作
  await saveToDatabase(message);
}

# Python 版本
import json
import logging

logger = logging.getLogger()

def handler(event, context):
    batch_item_failures = []

    for record in event['Records']:
        try:
            body = json.loads(record['body'])
            process_message(body)
        except Exception as e:
            logger.error(f"Failed to process {record['messageId']}: {e}")
            batch_item_failures.append({
                'itemIdentifier': record['messageId']
            })

    return {'batchItemFailures': batch_ite

反模式

❌ 单体 Lambda

为什么不好: 大型部署包导致冷启动缓慢。 难以扩展单个操作。 更新影响整个系统。

❌ 大型依赖

为什么不好: 增加部署包大小。 显著减慢冷启动。 大部分SDK/库可能未使用。

❌ VPC 中的同步调用

为什么不好: 附加VPC的Lambda具有ENI设置开销。 阻塞DNS查找或连接会恶化冷启动。

⚠️ 尖锐边缘

问题	严重性	解决方案
问题	高	## 测量您的INIT阶段
问题	高	## 设置适当的超时
问题	高	## 增加内存分配
问题	中	## 验证VPC配置
问题	中	## 告诉Lambda不要等待事件循环
问题	中	## 对于大文件上传
问题	高	## 使用不同的存储桶/前缀