event-driven-serverless-systems

AWS 无服务器与事件驱动架构

本技能基于架构良好框架原则，为在 AWS 上构建无服务器应用和事件驱动架构提供全面指导。

AWS 文档要求

重要：本技能需要使用 AWS MCP 工具来获取准确、最新的 AWS 信息。

回答 AWS 相关问题前

1. 始终使用 AWS MCP 工具（如果可用）进行验证：

   • mcp__aws-mcp__aws___search_documentation 或 mcp__*awsdocs*__aws___search_documentation - 搜索 AWS 文档
   • mcp__aws-mcp__aws___read_documentation 或 mcp__*awsdocs*__aws___read_documentation - 阅读特定页面
   • mcp__aws-mcp__aws___get_regional_availability - 检查服务可用性

2. 如果 AWS MCP 工具不可用：

   • 指引用户配置 AWS MCP：参见 AWS MCP 设置指南
   • 帮助判断哪种选项适合其环境：
     • 已安装 uvx 且拥有 AWS 凭证 → 完整 AWS MCP Server
     • 没有 Python/凭证 → AWS Documentation MCP（无需身份验证）
   • 如果无法判断 → 询问用户要使用哪种选项

无服务器 MCP 服务

本技能可利用专用的无服务器 MCP 服务，提升开发工作流效率：

AWS Serverless MCP Server

用途：通过 SAM CLI 管理完整的无服务器应用生命周期

• 初始化新的无服务器应用
• 部署无服务器应用
• 在本地测试 Lambda 函数
• 生成 SAM 模板
• 管理无服务器应用生命周期

AWS Lambda Tool MCP Server

用途：将 Lambda 函数作为工具执行

• 直接调用 Lambda 函数
• 测试 Lambda 集成
• 执行需要访问私有资源的工作流
• 运行基于 Lambda 的自动化任务

AWS Step Functions MCP Server

用途：执行复杂工作流和编排

• 创建和管理状态机
• 执行工作流编排
• 处理分布式事务
• 实现 Saga 模式
• 协调微服务

Amazon SNS/SQS MCP Server

用途：事件驱动的消息传递和队列管理

• 向 SNS 主题发布消息
• 通过 SQS 队列发送/接收消息
• 管理事件驱动通信
• 实现发布/订阅模式
• 处理异步任务

何时使用本技能

在以下情况下使用本技能：

• 使用 Lambda 构建无服务器应用
• 设计事件驱动架构
• 实现微服务模式
• 创建异步处理工作流
• 编排多服务事务
• 构建实时数据处理管道
• 为分布式事务实现 Saga 模式
• 针对规模扩展和弹性进行设计

AWS 架构良好型无服务器设计原则

1. 快速、简单、专一

函数应简洁且用途单一

// ✅ GOOD - Single purpose, focused function
export const processOrder = async (event: OrderEvent) => {
  // Only handles order processing
  const order = await validateOrder(event);
  await saveOrder(order);
  await publishOrderCreatedEvent(order);
  return { statusCode: 200, body: JSON.stringify({ orderId: order.id }) };
};

// ❌ BAD - Function does too much
export const handleEverything = async (event: any) => {
  // Handles orders, inventory, payments, shipping...
  // Too many responsibilities
};

保持函数的环境效率并关注成本：

• 尽量缩短冷启动时间
• 优化内存分配
• 仅在需要时使用预置并发
• 利用连接复用

2. 关注并发请求，而非请求总数

针对并发而非请求量进行设计

Lambda 通过水平扩展来处理负载，设计时应重点考虑：

• 并发执行限制
• 下游服务限流
• 共享资源竞争
• 连接池大小

// Consider concurrent Lambda executions accessing DynamoDB
const table = new dynamodb.Table(this, 'Table', {
  billingMode: dynamodb.BillingMode.PAY_PER_REQUEST, // Auto-scales with load
});

// Or with provisioned capacity + auto-scaling
const table = new dynamodb.Table(this, 'Table', {
  billingMode: dynamodb.BillingMode.PROVISIONED,
  readCapacity: 5,
  writeCapacity: 5,
});

// Enable auto-scaling for concurrent load
table.autoScaleReadCapacity({ minCapacity: 5, maxCapacity: 100 });
table.autoScaleWriteCapacity({ minCapacity: 5, maxCapacity: 100 });

