3. 事件网格发布与消费实战

上一节我们理解了 Event Grid 的“事件源—主题—订阅—处理程序”模型。本节把模型落到可运行的最小闭环：在 Azure 门户或 CLI 中创建一个自定义主题（Custom Topic），用一段代码向它发布符合 CloudEvents 1.0 规范的事件，再通过 Event Grid Viewer 这个现成的 Web 处理程序订阅并肉眼确认事件确实到达。完成之后，你会得到一条可反复使用的“发布—订阅—验证”链路，后续章节的筛选、重试、死信等能力都可以挂在这条链路上做实验。

一、先明确三个对象与一个关键参数

自定义主题是 Event Grid 中由用户自己创建、自己发布事件的资源，它和 Blob、资源组这类系统主题不同，事件来源完全由你控制。创建时有两个参数必须记住：主题终结点（topicEndpoint）形如 https://<topic-name>.<region>-1.eventgrid.azure.net/api/events，是发布事件的 HTTP 目标；访问密钥（access key）用于生成发布时的鉴权头。这两个值在门户“自定义主题 → 访问密钥”页可以拿到，也是后面代码里唯一需要替换的敏感信息。

订阅端我们使用 Event Grid Viewer。它是一个开源的单页应用，部署到 Azure 应用服务后会暴露一个 /api/updates 之类的 Webhook 地址，并在页面上实时打印收到的事件 JSON。它的价值在于：不需要自己写消费端代码，就能把“事件是否到达、到达了几条、负载长什么样”直接显示出来，非常适合教学和排障。

二、创建自定义主题并部署 Viewer

用 Azure CLI 创建主题最省事，也便于复现。以下命令中的名称、区域、资源组请按自己的环境替换，示例参数仅作演示：

az group create --name rg-eg-lab --location eastasia

az eventgrid topic create \
  --name topic-orders-demo \
  --resource-group rg-eg-lab \
  --location eastasia \
  --input-schema cloudeventschemav1_0

az eventgrid topic show \
  --name topic-orders-demo \
  --resource-group rg-eg-lab \
  --query "endpoint" -o tsv

az eventgrid topic key list \
  --name topic-orders-demo \
  --resource-group rg-eg-lab \
  --query "key1" -o tsv

注意 --input-schema cloudeventschemav1_0 这一项。Event Grid 默认使用它自己的 EventGridSchema，如果这里不指定 CloudEvents，发布时就必须按 Event Grid 专有格式构造 JSON。本节的目标是 CloudEvents，所以创建主题时就把输入架构定下来，避免后面发布格式与主题期望不一致而返回 400。

Viewer 的部署可以走门户的“部署自定义模板”，也可以直接用仓库提供的 ARM 模板。部署完成后打开站点，页面顶部会显示它自己的 Webhook 地址，把它复制下来，下一步订阅要用。

三、创建事件订阅，把主题接到 Viewer

回到自定义主题，选择“事件订阅 → 新建”。终结点类型选 Web Hook，把 Viewer 的地址粘贴进去。Event Grid 在保存订阅时会向该地址发送一次验证请求，Viewer 会自动应答，因此正常情况下几秒内订阅状态就变为“已创建”。如果一直卡在验证失败，通常是地址写错、站点未启动或返回了非 200 状态码。

订阅创建成功后，主题与 Viewer 之间就建立了一条单向推送通道：主题收到事件后，按订阅配置把事件 POST 给 Viewer。这条通道的方向、参与者和关键参数，可以用下面这张图来对照理解。

四、构造并发布一条 CloudEvents 事件

CloudEvents 1.0 的必填属性有四个：id、source、specversion、type。发布到 Event Grid 自定义主题时，请求体是一个事件数组，Content-Type 用 application/cloudevents+json，鉴权用主题密钥放在 aeg-sas-key 头里。下面是一段可直接运行的 Python 示例，参数均为演示值：

import requests, uuid, json

endpoint = "https://topic-orders-demo.eastasia-1.eventgrid.azure.net/api/events"
key = "<your-topic-key>"

event = [{
    "id": str(uuid.uuid4()),
    "source": "/orders/service",
    "specversion": "1.0",
    "type": "com.contoso.order.created",
    "subject": "orders/1001",
    "time": "2025-01-15T08:30:00Z",
    "datacontenttype": "application/json",
    "data": {"orderId": "1001", "amount": 268.00, "currency": "CNY"}
}]

r = requests.post(endpoint,
                  headers={"aeg-sas-key": key,
                           "Content-Type": "application/cloudevents+json"},
                  data=json.dumps(event))
print(r.status_code, r.text)

几个容易踩坑的点：id 建议用 UUID，重复的 id 在部分场景会被去重；source 是标识事件产生者的 URI 风格字符串，不是主题地址；type 用反向域名风格，方便后续按类型筛选；subject 可选，但它是做前缀/后缀筛选时最常用的字段。发布成功时返回 202 Accepted，响应体为空。如果返回 400，优先检查 JSON 是否为数组、必填字段是否齐全、Content-Type 是否写对；返回 401 则说明密钥或请求头名称有误。

五、在 Viewer 中确认事件到达

发布成功后切到 Viewer 页面，正常情况下 1～3 秒内会出现一张新的事件卡片，展开后能看到完整的 CloudEvents 结构：id 是刚才生成的 UUID，type 为 com.contoso.order.created，data 里是订单金额与币种。这一步的判读逻辑很直接：页面出现事件 = 主题接收成功且订阅推送成功；页面没有事件但发布返回 202，则问题多半出在订阅端（Webhook 地址、验证状态、处理程序是否返回 200）。

为了确认链路可重复，可以连续发布三条不同 orderId 的事件，观察 Viewer 是否按顺序出现三张卡片。如果只出现部分，可以到主题的“指标”页看 Published Events 与 Delivered Events 两个计数：前者统计进入主题的事件数，后者统计成功投递到处理程序的事件数。两者不一致，说明投递环节有失败，可结合“事件订阅 → 指标”里的 Delivery Failed Events 进一步定位。

六、本节可直接复用的验证清单

到这里，一条最小可用的事件发布与消费链路已经跑通。你手里有了一个可反复发布的自定义主题、一个能实时显示事件的 Viewer，以及一套判断“事件到底走到哪一步”的观察点。后续无论是加筛选、改重试策略，还是换成函数、队列等真实处理程序，都可以在这条链路上直接替换和对比。