示例工程【免费下载链接】examplesServerless Examples – A collection of boilerplates and examples of serverless architectures built with the Serverless Framework on AWS Lambda, Microsoft Azure, Google Cloud Functions, and more.项目地址https://gitcode.com/gh_mirrors/exa/examples点击查看免费下载本文是一份基于 examples 仓库中aws-ruby-sqs-with-dynamodb示例的完整实战指南。该示例以「抽奖优惠券lottery coupon」为业务场景演示了如何用 Serverless Framework v4 将 API Gateway、SQS、AWS Lambda 与 DynamoDB 串成一条「同步接收、异步落库」的生产者-消费者链路并借助 Lift 的 queue 构造与 CloudWatch 完成队列、死信与告警管理。读完本文你将掌握 Ruby Lambda 中 dry-schema 请求校验、SQS 消息生产与消费、DynamoDB 写入以及该架构的部署、测试与运维全流程。架构总览一张图讲清全链路示例的业务模型非常聚焦每张抽奖优惠券由id、first_name、last_name、coupon_value以及created_at时间戳五个字段构成。客户端通过 API Gateway 提交表单数据网关触发 Lambda 生产者函数将消息推入 SQS 队列随后队列触发的 Lambda 消费者worker从队列拉取消息并在 DynamoDB 表中创建一条新的优惠券记录全程由 CloudWatch 负责日志与告警。组件清单与职责组件在本示例中的职责API Gateway暴露POST /lotteryHTTP 端点触发createLotteryCoupon生产者函数AWS Lambda生产者校验请求参数将合法请求封装为消息发送到 SQSSQS 队列由 Lift 的 queue 构造创建作为生产者与消费者之间的异步缓冲AWS Lambdaworker订阅队列逐条消费消息并写入 DynamoDBDynamoDB存储优惠券记录表名由「服务名 stage」拼接而成SQS DLQ死信队列存放处理失败的消息供人工排查或重试CloudWatch采集 Lambda 运行日志DLQ 收到失败消息且告警开启时发送邮件通知端到端请求流程客户端向POST https://…/dev/lottery发送 JSON 请求体first_name、last_name、coupon_value。createLotteryCoupon函数handler.rb解析请求体并用 dry-schema 校验参数合法性。校验失败立即返回422 Unprocessable Entity与结构化错误校验通过调用CreateSqsMessageService把参数序列化为 JSON 消息通过SqsAdapter发送到队列。客户端立即收到{message: Message sent}无需等待落库——这正是异步解耦的价值。worker 函数worker.rb被 SQS 触发取出消息体由CreateLotteryCouponService补全id与created_at后经DynamoDBAdapter调用put_item写入 DynamoDB。Lift queue 构造带来的开箱能力SQS 相关能力并非手写 CloudFormation而是通过 serverless-lift 中以constructs声明式使用。Lift 的 queue 构造会自动创建SQS 队列作为生产者与 worker 之间的消息缓冲worker Lambda 函数绑定队列事件逐条处理队列中的消息SQS DLQ死信队列所有处理失败的消息都会转移到这里避免阻塞主队列CloudWatch 告警当 DLQ 中存在失败消息且alarm选项启用时本示例中为开启状态通过邮件发出告警。项目结构速览aws-ruby-sqs-with-dynamodb/ ├── Gemfile # Ruby 依赖aws-sdk、dry-schema ├── serverless.yml # 服务定义、Lift 构造与 DynamoDB 表资源 ├── package.json # 插件依赖serverless-lift、serverless-ruby-layer └── src/ ├── handlers/ │ └── lottery/ │ ├── handler.rb # 生产者接收 HTTP 请求并发送 SQS 消息 │ └── worker.rb # 消费者消费 SQS 消息并写入 DynamoDB └── common/ ├── adapters/ # 基础设施适配层 ├── helpers/ # 通用工具方法 ├── schemas/ # dry-schema 数据结构定义 ├── serializers/ # 错误响应序列化 ├── services/ # 单一职责的服务对象 └── validators/ # 请求校验器原文档给出的完整结构说明如下与实际源码一一对应路径说明./src项目全部业务代码./src/handlers/handler生产 SQS 消息的 Lambda 函数实际文件为 src/handlers/lottery/handler.rb./src/handlers/worker消费 SQS 消息的 Lambda 函数实际文件为 src/handlers/lottery/worker.rb./src/common/公共可复用代码./src/common/adapters/dynamo_db_adapter.rb基于 AWS SDK for Ruby 的 DynamoDB 通信适配器本示例仅用于新建记录./src/common/adapters/sqs_adapter.rb基于 AWS SDK for Ruby 的 SQS 通信适配器本示例仅用于发送新消息./src/common/services/create_lottery_coupon_service.rb服务对象模式Ruby/Rails 开发者常用一个类只做一件事此处负责向 DynamoDB 创建优惠券记录./src/common/services/create_sqs_message_service.rb负责向 SQS 发送消息./src/common/helpers/requests_helper.rb请求辅助方法此处用于解析并符号化symbolize哈希键./src/common/schemas/lottery_coupon_schema.rb应用内实体结构定义此处声明优惠券的必填键与数据类型./src/common/serializers/error_serializer.rb422 错误序列化器将 dry-schema 校验错误整理为规范结构./src/common/validators/lottery_coupon_validator.rb优惠券记录校验器基于 dry-schema 校验入参并返回错误信息依赖选型两个插件 三个 GemServerless 插件package.json插件说明serverless-ruby-layer将Gemfile中的 Ruby gems 打包成 Lambda layer 并随函数部署serverless-lift在 Serverless Framework 中使用 AWS CDK 构造本示例使用其中的 queue 构造两个插件均为项目依赖package.json 中锁定版本为serverless-lift: ^1.1.1、serverless-ruby-layer: ^1.5.0。Ruby gemsGemfileGem说明aws-sdk-dynamodbAWS SDK for Ruby 的 DynamoDB 组件用于创建新记录aws-sdk-sqsAWS SDK for Ruby 的 SQS 组件用于生产与处理队列消息dry-schema~ 1.6, 1.6.2校验入参的结构与数据类型serverless.yml 深度拆解整个服务定义集中在 serverless.yml下面按段落逐一解读。服务与 Provider 基础配置service: serverless-ruby-sqs-dynamodb frameworkVersion: 4 provider: name: aws runtime: ruby4.0 architecture: arm64 memorySize: 256 iam: role: statements: - Effect: Allow Action: - dynamodb:PutItem Resource: - !GetAtt Table.Arnservice是资源命名的根最终 DynamoDB 表名、栈名、队列名都以它为前缀frameworkVersion: 4限定使用 Serverless Framework v4runtime: ruby4.0与architecture: arm64指定 Ruby 运行时与 64 位 ARM 架构memorySize: 256设定 Lambda 内存兼顾性能与成本IAM 采用最小权限原则仅授予dynamodb:PutItem且资源限定在!GetAtt Table.Arn即下方resources中定义的 DynamoDB 表两个函数共用这一角色。Lift queue 构造constructs: lotteryQueue: type: queue worker: handler: src/handlers/lottery/worker.run environment: TABLE_NAME: ${self:custom.tableName} alarm: ${self:custom.alertEmail} batchSize: 5lotteryQueue是构造实例名Lift 会据此生成 SQS 队列、worker Lambda、DLQ 与告警worker.handler指向src/handlers/lottery/worker.rb中的run方法worker.environment.TABLE_NAME通过${self:custom.tableName}注入表名供 worker 运行时读取alarm设为${self:custom.alertEmail}默认mailexample.com一旦 DLQ 积累失败消息即触发邮件告警batchSize: 5控制每次向 worker 批量投递 5 条消息与标准 SQS 批处理语义一致。生产者函数functions: createLotteryCoupon: handler: src/handlers/lottery/handler.run events: - http: method: post path: lottery environment: QUEUE_URL: ${construct:lotteryQueue.queueUrl}函数名createLotteryCoupon处理器为 handler.rb 的run方法通过http事件暴露POST /lotteryREST 端点由 API Gateway 承载关键点QUEUE_URL通过${construct:lotteryQueue.queueUrl}直接引用 Lift 构造生成的队列 URL 并注入环境变量生产者据此发送消息。DynamoDB 表资源resources: Resources: Table: Type: AWS::DynamoDB::Table Properties: TableName: ${self:custom.tableName} Tags: - Key: Application Value: ${self:service} - Key: Stage Value: ${sls:stage} BillingMode: PAY_PER_REQUEST AttributeDefinitions: - AttributeName: id AttributeType: S KeySchema: - AttributeName: id KeyType: HASH表名由custom.tableName即service-stage拼接决定附带Application、Stage两个标签便于成本归属与资源检索BillingMode: PAY_PER_REQUEST采用按需计费无需预置读写容量适合抽奖这类突发流量场景主键只定义一个id字符串类型KeyType: HASH恰好对应CreateLotteryCouponService中用 UUID 生成的记录主键。其余自定义配置custom: tableName: ${self:service}-${sls:stage} rubyLayer: include_functions: - createLotteryCoupon - lotteryQueueWorker alertEmail: mailexample.comrubyLayer.include_functions指定需要挂载 Ruby gems layer 的两个函数alertEmail为告警接收邮箱按需改为真实地址即可。生产者链路从 HTTP 请求到 SQS 消息入口函数handler.rb 是整个链路的入口def run(event:, context:) body transform_event_body(event[body]) validation LotteryCouponValidator.new(body) return unprocessable_entity(validation.errors) if validation.failure? create_sqs_message(body) end流程为解析事件体 → 校验 → 失败返回 422 / 成功发送消息。其中transform_event_body借助 requests_helper.rb 完成 JSON 解析与键符号化def transform_body(body) JSON.parse(body).transform_keys(:to_sym) enddry-schema 请求校验校验的核心规则定义在 lottery_coupon_schema.rbLotteryCouponSchema Dry::Schema.Params do required(:first_name).filled(:string) required(:last_name).filled(:string) required(:coupon_value).value(:integer, gt?: 0) end三个字段均为必填前两者必须是非空字符串coupon_value必须是大于 0 的整数——从 schema 层拦截了非法请求。校验器 lottery_coupon_validator.rb 封装对 schema 的调用def initialize(coupon) coupon coupon result LotteryCouponSchema.call(coupon) end def failure? result.failure? end def errors result.errors.to_h end422 错误的结构化输出校验失败时error_serializer.rb 会把 dry-schema 的错误哈希转换为 JSON API 风格的结构def self.dry_errors(exception) exception.map do |field, error_message_array| { status: 422, source: { pointer: field }, detail: error_message_array[0] } end.flatten end最终由 handler 组装为422响应例如{ errors: [ { status: 422, source: { pointer: coupon_value }, detail: must be greater than 0 } ] }发送消息到 SQS校验通过后create_sqs_message_service.rb 将三个字段序列化为 JSONdef sqs_message(params) { first_name: params[:first_name], last_name: params[:last_name], coupon_value: params[:coupon_value] }.to_json end底层发送由 sqs_adapter.rb 完成def send_message(message) client.send_message( queue_url: ENV[QUEUE_URL], message_body: message ) logger.info(Created SQS message) rescue Aws::SQS::Errors::ServiceError error logger.error(error) end注意queue_url取自环境变量QUEUE_URL这正是 serverless.yml 中${construct:lotteryQueue.queueUrl}注入的值。发送成功后 handler 返回200 {message: Message sent}任何异常则兜底返回500。消费者链路从 SQS 消息到 DynamoDB 记录worker 函数 worker.rb 由 SQS 队列触发代码非常精简def run(event:, context:) params transform_event_body(event[Records][0][body]) CreateLotteryCouponService.new(params).call end它从 SQS 事件中取出Records[0].body即生产者写入的 JSON同样经Requests解析为符号化哈希再交给服务对象落库。create_lottery_coupon_service.rb 负责补全实体字段并保存LotteryCouponSchema Struct.new(:id, :first_name, :last_name, :coupon_value, :created_at) def build_coupon LotteryCouponSchema.new(SecureRandom.uuid, attributes[:first_name], attributes[:last_name], attributes[:coupon_value], DateTime.now.iso8601).to_h end def save(coupon) DynamoDBAdapter.new.save_item(coupon) endid用SecureRandom.uuid生成全局唯一主键与 DynamoDB 表的 HASH 键设计吻合created_at用DateTime.now.iso8601生成 ISO 8601 时间戳字段顺序固定为id, first_name, last_name, coupon_value, created_at与 README 描述的记录结构一致。写入动作最终落在 dynamo_db_adapter.rbdef save_item(item) client.put_item(table_item(item)) logger.info(Created dynamoDB item with id#{item[:id]}) rescue Aws::DynamoDB::Errors::ServiceError error logger.error(error) end def table_item(item) { table_name: ENV[TABLE_NAME], item: item } end表名同样来自环境变量TABLE_NAME由 Lift 构造在 serverless.yml 中通过environment.TABLE_NAME: ${self:custom.tableName}注入与resources中声明的 DynamoDB 表保持一致。安装、部署与端到端验证安装依赖npm install安装 Serverless Framework 插件serverless-lift、serverless-ruby-layer及其传递依赖。部署sls deploy使用~/.aws/credentials中指定的 profile 部署AWS_PROFILEYOUR_PROFILE_NAME sls deploy部署到指定 stage例如stagingsls deploy --stage staging注意 stage 会影响资源命名DynamoDB 表名、Lambda 名称、队列名均会带上 stage 后缀。部署成功的输出类似Service Information service: serverless-ruby-sqs-dynamodb stage: dev region: your-region-here stack: serverless-ruby-sqs-dynamodb-dev resources: 22 api keys: None endpoints: POST - https://XXXXXXXXXX.execute-api.your-region-here.amazonaws.com/dev/lottery functions: createLotteryCoupon: serverless-ruby-sqs-dynamodb-dev-createLotteryCoupon lotteryQueueWorker: serverless-ruby-sqs-dynamodb-dev-lotteryQueueWorker layers: gem: arn:aws:lambda:your-region-here:XXXXXXXXXXXXXX:layer:serverless-ruby-sqs-dynamodb-dev-ruby-bundle:3 lotteryQueue: queueUrl: https://sqs.your-region-here.amazonaws.com/XXXXXXXXXXXX/serverless-ruby-sqs-dynamodb-dev-lotteryQueue关键信息解读输出项含义endpointsAPI Gateway 暴露的POST /dev/lottery地址即客户端调用入口functions两个 Lambda生产者createLotteryCoupon与队列 workerlotteryQueueWorkerlayers.gemserverless-ruby-layer 将 Gemfile 打包成的 Ruby layer ARN函数通过 GEM_PATH 使用lotteryQueue.queueUrlLift 创建的 SQS 队列 URL部署过程中即被注入生产者的QUEUE_URL部署日志中「Zipping the gemfiles to …/ruby_layer/gemLayer.zip」「Configuring Layer and GEM_PATH to the functions」等步骤正是 serverless-ruby-layer 插件在后台完成的 gem 打包与 layer 挂载。发送测试请求部署完成后即可调用端点curl 或 Postman 均可curl --request POST https://XXXXXXXXXX.execute-api.your-region-here.amazonaws.com/dev/lottery --header Content-Type: application/json --data-raw {first_name: Daniel, last_name: Ani, coupon_value: 99}预期响应{ message: Message sent }在 DynamoDB 中验证记录请求成功只代表消息已进入 SQS 队列落库由 worker 异步完成。稍等片刻后打开 DynamoDB 控制台即可看到 worker 创建的记录包含idUUID、first_name、last_name、coupon_value与created_at五个字段注意DynamoDB 表名是「服务名 stage」的组合。以devstage 为例表名为serverless-ruby-sqs-dynamodb-dev监控与告警运维查看 CloudWatch 日志排查生产与消费两条链路时可以从 Lambda 控制台进入对应函数查看 CloudWatch 日志createLotteryCoupon与lotteryQueueWorker均可也可以直接在 CloudWatch Logs 中按日志组查看两条链路分别会输出Created SQS message生产者侧与Created dynamoDB item with id…worker 侧等日志配合 CloudWatch 可以清晰追踪消息从进队到落库的完整过程。配置 SQS 死信队列邮件告警当消息处理失败被送入 DLQ 时可以配置邮件告警。修改 serverless.yml 中 queue 构造的alarm值为你的邮箱即可constructs: lotteryQueue: type: queue alarm: your-emailexample.com # 替换默认的 mailexample.com部署生效后AWS 会向该邮箱发送订阅确认邮件务必点击确认订阅AWS Notification 通知告警才会真正生效。调整日志保留期默认日志保留期为 30 天如需调整修改 serverless.yml 中的logRetentionInDays属性即可logRetentionInDays: 30补充说明当前仓库的 serverless.yml 中未显式列出该字段原文档指出该值位于 serverless.yml读者可按上述语法自行添加或调整以匹配自身日志审计与成本策略。扩展与清理从 Lift 插件安全脱离EjectLift 一个很有价值的特性是 eject由于 Lift 底层就是 CloudFormation它不会把项目「锁死」在插件里。当项目规模超出插件能力时可以随时执行 eject 把构造展开为原生 CloudFormation 资源之后不再依赖 serverless-lift 也能继续部署演进。移除服务不再需要时一条命令即可清理全部资源sls remove该命令会删除 CloudFormation 栈及其管理的 Lambda、DynamoDB 表、SQS 队列、DLQ、日志组与告警等全部资源。小结通过这个示例可以看到一条完整的异步事件驱动链路API Gateway 同步接收并快速应答SQS 解耦生产与消费节奏worker 在后台可靠落库DLQ 与 CloudWatch 兜底失败场景。源码层面handler.rb 与 worker.rb 保持薄入口业务逻辑下沉到 services / adapters / validators 等职责单一的文件schema 校验、错误序列化、基础设施适配彼此解耦是 Ruby Serverless 项目中值得直接参考的分层写法。赞分享示例工程【免费下载链接】examplesServerless Examples – A collection of boilerplates and examples of serverless architectures built with the Serverless Framework on AWS Lambda, Microsoft Azure, Google Cloud Functions, and more.项目地址https://gitcode.com/gh_mirrors/exa/examples点击查看免费下载相关推荐aws-ruby-cron-with-dynamodb 实战用 Serverless Framework 在 AWS 上构建 Ruby 定时任务并写入 DynamoDBaws ruby cron with dynamodb 实战用 Serverless Framework 在 AWS 上构建 Ruby 定时任务并写入 Dyn示例工程用 Serverless Framework 在 AWS Lambda 上构建 GraphQL DynamoDB APINode.js 实战用 Serverless Framework 在 AWS Lambda 上构建 GraphQL DynamoDB APINode.js 实战 本文以仓库示例工程Chalice Todo 示例应用实战用 AWS Lambda API Gateway DynamoDB 构建带 JWT 认证的 Serverless REST APIChalice Todo 示例应用实战用 AWS Lambda API Gateway DynamoDB 构建带 JWT 认证的 Serverless后端云原生上一篇Interview Script: [Research Topic]下一篇Ant Design Blazor TreeSelect 勾选回填策略ShowCheckedStrategy完整实战指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 
长河君 · 2026/10/14 3:08:44 · 阅读 1.2k Serverless AWS Ruby SQS with DynamoDB 实战：用 API Gateway + SQS + Lambda 构建 DynamoDB 异步写入链路
示例工程【免费下载链接】examplesServerless Examples – A collection of boilerplates and examples of serverless architectures built with the Serverless Framework on AWS Lambda, Microsoft Azure, Google Cloud Functions, and more.项目地址https://gitcode.com/gh_mirrors/exa/examples点击查看免费下载本文是一份基于 examples 仓库中aws-ruby-sqs-with-dynamodb示例的完整实战指南。该示例以「抽奖优惠券lottery coupon」为业务场景演示了如何用 Serverless Framework v4 将 API Gateway、SQS、AWS Lambda 与 DynamoDB 串成一条「同步接收、异步落库」的生产者-消费者链路并借助 Lift 的 queue 构造与 CloudWatch 完成队列、死信与告警管理。读完本文你将掌握 Ruby Lambda 中 dry-schema 请求校验、SQS 消息生产与消费、DynamoDB 写入以及该架构的部署、测试与运维全流程。架构总览一张图讲清全链路示例的业务模型非常聚焦每张抽奖优惠券由id、first_name、last_name、coupon_value以及created_at时间戳五个字段构成。客户端通过 API Gateway 提交表单数据网关触发 Lambda 生产者函数将消息推入 SQS 队列随后队列触发的 Lambda 消费者worker从队列拉取消息并在 DynamoDB 表中创建一条新的优惠券记录全程由 CloudWatch 负责日志与告警。组件清单与职责组件在本示例中的职责API Gateway暴露POST /lotteryHTTP 端点触发createLotteryCoupon生产者函数AWS Lambda生产者校验请求参数将合法请求封装为消息发送到 SQSSQS 队列由 Lift 的 queue 构造创建作为生产者与消费者之间的异步缓冲AWS Lambdaworker订阅队列逐条消费消息并写入 DynamoDBDynamoDB存储优惠券记录表名由「服务名 stage」拼接而成SQS DLQ死信队列存放处理失败的消息供人工排查或重试CloudWatch采集 Lambda 运行日志DLQ 收到失败消息且告警开启时发送邮件通知端到端请求流程客户端向POST https://…/dev/lottery发送 JSON 请求体first_name、last_name、coupon_value。createLotteryCoupon函数handler.rb解析请求体并用 dry-schema 校验参数合法性。校验失败立即返回422 Unprocessable Entity与结构化错误校验通过调用CreateSqsMessageService把参数序列化为 JSON 消息通过SqsAdapter发送到队列。客户端立即收到{message: Message sent}无需等待落库——这正是异步解耦的价值。worker 函数worker.rb被 SQS 触发取出消息体由CreateLotteryCouponService补全id与created_at后经DynamoDBAdapter调用put_item写入 DynamoDB。Lift queue 构造带来的开箱能力SQS 相关能力并非手写 CloudFormation而是通过 serverless-lift 中以constructs声明式使用。Lift 的 queue 构造会自动创建SQS 队列作为生产者与 worker 之间的消息缓冲worker Lambda 函数绑定队列事件逐条处理队列中的消息SQS DLQ死信队列所有处理失败的消息都会转移到这里避免阻塞主队列CloudWatch 告警当 DLQ 中存在失败消息且alarm选项启用时本示例中为开启状态通过邮件发出告警。项目结构速览aws-ruby-sqs-with-dynamodb/ ├── Gemfile # Ruby 依赖aws-sdk、dry-schema ├── serverless.yml # 服务定义、Lift 构造与 DynamoDB 表资源 ├── package.json # 插件依赖serverless-lift、serverless-ruby-layer └── src/ ├── handlers/ │ └── lottery/ │ ├── handler.rb # 生产者接收 HTTP 请求并发送 SQS 消息 │ └── worker.rb # 消费者消费 SQS 消息并写入 DynamoDB └── common/ ├── adapters/ # 基础设施适配层 ├── helpers/ # 通用工具方法 ├── schemas/ # dry-schema 数据结构定义 ├── serializers/ # 错误响应序列化 ├── services/ # 单一职责的服务对象 └── validators/ # 请求校验器原文档给出的完整结构说明如下与实际源码一一对应路径说明./src项目全部业务代码./src/handlers/handler生产 SQS 消息的 Lambda 函数实际文件为 src/handlers/lottery/handler.rb./src/handlers/worker消费 SQS 消息的 Lambda 函数实际文件为 src/handlers/lottery/worker.rb./src/common/公共可复用代码./src/common/adapters/dynamo_db_adapter.rb基于 AWS SDK for Ruby 的 DynamoDB 通信适配器本示例仅用于新建记录./src/common/adapters/sqs_adapter.rb基于 AWS SDK for Ruby 的 SQS 通信适配器本示例仅用于发送新消息./src/common/services/create_lottery_coupon_service.rb服务对象模式Ruby/Rails 开发者常用一个类只做一件事此处负责向 DynamoDB 创建优惠券记录./src/common/services/create_sqs_message_service.rb负责向 SQS 发送消息./src/common/helpers/requests_helper.rb请求辅助方法此处用于解析并符号化symbolize哈希键./src/common/schemas/lottery_coupon_schema.rb应用内实体结构定义此处声明优惠券的必填键与数据类型./src/common/serializers/error_serializer.rb422 错误序列化器将 dry-schema 校验错误整理为规范结构./src/common/validators/lottery_coupon_validator.rb优惠券记录校验器基于 dry-schema 校验入参并返回错误信息依赖选型两个插件 三个 GemServerless 插件package.json插件说明serverless-ruby-layer将Gemfile中的 Ruby gems 打包成 Lambda layer 并随函数部署serverless-lift在 Serverless Framework 中使用 AWS CDK 构造本示例使用其中的 queue 构造两个插件均为项目依赖package.json 中锁定版本为serverless-lift: ^1.1.1、serverless-ruby-layer: ^1.5.0。Ruby gemsGemfileGem说明aws-sdk-dynamodbAWS SDK for Ruby 的 DynamoDB 组件用于创建新记录aws-sdk-sqsAWS SDK for Ruby 的 SQS 组件用于生产与处理队列消息dry-schema~ 1.6, 1.6.2校验入参的结构与数据类型serverless.yml 深度拆解整个服务定义集中在 serverless.yml下面按段落逐一解读。服务与 Provider 基础配置service: serverless-ruby-sqs-dynamodb frameworkVersion: 4 provider: name: aws runtime: ruby4.0 architecture: arm64 memorySize: 256 iam: role: statements: - Effect: Allow Action: - dynamodb:PutItem Resource: - !GetAtt Table.Arnservice是资源命名的根最终 DynamoDB 表名、栈名、队列名都以它为前缀frameworkVersion: 4限定使用 Serverless Framework v4runtime: ruby4.0与architecture: arm64指定 Ruby 运行时与 64 位 ARM 架构memorySize: 256设定 Lambda 内存兼顾性能与成本IAM 采用最小权限原则仅授予dynamodb:PutItem且资源限定在!GetAtt Table.Arn即下方resources中定义的 DynamoDB 表两个函数共用这一角色。Lift queue 构造constructs: lotteryQueue: type: queue worker: handler: src/handlers/lottery/worker.run environment: TABLE_NAME: ${self:custom.tableName} alarm: ${self:custom.alertEmail} batchSize: 5lotteryQueue是构造实例名Lift 会据此生成 SQS 队列、worker Lambda、DLQ 与告警worker.handler指向src/handlers/lottery/worker.rb中的run方法worker.environment.TABLE_NAME通过${self:custom.tableName}注入表名供 worker 运行时读取alarm设为${self:custom.alertEmail}默认mailexample.com一旦 DLQ 积累失败消息即触发邮件告警batchSize: 5控制每次向 worker 批量投递 5 条消息与标准 SQS 批处理语义一致。生产者函数functions: createLotteryCoupon: handler: src/handlers/lottery/handler.run events: - http: method: post path: lottery environment: QUEUE_URL: ${construct:lotteryQueue.queueUrl}函数名createLotteryCoupon处理器为 handler.rb 的run方法通过http事件暴露POST /lotteryREST 端点由 API Gateway 承载关键点QUEUE_URL通过${construct:lotteryQueue.queueUrl}直接引用 Lift 构造生成的队列 URL 并注入环境变量生产者据此发送消息。DynamoDB 表资源resources: Resources: Table: Type: AWS::DynamoDB::Table Properties: TableName: ${self:custom.tableName} Tags: - Key: Application Value: ${self:service} - Key: Stage Value: ${sls:stage} BillingMode: PAY_PER_REQUEST AttributeDefinitions: - AttributeName: id AttributeType: S KeySchema: - AttributeName: id KeyType: HASH表名由custom.tableName即service-stage拼接决定附带Application、Stage两个标签便于成本归属与资源检索BillingMode: PAY_PER_REQUEST采用按需计费无需预置读写容量适合抽奖这类突发流量场景主键只定义一个id字符串类型KeyType: HASH恰好对应CreateLotteryCouponService中用 UUID 生成的记录主键。其余自定义配置custom: tableName: ${self:service}-${sls:stage} rubyLayer: include_functions: - createLotteryCoupon - lotteryQueueWorker alertEmail: mailexample.comrubyLayer.include_functions指定需要挂载 Ruby gems layer 的两个函数alertEmail为告警接收邮箱按需改为真实地址即可。生产者链路从 HTTP 请求到 SQS 消息入口函数handler.rb 是整个链路的入口def run(event:, context:) body transform_event_body(event[body]) validation LotteryCouponValidator.new(body) return unprocessable_entity(validation.errors) if validation.failure? create_sqs_message(body) end流程为解析事件体 → 校验 → 失败返回 422 / 成功发送消息。其中transform_event_body借助 requests_helper.rb 完成 JSON 解析与键符号化def transform_body(body) JSON.parse(body).transform_keys(:to_sym) enddry-schema 请求校验校验的核心规则定义在 lottery_coupon_schema.rbLotteryCouponSchema Dry::Schema.Params do required(:first_name).filled(:string) required(:last_name).filled(:string) required(:coupon_value).value(:integer, gt?: 0) end三个字段均为必填前两者必须是非空字符串coupon_value必须是大于 0 的整数——从 schema 层拦截了非法请求。校验器 lottery_coupon_validator.rb 封装对 schema 的调用def initialize(coupon) coupon coupon result LotteryCouponSchema.call(coupon) end def failure? result.failure? end def errors result.errors.to_h end422 错误的结构化输出校验失败时error_serializer.rb 会把 dry-schema 的错误哈希转换为 JSON API 风格的结构def self.dry_errors(exception) exception.map do |field, error_message_array| { status: 422, source: { pointer: field }, detail: error_message_array[0] } end.flatten end最终由 handler 组装为422响应例如{ errors: [ { status: 422, source: { pointer: coupon_value }, detail: must be greater than 0 } ] }发送消息到 SQS校验通过后create_sqs_message_service.rb 将三个字段序列化为 JSONdef sqs_message(params) { first_name: params[:first_name], last_name: params[:last_name], coupon_value: params[:coupon_value] }.to_json end底层发送由 sqs_adapter.rb 完成def send_message(message) client.send_message( queue_url: ENV[QUEUE_URL], message_body: message ) logger.info(Created SQS message) rescue Aws::SQS::Errors::ServiceError error logger.error(error) end注意queue_url取自环境变量QUEUE_URL这正是 serverless.yml 中${construct:lotteryQueue.queueUrl}注入的值。发送成功后 handler 返回200 {message: Message sent}任何异常则兜底返回500。消费者链路从 SQS 消息到 DynamoDB 记录worker 函数 worker.rb 由 SQS 队列触发代码非常精简def run(event:, context:) params transform_event_body(event[Records][0][body]) CreateLotteryCouponService.new(params).call end它从 SQS 事件中取出Records[0].body即生产者写入的 JSON同样经Requests解析为符号化哈希再交给服务对象落库。create_lottery_coupon_service.rb 负责补全实体字段并保存LotteryCouponSchema Struct.new(:id, :first_name, :last_name, :coupon_value, :created_at) def build_coupon LotteryCouponSchema.new(SecureRandom.uuid, attributes[:first_name], attributes[:last_name], attributes[:coupon_value], DateTime.now.iso8601).to_h end def save(coupon) DynamoDBAdapter.new.save_item(coupon) endid用SecureRandom.uuid生成全局唯一主键与 DynamoDB 表的 HASH 键设计吻合created_at用DateTime.now.iso8601生成 ISO 8601 时间戳字段顺序固定为id, first_name, last_name, coupon_value, created_at与 README 描述的记录结构一致。写入动作最终落在 dynamo_db_adapter.rbdef save_item(item) client.put_item(table_item(item)) logger.info(Created dynamoDB item with id#{item[:id]}) rescue Aws::DynamoDB::Errors::ServiceError error logger.error(error) end def table_item(item) { table_name: ENV[TABLE_NAME], item: item } end表名同样来自环境变量TABLE_NAME由 Lift 构造在 serverless.yml 中通过environment.TABLE_NAME: ${self:custom.tableName}注入与resources中声明的 DynamoDB 表保持一致。安装、部署与端到端验证安装依赖npm install安装 Serverless Framework 插件serverless-lift、serverless-ruby-layer及其传递依赖。部署sls deploy使用~/.aws/credentials中指定的 profile 部署AWS_PROFILEYOUR_PROFILE_NAME sls deploy部署到指定 stage例如stagingsls deploy --stage staging注意 stage 会影响资源命名DynamoDB 表名、Lambda 名称、队列名均会带上 stage 后缀。部署成功的输出类似Service Information service: serverless-ruby-sqs-dynamodb stage: dev region: your-region-here stack: serverless-ruby-sqs-dynamodb-dev resources: 22 api keys: None endpoints: POST - https://XXXXXXXXXX.execute-api.your-region-here.amazonaws.com/dev/lottery functions: createLotteryCoupon: serverless-ruby-sqs-dynamodb-dev-createLotteryCoupon lotteryQueueWorker: serverless-ruby-sqs-dynamodb-dev-lotteryQueueWorker layers: gem: arn:aws:lambda:your-region-here:XXXXXXXXXXXXXX:layer:serverless-ruby-sqs-dynamodb-dev-ruby-bundle:3 lotteryQueue: queueUrl: https://sqs.your-region-here.amazonaws.com/XXXXXXXXXXXX/serverless-ruby-sqs-dynamodb-dev-lotteryQueue关键信息解读输出项含义endpointsAPI Gateway 暴露的POST /dev/lottery地址即客户端调用入口functions两个 Lambda生产者createLotteryCoupon与队列 workerlotteryQueueWorkerlayers.gemserverless-ruby-layer 将 Gemfile 打包成的 Ruby layer ARN函数通过 GEM_PATH 使用lotteryQueue.queueUrlLift 创建的 SQS 队列 URL部署过程中即被注入生产者的QUEUE_URL部署日志中「Zipping the gemfiles to …/ruby_layer/gemLayer.zip」「Configuring Layer and GEM_PATH to the functions」等步骤正是 serverless-ruby-layer 插件在后台完成的 gem 打包与 layer 挂载。发送测试请求部署完成后即可调用端点curl 或 Postman 均可curl --request POST https://XXXXXXXXXX.execute-api.your-region-here.amazonaws.com/dev/lottery --header Content-Type: application/json --data-raw {first_name: Daniel, last_name: Ani, coupon_value: 99}预期响应{ message: Message sent }在 DynamoDB 中验证记录请求成功只代表消息已进入 SQS 队列落库由 worker 异步完成。稍等片刻后打开 DynamoDB 控制台即可看到 worker 创建的记录包含idUUID、first_name、last_name、coupon_value与created_at五个字段注意DynamoDB 表名是「服务名 stage」的组合。以devstage 为例表名为serverless-ruby-sqs-dynamodb-dev监控与告警运维查看 CloudWatch 日志排查生产与消费两条链路时可以从 Lambda 控制台进入对应函数查看 CloudWatch 日志createLotteryCoupon与lotteryQueueWorker均可也可以直接在 CloudWatch Logs 中按日志组查看两条链路分别会输出Created SQS message生产者侧与Created dynamoDB item with id…worker 侧等日志配合 CloudWatch 可以清晰追踪消息从进队到落库的完整过程。配置 SQS 死信队列邮件告警当消息处理失败被送入 DLQ 时可以配置邮件告警。修改 serverless.yml 中 queue 构造的alarm值为你的邮箱即可constructs: lotteryQueue: type: queue alarm: your-emailexample.com # 替换默认的 mailexample.com部署生效后AWS 会向该邮箱发送订阅确认邮件务必点击确认订阅AWS Notification 通知告警才会真正生效。调整日志保留期默认日志保留期为 30 天如需调整修改 serverless.yml 中的logRetentionInDays属性即可logRetentionInDays: 30补充说明当前仓库的 serverless.yml 中未显式列出该字段原文档指出该值位于 serverless.yml读者可按上述语法自行添加或调整以匹配自身日志审计与成本策略。扩展与清理从 Lift 插件安全脱离EjectLift 一个很有价值的特性是 eject由于 Lift 底层就是 CloudFormation它不会把项目「锁死」在插件里。当项目规模超出插件能力时可以随时执行 eject 把构造展开为原生 CloudFormation 资源之后不再依赖 serverless-lift 也能继续部署演进。移除服务不再需要时一条命令即可清理全部资源sls remove该命令会删除 CloudFormation 栈及其管理的 Lambda、DynamoDB 表、SQS 队列、DLQ、日志组与告警等全部资源。小结通过这个示例可以看到一条完整的异步事件驱动链路API Gateway 同步接收并快速应答SQS 解耦生产与消费节奏worker 在后台可靠落库DLQ 与 CloudWatch 兜底失败场景。源码层面handler.rb 与 worker.rb 保持薄入口业务逻辑下沉到 services / adapters / validators 等职责单一的文件schema 校验、错误序列化、基础设施适配彼此解耦是 Ruby Serverless 项目中值得直接参考的分层写法。赞分享示例工程【免费下载链接】examplesServerless Examples – A collection of boilerplates and examples of serverless architectures built with the Serverless Framework on AWS Lambda, Microsoft Azure, Google Cloud Functions, and more.项目地址https://gitcode.com/gh_mirrors/exa/examples点击查看免费下载相关推荐aws-ruby-cron-with-dynamodb 实战用 Serverless Framework 在 AWS 上构建 Ruby 定时任务并写入 DynamoDBaws ruby cron with dynamodb 实战用 Serverless Framework 在 AWS 上构建 Ruby 定时任务并写入 Dyn示例工程用 Serverless Framework 在 AWS Lambda 上构建 GraphQL DynamoDB APINode.js 实战用 Serverless Framework 在 AWS Lambda 上构建 GraphQL DynamoDB APINode.js 实战 本文以仓库示例工程Chalice Todo 示例应用实战用 AWS Lambda API Gateway DynamoDB 构建带 JWT 认证的 Serverless REST APIChalice Todo 示例应用实战用 AWS Lambda API Gateway DynamoDB 构建带 JWT 认证的 Serverless后端云原生上一篇Interview Script: [Research Topic]下一篇Ant Design Blazor TreeSelect 勾选回填策略ShowCheckedStrategy完整实战指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考