时间戳

首页 > 解决方案 > 用于限制每秒请求的信号量不起作用

c# - 用于限制每秒请求的信号量不起作用

问题描述

我正在使用 Google Analytics，并且该服务有 10 个并发请求的限制。我不得不以某种方式限制我的 API，所以我决定使用信号量，但它似乎不起作用。所有请求同时触发。我在我的代码中找不到问题。

public async Task<SiteAnalyticsDTO> Handle(GetSiteAnalyticsParameter query)
{
    var todayVisits = _googleAnalyticsService.GetTodayVisitsNumber();
    var todayTraffic = _googleAnalyticsService.GetTodayTraffic();
    var newAndReturningUsers = _googleAnalyticsService.GetNewAndReturningUsersNumber();
    var averageSessionDuration = _googleAnalyticsService.GetAverageSessionDuration();
    var deviceCategory = _googleAnalyticsService.GetSessionNumberByDeviceCategory();
    var topPages = _googleAnalyticsService.GetTodaysTopPages();
    var guestsAndRegisteredUsers = _googleAnalyticsService.GetGuestsVsRegisteredUsers();
    var averageNumberOfSessionsPerDay = _googleAnalyticsService.GetAverageSessionsNumber();
    var visitsPerWeekday = _googleAnalyticsService.GetTrafficByWeekday();
    var visitsByHours = _googleAnalyticsService.GetTrafficByTimeOfDay();
    var usersByPrefectures = _googleAnalyticsService.GetUsersByPrefectures();
    var usersByCountry = _googleAnalyticsService.GetUsersByCountry();

    var tasks = new List<Task>()
        {
            todayVisits, todayTraffic, newAndReturningUsers,
            averageSessionDuration, deviceCategory, topPages,
            guestsAndRegisteredUsers, averageNumberOfSessionsPerDay, visitsPerWeekday,
            visitsByHours, usersByPrefectures, usersByCountry
        };

    var throttler = new SemaphoreSlim(MaxRequests, MaxRequests);

    foreach(var task in tasks)
    {
        await throttler.WaitAsync();

        try
        {
            await task;
            await Task.Delay(1000); // It's important due to limits of Google Analytics requests (10 queries per second per IP address)
        }
        finally
        {
            throttler.Release();
        }
    }

    await Task.WhenAll(tasks);

    return new SiteAnalyticsDTO()
        {
            TodayVisits = await todayVisits,
            TodayTraffic = await todayTraffic,
            NewAndReturningUsers = await newAndReturningUsers,
            AverageSessionDuration = await averageSessionDuration,
            DeviceCategory = await deviceCategory,
            TopPages = await topPages,
            GuestsAndRegisteredUsers = await guestsAndRegisteredUsers,
            AverageNumberOfSessionsPerDay = await averageNumberOfSessionsPerDay,
            VisitsPerWeekday = await visitsPerWeekday,
            VisitsByHours = await visitsByHours,
            UsersByPrefectures = await usersByPrefectures,
            UsersByCountry = await usersByCountry
        };
}

这是谷歌分析调用的一些示例方法：

  public async Task<int> GetTodayVisitsNumber(List<long> listingIds = null)
    {
        string filter = GetFilter(listingIds);

        var getReportsRequest = GetReportsRequestModel(GetTodayDateRange(), "ga:sessionCount", "ga:sessions", _configuration.MainViewId, filter);
        var response = await _service.Reports.BatchGet(getReportsRequest).ExecuteAsync();
        Console.WriteLine(response);
        var data = response.Reports.FirstOrDefault();

        return Convert.ToInt32(data?.Data.Totals[0].Values[0]);
    }

标签： c#.net-coreasync-awaitgoogle-analyticssemaphore

解决方案

所有请求同时触发。

让我们看看这里

var todayVisits = _googleAnalyticsService.GetTodayVisitsNumber();
var todayTraffic = _googleAnalyticsService.GetTodayTraffic();
var newAndReturningUsers = _googleAnalyticsService.GetNewAndReturningUsersNumber();
var averageSessionDuration = _googleAnalyticsService.GetAverageSessionDuration();
var deviceCategory = _googleAnalyticsService.GetSessionNumberByDeviceCategory();
var topPages = _googleAnalyticsService.GetTodaysTopPages();
var guestsAndRegisteredUsers = _googleAnalyticsService.GetGuestsVsRegisteredUsers();
var averageNumberOfSessionsPerDay = _googleAnalyticsService.GetAverageSessionsNumber();
var visitsPerWeekday = _googleAnalyticsService.GetTrafficByWeekday();
var visitsByHours = _googleAnalyticsService.GetTrafficByTimeOfDay();
var usersByPrefectures = _googleAnalyticsService.GetUsersByPrefectures();
var usersByCountry = _googleAnalyticsService.GetUsersByCountry();

