如何将Azure Cosmos DB Change Feed作为事件源接入Azure Event Grid

这是当前成本最低、开发量最小的实现方案，利用Azure Functions原生支持的Cosmos DB Change Feed触发器读取变更，再将变更推送到Event Grid即可。

具体配置步骤 #

• 提前准备好所需资源：Azure Cosmos DB账号（对应业务容器已创建、Change Feed正常开启）、Azure Event Grid自定义主题、Azure Functions实例（运行时选择V3及以上LTS版本，支持所有主流开发语言）
• 配置Cosmos DB Change Feed触发器
  在Function中添加Cosmos DB触发器，按提示填写配置：Cosmos DB连接字符串（建议存在Function应用配置中，不要硬编码）、待监听的数据库名、待监听的容器名、租约容器名（勾选自动创建即可，租约容器用来记录Change Feed的读取偏移量，避免重复消费）
• 编写Event Grid推送逻辑
  先在Function应用配置中存入Event Grid自定义主题的端点地址和访问密钥，之后在触发器的处理代码中，将Change Feed返回的变更数据封装为Event Grid标准事件格式，调用Event Grid发布接口推送即可。
  参考代码（C#）：

using Azure.Messaging.EventGrid;
using Microsoft.Azure.WebJobs;
using Microsoft.Extensions.Logging;
using System;
using System.Collections.Generic;

public static void Run([CosmosDBTrigger(
    databaseName = "你的业务数据库名",
    collectionName = "你的业务容器名",
    ConnectionStringSetting = "CosmosDB_Connection_String",
    LeaseCollectionName = "leases",
    CreateLeaseCollectionIfNotExists = true)]IReadOnlyList<dynamic> changeItems, ILogger log)
{
    if (changeItems == null || changeItems.Count == 0)
        return;

    // 初始化Event Grid客户端
    var egClient = new EventGridClient(
        new Uri(Environment.GetEnvironmentVariable("EventGrid_Endpoint")),
        new Azure.AzureKeyCredential(Environment.GetEnvironmentVariable("EventGrid_Access_Key"))
    );

    // 构造Event Grid事件列表
    List<EventGridEvent> eventList = new();
    foreach (var item in changeItems)
    {
        eventList.Add(new EventGridEvent(
            subject: $"cosmos/change/{item.id}",
            eventType: "CosmosDB.ItemChanged",
            dataVersion: "1.0",
            data: item
        ));
    }

    // 批量推送事件
    egClient.SendEventsAsync(eventList).Wait();
    log.LogInformation($"已成功推送{eventList.Count}条变更事件到Event Grid");
}

• 测试验证
  给Function的托管身份授予Event Grid的EventGrid Data Sender角色权限（也可以直接用之前配置的访问密钥），之后在Cosmos DB中新增/修改一条数据，验证Event Grid的订阅端是否能正常收到变更事件即可。

如果你不想使用无服务器方案，也可以基于Cosmos DB的Change Feed Processor SDK编写常驻服务，部署在AKS、App Service或自有服务器上，拉取Change Feed变更后推送到Event Grid。该方案适合有强定制需求的场景，但需要自行维护服务可用性、进度持久化、异常重试等逻辑，运维成本更高。

优化建议 #

• 批量处理：变更量较大时可以设置攒批阈值，攒够一定数量的变更再推送到Event Grid，减少调用次数、降低成本
• 异常兜底：配置Function重试策略，推送失败的事件可存入死信队列，避免数据丢失
• 事件过滤：不需要推送全量变更时，可以在Function中先做字段或条件过滤，仅推送符合要求的变更到Event Grid，降低下游处理压力
• 低代码替代：变更频率极低且不想写代码的场景，可以用Azure Logic Apps替代Function，直接配置Change Feed和Event Grid的连接器即可完成流程搭建

内容的提问来源于stack exchange，提问作者lostman