3. 不共享任何内容

函数运行时环境的生命周期很短

// ❌ BAD - Relying on local file system
export const handler = async (event: any) => {
  fs.writeFileSync('/tmp/data.json', JSON.stringify(data)); // Lost after execution
};

// ✅ GOOD - Use persistent storage
export const handler = async (event: any) => {
  await s3.putObject({
    Bucket: process.env.BUCKET_NAME,
    Key: 'data.json',
    Body: JSON.stringify(data),
  });
};

状态管理：

• 使用 DynamoDB 保存持久状态
• 使用 Step Functions 保存工作流状态
• 使用 ElastiCache 保存会话状态
• 使用 S3 存储文件

4. 不假设与硬件绑定

应用必须与硬件无关

基础设施可能在不另行通知的情况下发生变化：

• Lambda 函数可能运行在不同硬件上
• 容器实例可能会被替换
• 不要对底层基础设施作出假设

针对可移植性进行设计：

• 使用环境变量进行配置
• 避免针对特定硬件进行优化
• 在不同环境中进行测试

5. 使用状态机编排，而不是函数链式调用

使用 Step Functions 进行编排

// ❌ BAD - Lambda function chaining
export const handler1 = async (event: any) => {
  const result = await processStep1(event);
  await lambda.invoke({
    FunctionName: 'handler2',
    Payload: JSON.stringify(result),
  });
};

// ✅ GOOD - Step Functions orchestration
const stateMachine = new stepfunctions.StateMachine(this, 'OrderWorkflow', {
  definition: stepfunctions.Chain
    .start(validateOrder)
    .next(processPayment)
    .next(shipOrder)
    .next(sendConfirmation),
});

Step Functions 的优势：

• 可视化工作流表示
• 内置错误处理和重试机制
• 执行历史记录和调试
• 并行和顺序执行
• 无需编写代码即可集成服务

6. 使用事件触发事务

优先采用事件驱动，而不是同步请求/响应

// Pattern: Event-driven processing
const bucket = new s3.Bucket(this, 'DataBucket');

bucket.addEventNotification(
  s3.EventType.OBJECT_CREATED,
  new s3n.LambdaDestination(processFunction),
  { prefix: 'uploads/' }
);

// Pattern: EventBridge integration
const rule = new events.Rule(this, 'OrderRule', {
  eventPattern: {
    source: ['orders'],
    detailType: ['OrderPlaced'],
  },
});

rule.addTarget(new targets.LambdaFunction(processOrderFunction));

优势：

• 服务之间松耦合
• 异步处理
• 更强的容错能力
• 可独立扩展

7. 针对故障和重复操作进行设计

操作必须具备幂等性

// ✅ GOOD - Idempotent operation
export const handler = async (event: SQSEvent) => {
  for (const record of event.Records) {
    const orderId = JSON.parse(record.body).orderId;

    // Check if already processed (idempotency)
    const existing = await dynamodb.getItem({
      TableName: process.env.TABLE_NAME,
      Key: { orderId },
    });

    if (existing.Item) {
      console.log('Order already processed:', orderId);
      continue; // Skip duplicate
    }

    // Process order
    await processOrder(orderId);

    // Mark as processed
    await dynamodb.putItem({
      TableName: process.env.TABLE_NAME,
      Item: { orderId, processedAt: Date.now() },
    });
  }
};

使用指数退避实现重试逻辑：

async function withRetry<T>(fn: () => Promise<T>, maxRetries = 3): Promise<T> {
  for (let i = 0; i < maxRetries; i++) {
    try {
      return await fn();
    } catch (error) {
      if (i === maxRetries - 1) throw error;
      await new Promise(resolve => setTimeout(resolve, Math.pow(2, i) * 1000));
    }
  }
  throw new Error('Max retries exceeded');
}

事件驱动架构模式

模式 1：事件路由器（EventBridge）

使用 EventBridge 进行事件路由和筛选：

// Create custom event bus
const eventBus = new events.EventBus(this, 'AppEventBus', {
  eventBusName: 'application-events',
});