您正在存储每种方法的结果。当您使用诸如“ methodName();”之类的括号标记时，您调用该方法并将结果存储在var.

然后，您将这些方法的结果存储在一个列表中，然后await每个方法都使用 aSemaphore来限制一次可以等待的任务数量。

问题是：每个都await立即完成，因为您在上面最初调用它们时已经（同步）等待它们。

这使您相信SemaphoreSlim不工作，因为如果每个Task在等待时立即返回（因为它们已经被调用），那么它们之间就没有时间了。

存储async方法以供以后使用，而不是一次调用它们。
您不能将委托存储在var您必须将它们存储在显式类型变量Func<TResult>中。

例如：

Func<Task<object>> todayVisits = _googleAnalyticsService.GetTodayVisitsNumber;

编者注，我不知道这些方法返回什么我替换对象尽可能通用

现在 - 如果我们将每一个都存储在一个变量中会非常麻烦，所以我们不要将它们存储在单独的变量中，而是直接将它们放在这样的列表中：

var awaitableTasks = new List<Func<Task<object>>>()
{
    _googleAnalyticsService.GetTodayVisitsNumber,
    _googleAnalyticsService.GetTodayTraffic,
    _googleAnalyticsService.GetNewAndReturningUsersNumber,
    _googleAnalyticsService.GetAverageSessionDuration,
    _googleAnalyticsService.GetSessionNumberByDeviceCategory,
    _googleAnalyticsService.GetTodaysTopPages,
    _googleAnalyticsService.GetGuestsVsRegisteredUsers,
    _googleAnalyticsService.GetAverageSessionsNumber,
    _googleAnalyticsService.GetTrafficByWeekday,
    _googleAnalyticsService.GetTrafficByTimeOfDay,
    _googleAnalyticsService.GetUsersByPrefectures,
    _googleAnalyticsService.GetUsersByCountry
};

因为这些新对象本身不是任务，而是返回 a 的方法，所以Task我们必须更改存储和调用它们的方式，为此我们将使用本地方法，因此我将回顾我所做的每一个更改。

让我们创建它Semaphore并创建一个我们可以放置任务以跟踪它们的地方。

让我们也创建一个可以存储每个任务结果的地方await。

var throttler = new SemaphoreSlim(MaxRequests, MaxRequests);

var tasks = new List<Task>();

ConcurrentDictionary<string, object> results = new();

让我们创建一个具有几个职责的本地方法

  1. 接受 aFunc<Task<object>>作为参数
  2. Await方法
  3. 将该方法的结果放在我们以后可以得到的地方
  4. Semphore即使遇到错误也要释放
async Task Worker(Func<Task<object>> awaitableFunc)
{
    try
    {
        resultDict.TryAdd(awaitableFunc.GetMethodInfo().Name, await awaitableFunc());
    }
    finally
    {
        throttler.Release();
    }
}

编者注：您可以使用 lambda 表达式完成相同的操作，但为了清晰和格式化，我更喜欢使用本地方法。

启动工人并存储他们返回的任务。

这样..如果在创建最后一对时它们还没有完成，那么我们可以在创建最终对象之前等待它们完成（因为我们将需要它们提供的所有结果来创建最终对象）。

foreach (var task in awaitableTasks)
{
    await throttler.WaitAsync();
    tasks.Add(Task.Run(() => Worker(task)));
}

// wait for the tasks to finish
await Task.WhenAll(tasks);

创建最终对象然后返回它。

