使用AWS Lambda和DynamoDB Streams实现事件驱动系统

随着云原生架构成为主流，开发者越来越多地采用事件驱动设计来构建可扩展且松耦合的应用程序。其中一种强大的模式是将AWS Lambda与DynamoDB Streams结合使用。这种配置能够实现对数据变化的实时、无服务器响应，而无需轮询或手动管理基础设施。

为什么选择事件驱动架构？

事件驱动架构具有以下关键优势：

可扩展性：并行执行和弹性计算
松耦合：组件通过事件通信，而非硬连接集成
响应性：近乎实时地处理变化

当与AWS Lambda等无服务器服务结合使用时，这些优势转化为成本效益高、弹性强且易于维护的系统。

系统架构

核心思想如下：

配置启用Streams的DynamoDB表
当插入、更新或删除行时，会生成流记录
AWS Lambda自动调用并处理这些记录批次
Lambda处理数据并触发下游工作流（如消息传递、分析、更新）

常见用例

想象一个跟踪配置文件更新的系统。当用户更改其详细信息时：

DynamoDB表被更新
通过流触发Lambda函数
Lambda验证更新、记录并推送通知

整个过程完全自动化，无需维护服务器。

实现步骤

步骤1：启用DynamoDB Streams

为表启用Streams并设置适当的视图类型：

1
2
3
4


"StreamSpecification": {
  "StreamEnabled": true,
  "StreamViewType": "NEW_AND_OLD_IMAGES"
}

步骤2：将Lambda连接到Stream

使用AWS控制台或基础设施即代码（如SAM、CDK），在流ARN和Lambda之间创建事件源映射。

步骤3：编写Lambda处理程序

以下是基本的Node.js示例：

1
2
3
4
5
6
7


exports.handler = async (event) => {
  for (const record of event.Records) {
    const newImage = AWS.DynamoDB.Converter.unmarshall(record.dynamodb.NewImage);
    console.log('Processing update:', newImage);
    // 执行业务逻辑
  }
};

步骤4：增强容错能力

重试行为：为失败消息配置DLQ（死信队列）
幂等性：设计逻辑以安全处理重复传递
监控：使用CloudWatch和X-Ray跟踪和记录调用

操作洞察与最佳实践

对延迟敏感的Lambda使用预置并发
调整批次大小和并行度
使用CloudWatch日志、指标和X-Ray
保持函数执行时间在几秒内
DynamoDB Streams不保证跨分片事件的全局顺序。系统必须设计为容忍并正确处理乱序事件处理
流记录最多保留24小时。下游消费者必须及时处理事件以避免数据丢失
确保IAM角色和策略范围严格。过度宽松的配置可能会引入安全风险，尤其是当Lambda与多个AWS服务交互时

适用场景

需要近乎实时地响应数据变化而无需轮询
工作负载是无状态且高度可扩展的，非常适合无服务器执行
解决方案必须与SNS、SQS或Step Functions等其他AWS服务无缝集成

考虑其他方法的情况

系统需要严格、全局的事件顺序
需要支持涉及多个服务或数据库的复杂多步骤事务
应用程序要求保证仅一次处理，这在不自定义幂等性和去重逻辑的情况下难以实现

使用Localstack的概念验证

先决条件

Docker - https://www.docker.com
AWS CLI - https://aws.amazon.com/cli/
awslocal CLI - pip install awscli-local
Python 3.9+

步骤1：Docker Compose设置

在项目根目录创建docker-compose.yml文件：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19


version: '3.8'
services:
  localstack:
    image: localstack/localstack
    ports:
      - "4566:4566"  # LocalStack Gateway
      - "4510-4559:4510-4559"  # External services
    environment:
      - SERVICES=lambda,dynamodb
      - DEFAULT_REGION=us-east-1
      - DATA_DIR=/tmp/localstack/data
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock
      - ./lambda-localstack-project:/lambda-localstack-project
    networks:
      - localstack-network
networks:
  localstack-network:
    driver: bridge

然后启动LocalStack：

1

docker-compose up -d

步骤2：创建启用Streams的DynamoDB表

1
2
3
4
5
6


awslocal dynamodb create-table \
  --table-name UserProfileTable \
  --attribute-definitions AttributeName=id,AttributeType=S \
  --key-schema AttributeName=id,KeyType=HASH \
  --stream-specification StreamEnabled=true,StreamViewType=NEW_AND_OLD_IMAGES \
  --billing-mode PAY_PER_REQUEST

步骤3：编写Lambda处理程序

创建handler.py文件：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17


import json