// Define event schema
const schema = new events.Schema(this, 'OrderSchema', {
  schemaName: 'OrderPlaced',
  definition: events.SchemaDefinition.fromInline({
    openapi: '3.0.0',
    info: { version: '1.0.0', title: 'Order Events' },
    paths: {},
    components: {
      schemas: {
        OrderPlaced: {
          type: 'object',
          properties: {
            orderId: { type: 'string' },
            customerId: { type: 'string' },
            amount: { type: 'number' },
          },
        },
      },
    },
  }),
});

// Create rules for different consumers
new events.Rule(this, 'ProcessOrderRule', {
  eventBus,
  eventPattern: {
    source: ['orders'],
    detailType: ['OrderPlaced'],
  },
  targets: [new targets.LambdaFunction(processOrderFunction)],
});

new events.Rule(this, 'NotifyCustomerRule', {
  eventBus,
  eventPattern: {
    source: ['orders'],
    detailType: ['OrderPlaced'],
  },
  targets: [new targets.LambdaFunction(notifyCustomerFunction)],
});

模式 2：基于队列的处理（SQS）

使用 SQS 实现可靠的异步处理：

// Standard queue for at-least-once delivery
const queue = new sqs.Queue(this, 'ProcessingQueue', {
  visibilityTimeout: Duration.seconds(300),
  retentionPeriod: Duration.days(14),
  deadLetterQueue: {
    queue: dlq,
    maxReceiveCount: 3,
  },
});

// FIFO queue for ordered processing
const fifoQueue = new sqs.Queue(this, 'OrderedQueue', {
  fifo: true,
  contentBasedDeduplication: true,
  deduplicationScope: sqs.DeduplicationScope.MESSAGE_GROUP,
});

// Lambda consumer
new lambda.EventSourceMapping(this, 'QueueConsumer', {
  target: processingFunction,
  eventSourceArn: queue.queueArn,
  batchSize: 10,
  maxBatchingWindow: Duration.seconds(5),
});

模式 3：发布/订阅（SNS + SQS 扇出）

为多个使用者实现扇出模式：

// Create SNS topic
const topic = new sns.Topic(this, 'OrderTopic', {
  displayName: 'Order Events',
});

// Multiple SQS queues subscribe to topic
const inventoryQueue = new sqs.Queue(this, 'InventoryQueue');
const shippingQueue = new sqs.Queue(this, 'ShippingQueue');
const analyticsQueue = new sqs.Queue(this, 'AnalyticsQueue');

topic.addSubscription(new subscriptions.SqsSubscription(inventoryQueue));
topic.addSubscription(new subscriptions.SqsSubscription(shippingQueue));
topic.addSubscription(new subscriptions.SqsSubscription(analyticsQueue));

// Each queue has its own Lambda consumer
new lambda.EventSourceMapping(this, 'InventoryConsumer', {
  target: inventoryFunction,
  eventSourceArn: inventoryQueue.queueArn,
});

模式 4：使用 Step Functions 实现 Saga 模式

实现分布式事务：

const reserveFlight = new tasks.LambdaInvoke(this, 'ReserveFlight', {
  lambdaFunction: reserveFlightFunction,
  outputPath: '$.Payload',
});

const reserveHotel = new tasks.LambdaInvoke(this, 'ReserveHotel', {
  lambdaFunction: reserveHotelFunction,
  outputPath: '$.Payload',
});

const processPayment = new tasks.LambdaInvoke(this, 'ProcessPayment', {
  lambdaFunction: processPaymentFunction,
  outputPath: '$.Payload',
});

// Compensating transactions
const cancelFlight = new tasks.LambdaInvoke(this, 'CancelFlight', {
  lambdaFunction: cancelFlightFunction,
});

const cancelHotel = new tasks.LambdaInvoke(this, 'CancelHotel', {
  lambdaFunction: cancelHotelFunction,
});

// Define saga with compensation
const definition = reserveFlight
  .next(reserveHotel)
  .next(processPayment)
  .addCatch(cancelHotel.next(cancelFlight), {
    resultPath: '$.error',
  });

new stepfunctions.StateMachine(this, 'BookingStateMachine', {
  definition,
  timeout: Duration.minutes(5),
});

模式 5：事件溯源

将事件存储为事实来源：

// Event store with DynamoDB
const eventStore = new dynamodb.Table(this, 'EventStore', {
  partitionKey: { name: 'aggregateId', type: dynamodb.AttributeType.STRING },
  sortKey: { name: 'version', type: dynamodb.AttributeType.NUMBER },
  stream: dynamodb.StreamViewType.NEW_IMAGE,
});

