为队列消费者转换订单事件

介绍

生产者发送完整的订单事件，但消费者只需要订单 ID 和数量。你将把两个事件转换为精简的队列消息，同时保留路由过滤器。

请先完成 使用 EventBridge 路由订单事件。在本实验的独立 VM 中构建管道；不会复用之前的总线、规则或队列。

准备独立的路由资源

本步骤中，为履约任务创建自定义总线和空队列。

使用 Terminal 旁的 AWS View，将 CLI 查询与本实验的实际资源和结果对比。保留提供的参考数据。

生产者事件包含路由元数据、客户信息和嵌套订单。队列消费者只需要订单 ID 和数量。输入转换从事件中选择字段并构建目标正文，减少消费者对生产者事件封装的依赖。

这个全新工作环境提供配置好的 CLI 访问和无关的参考数据。它不会复用 EV01 资源。从项目目录开始。shell 赋值保存返回的标识符；--query 选择响应字段，--output text 让该字段可以在下一条命令中复用。

cd /home/labex/project
BUS_NAME=labex-ev02-bus
RULE_NAME=labex-ev02-orders
aws events create-event-bus --name "$BUS_NAME"
QUEUE_URL=$(aws sqs create-queue \
  --queue-name labex-ev02-jobs \
  --query QueueUrl \
  --output text)
QUEUE_ARN=$(aws sqs get-queue-attributes \
  --queue-url "$QUEUE_URL" \
  --attribute-names QueueArn \
  --query Attributes.QueueArn \
  --output text)

总线 ARN 标识事件目的地；队列 URL 用于消息操作，而队列 ARN 标识规则目标。确认初始状态：

aws events list-rules --event-bus-name "$BUS_NAME"
aws sqs get-queue-attributes \
  --queue-url "$QUEUE_URL" \
  --attribute-names ApproximateNumberOfMessages ApproximateNumberOfMessagesNotVisible

没有规则或消息。点击 Terminal 旁的 AWS View，检查相同的自定义总线和空队列。运行准备检查。

连接带输入转换器的目标

本步骤中，匹配下单事件、授权规则，并定义精简的消费者载荷。

事件详情转换为消费者正文

选择嵌套订单字段，构建消费者的 id/quantity 正文；保留路由过滤器。

规则模式匹配生产者和事件类别。带引号的 here-document 写入字面 JSON，不进行 shell 展开；file:// 将该文件读入 CLI 请求。

cat > order-pattern.json <<'JSON'
{"source":["labex.orders"],"detail-type":["OrderPlaced"]}
JSON
RULE_ARN=$(aws events put-rule \
  --name "$RULE_NAME" \
  --event-bus-name "$BUS_NAME" \
  --event-pattern file://order-pattern.json \
  --state ENABLED \
  --query RuleArn \
  --output text)

队列需要精确的来源规则授权。使用你的队列和规则 ARN 写入普通策略 JSON。随后，--rawfile 将其编码为字符串值形式的 SQS Policy 属性。

cat > queue-policy.json <<EOF
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Service": "events.amazonaws.com"
      },
      "Action": "sqs:SendMessage",
      "Resource": "$QUEUE_ARN",
      "Condition": {
        "ArnEquals": {
          "aws:SourceArn": "$RULE_ARN"
        }
      }
    }
  ]
}
EOF
jq -n --rawfile policy queue-policy.json '{Policy:$policy}' > queue-attributes.json
aws sqs set-queue-attributes \
  --queue-url "$QUEUE_URL" \
  --attributes file://queue-attributes.json

InputPathsMap 为 JSONPath 选出的字段命名。$ 表示事件根节点；$.detail.order.id 选择嵌套的订单 ID。InputTemplate 使用尖括号占位符构建消息。ID 占位符位于 JSON 字符串引号中；数量是 JSON 数字，没有引号。示例使用简单 ID 和正整数数量。

cat > targets.json <<EOF
[
  {
    "Id": "order-queue",
    "Arn": "$QUEUE_ARN",
    "InputTransformer": {
      "InputPathsMap": {
        "id": "\$.detail.order.id",
        "quantity": "\$.detail.order.quantity"
      },
      "InputTemplate": "{\"id\":\"<id>\",\"quantity\":<quantity>}"
    }
  }
]
EOF
aws events put-targets \
  --rule "$RULE_NAME" \
  --event-bus-name "$BUS_NAME" \
  --targets file://targets.json
aws events list-targets-by-rule --rule "$RULE_NAME" --event-bus-name "$BUS_NAME"

