如何在Polly ExecuteAsync()中集成Azure Cosmos DB FeedIterator？泛型Cosmos数据查询方法的Polly重试实现求助

嘿，我来帮你搞定Polly和Azure Cosmos DB FeedIterator的集成问题！咱们一步步拆解，解决你遇到的几个核心痛点：FeedIterator无法复用、重试触发条件、以及避免重复读取数据的问题。

核心思路先理清 #

首先得明确两个关键事实：

1. FeedIterator是状态性的：一旦调用过ReadNextAsync，它内部的状态就会更新，没法回退到之前的位置，所以重试时必须重新创建迭代器。
2. 重试触发条件：我们不需要直接判断状态码，而是通过捕获CosmosException来触发重试——因为response.EnsureSuccessStatusCode()会在请求失败时自动抛出这个异常，我们只需要筛选出值得重试的错误类型（比如限流429、服务器5xx错误）。

方案1：基础版——全查询重试（简单易实现） #

这个版本会在整个查询过程中遇到可重试异常时，从头开始重新执行查询。优点是代码改动小，适合数据量不大的场景。

public async Task<IEnumerable<T>> GetItemsAsync(QueryDefinition query)
{
    // 定义Polly重试策略：针对429限流和5xx服务器错误，用指数退避重试3次
    var retryPolicy = Policy
        .Handle<CosmosException>(ex => 
            ex.StatusCode == HttpStatusCode.TooManyRequests || // 429限流
            (int)ex.StatusCode >= 500) // 所有5xx服务器错误
        .WaitAndRetryAsync(
            retryCount: 3,
            sleepDurationProvider: retryAttempt => TimeSpan.FromSeconds(Math.Pow(2, retryAttempt)), // 指数退避：2s→4s→8s
            onRetryAsync: (exception, timespan, retryCount, context) =>
            {
                // 可选：添加重试日志，方便排查问题
                Console.WriteLine($"第{retryCount}次重试，原因：{exception.Message}，等待{timespan.TotalSeconds}秒");
                return Task.CompletedTask;
            });

    // 把整个查询逻辑包装到Polly的ExecuteAsync中，异常时自动重试
    return await retryPolicy.ExecuteAsync(async () =>
    {
        var results = new List<T>();
        // 每次重试都创建全新的FeedIterator
        using (FeedIterator resultsIterator = _container.GetItemQueryStreamIterator(query))
        {
            while (resultsIterator.HasMoreResults)
            {
                using (ResponseMessage response = await resultsIterator.ReadNextAsync())
                {
                    // 失败时抛出CosmosException，触发Polly重试
                    response.EnsureSuccessStatusCode();
                    
                    dynamic streamResponse = FromStream<dynamic>(response.Content);
                    var rawText = ((JsonElement)streamResponse).GetProperty("Documents").GetRawText();
                    var responseObj = JsonSerializer.Deserialize<List<T>>(rawText, _jsonSerializerOptions);
                    
                    if (responseObj != null) 
                        results.AddRange(responseObj);
                }
            }
        }
        return results;
    });
}

方案2：进阶版——分页重试（避免重复读取数据） #

如果你的数据量较大，从头重试会重复读取已成功的页面，效率很低。我们可以利用Cosmos的continuation token，记录当前读取到的位置，重试时只从失败的页面开始继续：

public async Task<IEnumerable<T>> GetItemsAsync(QueryDefinition query)
{
    var retryPolicy = Policy
        .Handle<CosmosException>(ex => 
            ex.StatusCode == HttpStatusCode.TooManyRequests ||
            (int)ex.StatusCode >= 500)
        .WaitAndRetryAsync(3, retryAttempt => TimeSpan.FromSeconds(Math.Pow(2, retryAttempt)));

    var results = new List<T>();
    string continuationToken = null;

    do
    {
        // 针对单页读取进行重试，失败时只重试过当前页
        await retryPolicy.ExecuteAsync(async () =>
        {
            // 根据当前的continuation token创建新的FeedIterator，从上次中断的位置继续
            using (FeedIterator resultsIterator = _container.GetItemQueryStreamIterator(
                query, 
                continuationToken: continuationToken))
            {
                using (ResponseMessage response = await resultsIterator.ReadNextAsync())
                {
                    response.EnsureSuccessStatusCode();
                    
                    dynamic streamResponse = FromStream<dynamic>(response.Content);
                    var rawText = ((JsonElement)streamResponse).GetProperty("Documents").GetRawText();
                    var responseObj = JsonSerializer.Deserialize<List<T>>(rawText, _jsonSerializerOptions);
                    
                    if (responseObj != null) 
                        results.AddRange(responseObj);
                    
                    // 更新continuation token，用于下一页读取
                    continuationToken = response.Headers.ContinuationToken;
                }
            }
        });
    } 
    while (!string.IsNullOrEmpty(continuationToken)); // 没有更多结果时停止

    return results;
}

关键细节说明 #

1. 为什么不能复用FeedIterator：FeedIterator内部维护了查询的状态（比如当前页的位置），调用ReadNextAsync后状态会更新，无法回退，所以每次重试必须创建新的迭代器。
2. 重试条件的选择：429限流是Cosmos最常见的需要重试的场景（Cosmos会返回重试间隔建议，你也可以用ex.RetryAfter来优化退避时间），5xx错误属于服务器临时故障，也适合重试。
3. 指数退避策略：这是Cosmos官方推荐的重试策略，通过指数级增加等待时间，避免给数据库造成持续压力。

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