c# - 如何正确限制从 WebJobs 对 DocumentDb 的访问

Question

我有一个带有 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 倍。

什么样的节流/测量/同步机制可以在这里很好地发挥作用？

Answer 1

从.NET SDK 1.8.0开始，我们会在合理的范围内自动处理请求率太大异常(默认重试9次，并在从服务器返回后进行下次重试)。

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

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