Azure Function处理Cosmos DB文档失败转Service Bus后多消息重试问题咨询

问题根因分析 #

首先得明确一个核心点：当你在f1中使用ICollector<Document>输出到Service Bus队列时，Azure Functions运行时会把每个Document对象作为独立的一条队列消息发送，而不是把所有Document打包成一个JSON数组。

而你在f2里尝试用IReadOnlyList<Document>作为触发器输入时，就会触发序列化异常——因为Service Bus触发器每次只会拿到单条Document的JSON对象，而非JSON数组，Json.NET自然没法把单个对象反序列化成集合类型，这就是你看到报错的根本原因。

解决方案1：按单条消息处理（推荐，符合最佳实践） #

这是最贴合Azure Function和Service Bus设计逻辑的方式，不需要修改f1的代码，只需要保持f2的输入为单个Document，让运行时自动处理队列里的每条消息：

调整后的f2代码（可正常处理批量场景）

#r "Microsoft.Azure.Documents.Client"
#r "Newtonsoft.Json"
using Microsoft.Azure.Documents;
using System.Collections.Generic;
using System;
using Newtonsoft.Json;

public static void Run(Document myQueueItem, TraceWriter log)
{
    log.Info($"Processing document: {myQueueItem.Id}");
    
    // 示例：从Document中提取字段生成SQL语句
    var docData = JsonConvert.DeserializeObject<Dictionary<string, object>>(myQueueItem.ToString());
    string sql = $"INSERT INTO TargetTable (Id, Name) VALUES ('{docData["Id"]}', '{docData["Name"]}')";
    
    // 这里执行你的SQL Server操作逻辑
    // ...
    
    log.Info($"Successfully processed document {myQueueItem.Id}");
}

为什么这能处理批量？因为f1用ICollector发送的每条Document都是独立的队列消息，Service Bus触发器会自动为每条消息触发一次f2，你可以在Function的配置里调整并发度，来控制批量处理的速度。

解决方案2：手动打包批量消息（仅特殊场景使用） #

如果你确实需要把多个Document打包成单条队列消息发送（比如减少消息数量），那需要手动处理序列化和反序列化：

修改后的f1代码（打包成JSON数组）

#r "Microsoft.Azure.Documents.Client"
#r "Microsoft.ServiceBus"
#r "Newtonsoft.Json"
using Microsoft.ServiceBus.Messaging;
using Microsoft.Azure.Documents;
using System.Collections.Generic;
using System;
using Newtonsoft.Json;

public static void Run(IReadOnlyList<Document> inputFeed, TraceWriter log, out string outputSbQueue)
{
    // 把所有Document打包成List，序列化成JSON字符串
    var docList = new List<Document>(inputFeed);
    outputSbQueue = JsonConvert.SerializeObject(docList);
    
    log.Info($"Sent batch of {docList.Count} documents as single queue message");
}

修改后的f2代码（解析JSON数组）

#r "Microsoft.Azure.Documents.Client"
#r "Newtonsoft.Json"
using Microsoft.Azure.Documents;
using System.Collections.Generic;
using System;
using Newtonsoft.Json;

public static void Run(string myQueueItem, TraceWriter log)
{
    try
    {
        // 把JSON字符串反序列化成Document列表
        var docList = JsonConvert.DeserializeObject<List<Document>>(myQueueItem);
        
        foreach(var doc in docList)
        {
            log.Info($"Processing document: {doc.Id}");
            
            // 生成SQL语句并执行操作
            var docData = JsonConvert.DeserializeObject<Dictionary<string, object>>(doc.ToString());
            string sql = $"INSERT INTO TargetTable (Id, Name) VALUES ('{docData["Id"]}', '{docData["Name"]}')";
            
            // 执行SQL逻辑
            // ...
        }
        
        log.Info($"Processed batch of {docList.Count} documents");
    }
    catch(Exception ex)
    {
        log.Error($"Failed to process batch message: {ex.Message}", ex);
        // 可以抛出异常让Service Bus重试，或者手动处理死信
        throw;
    }
}

⚠️ 注意：这种方式要注意Service Bus的单条消息大小限制（最大256KB），如果批量的Document总大小超过这个限制，消息会发送失败。

额外提示 #

• Service Bus触发器自带重试机制，如果f2处理失败，会自动重试（可在Function配置中调整重试策略），不需要自己写遍历队列的逻辑。
• 如果你需要保证消息处理的顺序，可以把Service Bus队列设置为会话模式，或者调整Function的并发度为1。

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