def lambda_handler(event, context):
    """
    Lambda function to process DynamoDB stream events and print them.
    """
    print("Received event:")
    print(json.dumps(event, indent=2))

    for record in event.get('Records', []):
        print(f"Event ID: {record.get('eventID')}")
        print(f"Event Name: {record.get('eventName')}")
        print(f"DynamoDB Record: {json.dumps(record.get('dynamodb'), indent=2)}")
    return {
        'statusCode': 200,
        'body': 'Event processed successfully'
    }

步骤4：打包Lambda函数

1

zip -r my-lambda-function.zip handler.py

步骤5：创建Lambda函数

1
2
3
4
5
6
7


awslocal lambda create-function \
  --function-name my-lambda-function \
  --runtime python3.9 \
  --role arn:aws:iam::000000000000:role/execution_role \
  --handler handler.lambda_handler \
  --zip-file fileb://function.zip \
  --timeout 30

步骤6：检索Stream ARN

1
2
3
4


awslocal dynamodb describe-table \
  --table-name UserProfileTable \
  --query "Table.LatestStreamArn" \
  --output text

步骤7：创建事件源映射

1
2
3
4
5


awslocal lambda create-event-source-mapping \
  --function-name my-lambda-function \
  --event-source <stream_arn> \
  --batch-size 1 \
  --starting-position TRIM_HORIZON

将<stream_arn>替换为上一步返回的值。

步骤8：向表中添加记录

1
2
3


awslocal dynamodb put-item \
  --table-name UserProfileTable \
  --item '{"id": {"S": "123"}, "name": {"S": "John Doe"}}'

步骤9：检查Docker日志查看Lambda函数打印的消息

输出应类似于以下内容：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53


Received event:
{
  "Records": [
    {
      "eventID": "98fba2f7",
      "eventName": "INSERT",
      "dynamodb": {
        "ApproximateCreationDateTime": 1749085375.0,
        "Keys": {
          "id": {
            "S": "123"
          }
        },
        "NewImage": {
          "id": {
            "S": "123"
          },
          "name": {
            "S": "John Doe"
          }
        },
        "SequenceNumber": "49663951772781148680876496028644551281859231867278983170",
        "SizeBytes": 42,
        "StreamViewType": "NEW_AND_OLD_IMAGES"
      },
      "eventSourceARN": "arn:aws:dynamodb:us-east-1:000000000000:table/UserProfileTable/stream/2025-06-05T01:00:30.711",
      "eventSource": "aws:dynamodb",
      "awsRegion": "us-east-1",
      "eventVersion": "1.1"
    }
  ]
}
Event ID: 98fba2f7
Event Name: INSERT
DynamoDB Record: {
  "ApproximateCreationDateTime": 1749085375.0,
  "Keys": {
    "id": {
      "S": "123"
    }
  },
  "NewImage": {
    "id": {
      "S": "123"
    },
    "name": {
      "S": "John Doe"
    }
  },
  "SequenceNumber": "49663951772781148680876496028644551281859231867278983170",
  "SizeBytes": 42,
  "StreamViewType": "NEW_AND_OLD_IMAGES"
}

总结

至此，您已成功构建了一个功能齐全、本地托管的事件驱动系统，模拟了生产就绪的AWS架构，而无需离开开发环境。

此实现展示了如何使用DynamoDB Streams捕获数据存储中的实时变化，以及如何使用AWS Lambda（一种无服务器计算服务）高效处理这些变化。通过结合LocalStack和Docker Compose，您创建了一个本地开发环境，模拟了关键的AWS服务，实现了快速迭代、测试和调试。

这些组件共同为构建现代事件驱动应用程序提供了可扩展、经济高效的基础。此设置非常适合异步处理、审计日志记录、数据丰富、实时通知等用例，同时遵循微服务和云原生设计的最佳实践。

有了这个基础，您可以通过集成其他AWS服务（如SNS、SQS、EventBridge或Step Functions）来扩展架构，以支持更复杂的工作流和企业级可扩展性。

结论

AWS Lambda和DynamoDB Streams共同为在云原生应用程序中实现事件驱动架构提供了强大的基础。通过实现对数据变化的实时响应，而无需持久服务器或轮询机制，这种组合降低了操作负担并加速了开发周期。开发者可以专注于编写业务逻辑，而AWS则处理扩展、容错和基础设施管理的繁重工作。

只需几个配置步骤，您就可以构建响应数据层中创建、更新或删除事件的工作流。无论您是丰富数据、触发通知、审计活动还是编排下游服务，这种无服务器方法允许您每天处理数百万个事件，同时保持高可用性和低成本。

除了技术优势外，事件驱动架构还促进了清晰的关注点分离、改进的系统响应能力和更大的灵活性。它使团队能够构建可以独立发展的松耦合服务，非常适合微服务和分布式系统。