c#之如何正确地限制从 WebJobs 访问 DocumentDb

我有一个带有 blob 和队列触发器的 Azure WebKob，用于将数据保存到 Azure DocumentDb。

有时我会收到一个错误:

Microsoft.Azure.Documents.RequestRateTooLargeException: Message: {"Errors":["Request rate is large"]}

目前我使用此代码限制请求。 WebJob 函数:

public async Task ParseCategoriesFromCsv(...) 
{ 
    double find = 2.23, add = 5.9, replace = 10.67; 
    double requestCharge = Math.Round(find + Math.Max(add, replace)); 
 
    await categoryProvider.SaveCategories(requestCharge , categories); 
} 

用于操作文档数据库客户端的类别提供程序:

public async Task<ResourceResponse<Document>[]> SaveCategories(double requestCharge, Category[] categories) 
{ 
    var requestDelay = TimeSpan.FromSeconds(60.0 / (collectionOptions.RequestUnits / requestCharge)); 
 
    var scheduler = new IntervalTaskScheduler(requestDelay, Scheduler.Default); // Rx 
 
    var client = new DocumentClient(endpoint, authorizationKey, 
        new ConnectionPolicy 
        { 
            ConnectionMode = documentDbOptions.ConnectionMode, 
            ConnectionProtocol = documentDbOptions.ConnectionProtocol 
        }); 
 
    return await Task.WhenAll(documents.Select(async d => 
       await scheduler.ScheduleTask( 
           () => client.PutDocumentToDb(collectionOptions.CollectionLink, d.SearchIndex, d)))); 
} 

节流/测量/同步请求的任务调度器:

private readonly Subject<Action> _requests = new Subject<Action>(); 
private readonly IDisposable _observable; 
 
public IntervalTaskScheduler(TimeSpan requestDelay, IScheduler scheduler) 
{ 
    _observable = _requests.Select(i => Observable.Empty<Action>() 
                                                  .Delay(requestDelay) 
                                                  .StartWith(i)) 
                           .Concat() 
                           .ObserveOn(scheduler) 
                           .Subscribe(action => action()); 
} 
 
public Task<T> ScheduleTask<T>(Func<Task<T>> request) 
{ 
    var tcs = new TaskCompletionSource<T>(); 
    _requests.OnNext(async () => 
    { 
        try 
        { 
            T result = await request(); 
            tcs.SetResult(result); 
        } 
        catch (Exception ex) 
        { 
            tcs.SetException(ex); 
        } 
    }); 
    return tcs.Task; 
} 

所以它基本上是来自 ResourceResponse<Document>.RequestCharge 的一些常量但是:

• 当我触发 1 个队列时，它工作正常，但当触发 8 个队列时，它会抛出错误。
• 如果将请求费用增加 8 倍，则 8 个队列可以正常工作，但只有 1 个队列的工作速度比它慢 8 倍。

什么节流/测量/同步机制可以在这里很好地工作？

请您参考如下方法:

从 .NET SDK 1.8.0 开始，我们会在合理的范围内自动处理 Request Rate too large 异常(默认重试 9 次，并在从服务器返回后接受重试以进行下一次重试)。

如果您需要更好的控制，您可以在传递给 DocumentClient 对象的 ConnectionPolicy 实例上配置 RetryOptions，我们将用它覆盖默认的重试策略。

因此，您不再需要像上面那样在应用程序代码中添加任何自定义逻辑来处理 429 异常。