// Lambda function stores events
export const handleCommand = async (event: any) => {
  const { aggregateId, eventType, eventData } = event;

  // Get current version
  const items = await dynamodb.query({
    TableName: process.env.EVENT_STORE,
    KeyConditionExpression: 'aggregateId = :id',
    ExpressionAttributeValues: { ':id': aggregateId },
    ScanIndexForward: false,
    Limit: 1,
  });

  const nextVersion = items.Items?.[0]?.version + 1 || 1;

  // Append new event
  await dynamodb.putItem({
    TableName: process.env.EVENT_STORE,
    Item: {
      aggregateId,
      version: nextVersion,
      eventType,
      eventData,
      timestamp: Date.now(),
    },
  });
};

// Projections read from event stream
eventStore.grantStreamRead(projectionFunction);

无服务器架构模式

模式 1：API 驱动的微服务

使用 Lambda 后端的 REST API：

const api = new apigateway.RestApi(this, 'Api', {
  restApiName: 'microservices-api',
  deployOptions: {
    throttlingRateLimit: 1000,
    throttlingBurstLimit: 2000,
    tracingEnabled: true,
  },
});

// User service
const users = api.root.addResource('users');
users.addMethod('GET', new apigateway.LambdaIntegration(getUsersFunction));
users.addMethod('POST', new apigateway.LambdaIntegration(createUserFunction));

// Order service
const orders = api.root.addResource('orders');
orders.addMethod('GET', new apigateway.LambdaIntegration(getOrdersFunction));
orders.addMethod('POST', new apigateway.LambdaIntegration(createOrderFunction));

模式 2：流处理

使用 Kinesis 实时处理数据：

const stream = new kinesis.Stream(this, 'DataStream', {
  shardCount: 2,
  retentionPeriod: Duration.days(7),
});

// Lambda processes stream records
new lambda.EventSourceMapping(this, 'StreamProcessor', {
  target: processFunction,
  eventSourceArn: stream.streamArn,
  batchSize: 100,
  maxBatchingWindow: Duration.seconds(5),
  parallelizationFactor: 10,
  startingPosition: lambda.StartingPosition.LATEST,
  retryAttempts: 3,
  bisectBatchOnError: true,
  onFailure: new lambdaDestinations.SqsDestination(dlq),
});

模式 3：异步任务处理

后台作业处理：

// SQS queue for tasks
const taskQueue = new sqs.Queue(this, 'TaskQueue', {
  visibilityTimeout: Duration.minutes(5),
  receiveMessageWaitTime: Duration.seconds(20), // Long polling
  deadLetterQueue: {
    queue: dlq,
    maxReceiveCount: 3,
  },
});

// Lambda worker processes tasks
const worker = new lambda.Function(this, 'TaskWorker', {
  // ... configuration
  reservedConcurrentExecutions: 10, // Control concurrency
});

new lambda.EventSourceMapping(this, 'TaskConsumer', {
  target: worker,
  eventSourceArn: taskQueue.queueArn,
  batchSize: 10,
  reportBatchItemFailures: true, // Partial batch failure handling
});

模式 4：定时作业

使用 EventBridge 定期处理：

// Daily cleanup job
new events.Rule(this, 'DailyCleanup', {
  schedule: events.Schedule.cron({ hour: '2', minute: '0' }),
  targets: [new targets.LambdaFunction(cleanupFunction)],
});

// Process every 5 minutes
new events.Rule(this, 'FrequentProcessing', {
  schedule: events.Schedule.rate(Duration.minutes(5)),
  targets: [new targets.LambdaFunction(processFunction)],
});

模式 5：Webhook 处理

处理外部 Webhook：

// API Gateway endpoint for webhooks
const webhookApi = new apigateway.RestApi(this, 'WebhookApi', {
  restApiName: 'webhooks',
});

const webhook = webhookApi.root.addResource('webhook');
webhook.addMethod('POST', new apigateway.LambdaIntegration(webhookFunction, {
  proxy: true,
  timeout: Duration.seconds(29), // API Gateway max
}));

