如何基于Azure Function Cosmos DB触发器实现写入SQL的至少一次投递保障

嘿，这个问题我刚好在项目里实战过，咱们从Cosmos DB触发器的特性到SQL写入的细节，一步步拆解怎么实现可靠的至少一次投递～

一、先利用Cosmos DB触发器的内置可靠性机制 #

Cosmos DB变更源触发器本身就基于租约容器实现了至少一次投递的基础能力：

• 租约容器会记录每个分区的处理进度（偏移量），只有当Function成功处理完一批变更并正常退出后，触发器才会更新租约的进度；
• 如果Function执行失败（比如SQL写入超时、抛出异常），租约不会更新，触发器会在配置的重试间隔后，重新拉取这批未确认的变更进行处理。

你可以在host.json里配置更贴合业务的重试策略，比如：

{
  "extensions": {
    "cosmosDb": {
      "retryPolicy": {
        "maxRetryAttempts": 5, // 最大重试次数
        "maxRetryWaitTimeInSeconds": 30 // 最大重试等待间隔
      }
    }
  }
}

另外要注意租约容器的RU配置要足够，避免因为租约更新慢导致重试延迟。

二、必须保证SQL写入操作的幂等性 #

至少一次投递意味着可能会出现重复执行的情况，比如Function重试时，同一条Cosmos DB变更会被再次处理。这时候必须让SQL的写入操作是幂等的，避免产生重复数据：

• 用唯一键约束：把Cosmos DB文档的id或者变更源自带的_lsn（日志序列号，全局唯一且递增）作为SQL表的主键或唯一键。这样重复写入时会触发主键冲突，直接忽略即可；
• 使用MERGE语句：代替单纯的INSERT，实现“存在则更新，不存在则插入”的逻辑，示例SQL：

MERGE INTO YourTargetTable t
USING (VALUES (@cosmosId, @flattenedField1, @flattenedField2)) s (CosmosId, Field1, Field2)
ON t.CosmosId = s.CosmosId
WHEN MATCHED THEN UPDATE SET t.Field1 = s.Field1, t.Field2 = s.Field2
WHEN NOT MATCHED THEN INSERT (CosmosId, Field1, Field2) VALUES (s.CosmosId, s.Field1, s.Field2);

• 记录已处理的变更标识：在SQL表里新增ProcessedLsn字段，存储Cosmos DB变更的_lsn。每次处理前先查询该Lsn是否已存在，只有不存在时才执行写入操作。

三、处理批量/部分失败的场景 #

如果你的触发器配置了批量拉取（比如一次拉取多条变更），要注意避免部分成功的情况：

• 优先单条处理：如果业务允许，尽量把批量拉取的条数设为1，这样每条变更的处理都是独立的，一条失败不会影响其他；
• 用事务包裹批量写入：如果必须批量处理，要将所有SQL写入操作放在同一个数据库事务中。比如在C#里使用SqlTransaction，一旦某条写入失败，就回滚整个事务，让触发器重试整个批次，保证要么全部成功，要么全部重试。

四、补充死信队列处理不可恢复的错误 #

如果遇到不可恢复的错误（比如数据格式完全不符合要求、SQL表结构变更导致写入失败），不能让触发器无限重试占用资源，这时候可以引入死信队列：

• 在Function中捕获异常，判断错误类型是否为不可恢复（比如数据校验失败）；
• 当重试次数耗尽后，将该条变更的原始数据、错误信息一起发送到Azure Service Bus队列（死信专用）；
• 发送成功后，正常结束Function，让触发器更新租约，避免后续重试；
• 后续可以通过监控死信队列，人工介入处理这些异常数据。

关于你提到的Event Hub方案的适配性 #

你之前看的Event Hub可靠性方案里，幂等性处理、事务包裹、死信队列这些思路是完全通用的。区别在于，Event Hub用checkpoint记录进度，而Cosmos DB触发器用租约容器，核心都是“只有处理成功才更新进度，失败则重试”，所以那些方案的核心逻辑可以直接借鉴。

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