FailedEntryCount 为零。列出的目标包含你的队列 ARN，以及两个映射路径和模板。配置成功仍需要实际投递测试。AWS View 显示已启用的规则和转换器；队列仍为空。运行连接检查。

对比两个实际消费者载荷

本步骤中，发布两个不同的下单事件，并检查生成的队列消息。

将事件详情写为普通 JSON 对象。PutEvents 要求每个 Detail 是 JSON 编码字符串；下面简短的 jq 命令只转换这个字段。每个事件还包含履约消费者不需要的模拟客户信息。第三个取消事件用于测试路由过滤器仍然生效。

cat > event-inputs.json <<EOF
[
  {
    "EventBusName": "$BUS_NAME",
    "Source": "labex.orders",
    "DetailType": "OrderPlaced",
    "Detail": {
      "order": {
        "id": "transform-order-a",
        "quantity": 2
      },
      "customer": {
        "email": "synthetic-a@example.test"
      }
    }
  },
  {
    "EventBusName": "$BUS_NAME",
    "Source": "labex.orders",
    "DetailType": "OrderPlaced",
    "Detail": {
      "order": {
        "id": "transform-order-b",
        "quantity": 4
      },
      "customer": {
        "email": "synthetic-b@example.test"
      }
    }
  },
  {
    "EventBusName": "$BUS_NAME",
    "Source": "labex.orders",
    "DetailType": "OrderCancelled",
    "Detail": {
      "order": {
        "id": "cancelled-order",
        "quantity": 9
      }
    }
  }
]
EOF
jq 'map(.Detail |= tojson)' event-inputs.json > events.json
aws events put-events --entries file://events.json
aws sqs get-queue-attributes \
  --queue-url "$QUEUE_URL" \
  --attribute-names ApproximateNumberOfMessages ApproximateNumberOfMessagesNotVisible

三个事件都被接受，但只有两个下单事件匹配。队列有两条可用消息。以零可见性超时领取后，消息在检查结束时立即可用。查询只显示消息 ID 和正文，保持接收句柄私有。

fromjson 让每个 JSON 正文易于阅读，同时保留其 SQS 消息 ID。它只改变显示的输出。

aws sqs receive-message \
  --queue-url "$QUEUE_URL" \
  --max-number-of-messages 10 \
  --visibility-timeout 0 \
  --output json | jq '[.Messages[] | {MessageId, Body: (.Body | fromjson)}]'

正文为 {"id":"transform-order-a","quantity":2} 和 {"id":"transform-order-b","quantity":4}；不考核它们的顺序。它们具有不同的消息 ID，以及从对应事件派生的值。两者都不包含客户邮箱、路由元数据或嵌套的 order 对象。取消的订单不存在。这证明转换与投递成功，不代表履约已完成或消息已确认。

AWS View 在实际精简队列正文旁显示目标映射。

下面的示例展示两个嵌套字段映射，以及两条实际的精简消费者消息。

AWS View 显示两条转换后的订单消息

运行载荷检查。

移除临时管道

本步骤中，移除你的路由资源，并保留无关状态。

先移除目标，再删除其规则，然后删除自定义总线和队列。队列只包含用于检查的模拟消息；删除会丢弃它们，并不表示已完成业务处理。

aws events remove-targets \
  --rule "$RULE_NAME" \
  --event-bus-name "$BUS_NAME" \
  --ids order-queue
aws events delete-rule --name "$RULE_NAME" --event-bus-name "$BUS_NAME"
aws events delete-event-bus --name "$BUS_NAME"
aws sqs delete-queue --queue-url "$QUEUE_URL"

成功的原生资源清单查询证明删除结果：

aws events list-event-buses
aws events list-rules --event-bus-name default
aws sqs list-queues
aws dynamodb scan --table-name labex-ev02-reference --query Items

只保留默认总线，没有规则；队列 URL 已不存在。参考项目仍包含 keep unchanged。网络或身份验证错误不能证明删除成功。AWS View 显示空的自定义资源和保留的参考资源。

移除本实验创建的普通文件：

rm -f event-inputs.json order-pattern.json queue-policy.json queue-attributes.json targets.json events.json

结束 VM 之前，运行清理检查。

总结

你将嵌套订单字段映射为精简的 SQS 消费者数据格式，验证了两个实际事件中的不同值，并保留事件过滤与来源规则权限。你移除了临时管道，同时保留参考数据。

下一个实验将在订单工作流中连接实际业务任务。

为队列消费者转换订单事件

介绍

相关认证考点

准备独立的路由资源

连接带输入转换器的目标

对比两个实际消费者载荷

移除临时管道

总结