return new SiteAnalyticsDTO()
{
    TodayVisits = resultDict[nameof(_googleAnalyticsService.GetTodayVisitsNumber)],
    TodayTraffic = resultDict[nameof(_googleAnalyticsService.GetTodayTraffic)],
    NewAndReturningUsers = resultDict[nameof(_googleAnalyticsService.GetNewAndReturningUsersNumber)],
    AverageSessionDuration = resultDict[nameof(_googleAnalyticsService.GetAverageSessionDuration)],
    DeviceCategory = resultDict[nameof(_googleAnalyticsService.GetSessionNumberByDeviceCategory)],
    TopPages = resultDict[nameof(_googleAnalyticsService.GetTodaysTopPages)],
    GuestsAndRegisteredUsers = resultDict[nameof(_googleAnalyticsService.GetGuestsVsRegisteredUsers)],
    AverageNumberOfSessionsPerDay = resultDict[nameof(_googleAnalyticsService.GetAverageSessionsNumber)],
    VisitsPerWeekday = resultDict[nameof(_googleAnalyticsService.GetTrafficByWeekday)],
    VisitsByHours = resultDict[nameof(_googleAnalyticsService.GetTrafficByTimeOfDay)],
    UsersByPrefectures = resultDict[nameof(_googleAnalyticsService.GetUsersByPrefectures)],
    UsersByCountry = resultDict[nameof(_googleAnalyticsService.GetUsersByCountry)]
};

将它们组合在一起，我认为我们有一些可能有用的东西，或者至少很容易修改以满足您的需求。

public static async Task<SiteAnalyticsDTO> Handle(GetSiteAnalyticsParameter query)
{
    // store these methods so we can iterate and execute them later
    var awaitableTasks = new List<Func<Task<object>>>()
    {
        _googleAnalyticsService.GetTodayVisitsNumber,
        _googleAnalyticsService.GetTodayTraffic,
        _googleAnalyticsService.GetNewAndReturningUsersNumber,
        _googleAnalyticsService.GetAverageSessionDuration,
        _googleAnalyticsService.GetSessionNumberByDeviceCategory,
        _googleAnalyticsService.GetTodaysTopPages,
        _googleAnalyticsService.GetGuestsVsRegisteredUsers,
        _googleAnalyticsService.GetAverageSessionsNumber,
        _googleAnalyticsService.GetTrafficByWeekday,
        _googleAnalyticsService.GetTrafficByTimeOfDay,
        _googleAnalyticsService.GetUsersByPrefectures,
        _googleAnalyticsService.GetUsersByCountry
    };

    // create a way to limit the number of concurrent requests
    var throttler = new SemaphoreSlim(MaxRequests, MaxRequests);

    // create a place to store the tasks we create
    var finalTasks = new List<Task>();

    // make sure we have some where to put our results
    ConcurrentDictionary<string, object> resultDict = new();

    // make a worker that accepts one of those methods, invokes it
    // then adds the result to the dict
    async Task Worker(Func<Task<object>> awaitableFunc)
    {
        try
        {
            resultDict.TryAdd(awaitableFunc.GetMethodInfo().Name, await awaitableFunc());
        }
        finally
        {
            // make sure even if we encounter an error we still release the semphore
            throttler.Release();
        }
    }

    // iterate over the tasks, wait for the sempahore
    // when we get a slot, create a worker and send it to the background
    foreach (var task in awaitableTasks)
    {
        await throttler.WaitAsync();
        finalTasks.Add(Task.Run(() => Worker(task)));
    }

    // wait for any remaining tasks to finish up in the background if they are still running
    await Task.WhenAll(finalTasks);

    // create the return object from the results of the dictionary
    return new SiteAnalyticsDTO()
    {
        TodayVisits = resultDict[nameof(_googleAnalyticsService.GetTodayVisitsNumber)],
        TodayTraffic = resultDict[nameof(_googleAnalyticsService.GetTodayTraffic)],
        NewAndReturningUsers = resultDict[nameof(_googleAnalyticsService.GetNewAndReturningUsersNumber)],
        AverageSessionDuration = resultDict[nameof(_googleAnalyticsService.GetAverageSessionDuration)],
        DeviceCategory = resultDict[nameof(_googleAnalyticsService.GetSessionNumberByDeviceCategory)],
        TopPages = resultDict[nameof(_googleAnalyticsService.GetTodaysTopPages)],
        GuestsAndRegisteredUsers = resultDict[nameof(_googleAnalyticsService.GetGuestsVsRegisteredUsers)],
        AverageNumberOfSessionsPerDay = resultDict[nameof(_googleAnalyticsService.GetAverageSessionsNumber)],
        VisitsPerWeekday = resultDict[nameof(_googleAnalyticsService.GetTrafficByWeekday)],
        VisitsByHours = resultDict[nameof(_googleAnalyticsService.GetTrafficByTimeOfDay)],
        UsersByPrefectures = resultDict[nameof(_googleAnalyticsService.GetUsersByPrefectures)],
        UsersByCountry = resultDict[nameof(_googleAnalyticsService.GetUsersByCountry)]
    };
}

推荐阅读