// Lambda handler validates and queues webhook
export const handler = async (event: APIGatewayProxyEvent) => {
  // Validate webhook signature
  const isValid = validateSignature(event.headers, event.body);
  if (!isValid) {
    return { statusCode: 401, body: 'Invalid signature' };
  }

  // Queue for async processing
  await sqs.sendMessage({
    QueueUrl: process.env.QUEUE_URL,
    MessageBody: event.body,
  });

  // Return immediately
  return { statusCode: 202, body: 'Accepted' };
};

最佳实践

错误处理

实现全面的错误处理：

export const handler = async (event: SQSEvent) => {
  const failures: SQSBatchItemFailure[] = [];

  for (const record of event.Records) {
    try {
      await processRecord(record);
    } catch (error) {
      console.error('Failed to process record:', record.messageId, error);
      failures.push({ itemIdentifier: record.messageId });
    }
  }

  // Return partial batch failures for retry
  return { batchItemFailures: failures };
};

死信队列

始终配置 DLQ 以处理错误：

const dlq = new sqs.Queue(this, 'DLQ', {
  retentionPeriod: Duration.days(14),
});

const queue = new sqs.Queue(this, 'Queue', {
  deadLetterQueue: {
    queue: dlq,
    maxReceiveCount: 3,
  },
});

// Monitor DLQ depth
new cloudwatch.Alarm(this, 'DLQAlarm', {
  metric: dlq.metricApproximateNumberOfMessagesVisible(),
  threshold: 1,
  evaluationPeriods: 1,
  alarmDescription: 'Messages in DLQ require attention',
});

可观测性

启用跟踪和监控：

new NodejsFunction(this, 'Function', {
  entry: 'src/handler.ts',
  tracing: lambda.Tracing.ACTIVE, // X-Ray tracing
  environment: {
    POWERTOOLS_SERVICE_NAME: 'order-service',
    POWERTOOLS_METRICS_NAMESPACE: 'MyApp',
    LOG_LEVEL: 'INFO',
  },
});

有效使用 MCP 服务

AWS Serverless MCP 用法

生命周期管理：

• 初始化新的无服务器项目
• 生成 SAM 模板
• 部署应用
• 部署前进行本地测试

Lambda Tool MCP 用法

函数执行：

• 直接测试 Lambda 函数
• 执行自动化工作流
• 访问私有资源
• 验证集成

Step Functions MCP 用法

工作流编排：

• 为复杂工作流创建状态机
• 执行分布式事务
• 实现 Saga 模式
• 协调微服务

SNS/SQS MCP 用法

消息操作：

• 测试发布/订阅模式
• 向队列发送测试消息
• 验证事件路由
• 调试消息处理

其他资源

此技能包含基于 AWS 最佳实践的全面参考文档：

• 无服务器模式：references/serverless-patterns.md

  • 核心无服务器架构和 API 模式
  • 数据处理和集成模式
  • 使用 Step Functions 进行编排
  • 应避免的反模式

• 事件驱动架构模式：references/eda-patterns.md

  • 事件路由和处理模式
  • 事件溯源和 Saga 模式
  • 幂等性和错误处理
  • 消息排序和去重

• 安全最佳实践：references/security-best-practices.md

  • 共同责任模型
  • IAM 最小权限模式
  • 数据保护和加密
  • 使用 VPC 实现网络安全

• 可观测性最佳实践：references/observability-best-practices.md

  • 三大支柱：指标、日志、追踪
  • 使用 Lambda Powertools 进行结构化日志记录
  • X-Ray 分布式追踪
  • CloudWatch 警报和仪表板

• 性能优化：references/performance-optimization.md

  • 冷启动优化技术
  • 内存和 CPU 优化
  • 缩减软件包大小
  • 预置并发模式

• 部署最佳实践：references/deployment-best-practices.md

  • CI/CD 流水线设计
  • 测试策略（单元、集成、负载）
  • 部署策略（金丝雀、蓝绿）
  • 回滚和安全机制

外部资源：

• AWS Well-Architected 无服务器架构指南：https://docs.aws.amazon.com/wellarchitected/latest/serverless-applications-lens/
• ServerlessLand.com：预构建的无服务器模式
• AWS 无服务器工作坊：https://serverlessland.com/learn?type=Workshops

如需了解详细的实现模式、反模式和代码示例，请参阅技能目录中的全面参考资料。